使用预分配URL作为EKS令牌调用API时遭遇401未授权问题

核心问题分析 #

出现401错误通常由以下几个原因导致：

• STS预签名URL过期时间超过上限（STS限制最大为3600秒，你的代码设为40000秒，直接导致签名无效）
• IAM服务账号缺少必要权限
• EKS集群的aws-auth ConfigMap未映射该IAM实体的RBAC权限
• 代码中占位符（如<EKS_CLUSTER_NAME>）未替换为实际值

分步修复方案 #

1. 修正令牌生成的过期时间 #

在get_bearer_token函数中，将expires_in从40000改为3600以内的数值（比如300秒）：

signed_url = signer.generate_presigned_url(
    params,
    region_name=session.region_name,
    expires_in=300,  # 修正为3600秒以内
    operation_name=''
)

2. 确保IAM服务账号具备必要权限 #

给服务账号附加以下IAM策略：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "eks:DescribeCluster",
                "sts:GetCallerIdentity"
            ],
            "Resource": "*"
        }
    ]
}

3. 配置EKS的aws-auth ConfigMap #

将IAM服务账号映射到EKS集群的RBAC权限组，比如赋予查看pod的权限：

apiVersion: v1
kind: ConfigMap
metadata:
  name: aws-auth
  namespace: kube-system
data:
  mapUsers: |
    - userarn: arn:aws:iam::你的AWS账号ID:user/你的服务账号名称
      username: 你的服务账号名称
      groups:
        - system:viewers  # 或自定义的有权限的组，如需更高权限可改用system:masters

执行kubectl apply -f aws-auth.yaml更新配置。

4. 简化并修正代码冗余 #

原get_session_token函数中创建的sts_client未被使用，可简化为：

def get_session_token(access_key, secret_key, region):
    return boto3.Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name=region
    )

修复后的完整代码 #

"Lambda function list pods in EKS cluster"
import base64
import logging
import re
import boto3

from botocore.signers import RequestSigner
from kubernetes import client, config


logger = logging.getLogger()
logger.setLevel(logging.INFO)

cluster_name = "你的EKS集群名称"
ACCESS_KEY = "你的服务账号Access Key"
SECRET_KEY = "你的服务账号Secret Key"
CLUSTER_REGION = "你的集群区域"

eks = boto3.client('eks', 
                   aws_access_key_id=ACCESS_KEY, 
                   aws_secret_access_key=SECRET_KEY, 
                   region_name=CLUSTER_REGION)

cluster_cache = {}

def get_session_token(access_key, secret_key, region):
    return boto3.Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key,
        region_name=region
    )

def get_cluster_info():
    "Retrieve cluster endpoint and certificate"
    cluster_info = eks.describe_cluster(name=cluster_name)
    return {
        "endpoint": cluster_info['cluster']['endpoint'],
        "ca": cluster_info['cluster']['certificateAuthority']['data']
    }

def get_bearer_token():
    "Create authentication token"
    session = get_session_token(ACCESS_KEY, SECRET_KEY, CLUSTER_REGION)
    sts = session.client('sts')
    service_id = sts.meta.service_model.service_id
    
    signer = RequestSigner(
        service_id,
        CLUSTER_REGION,
        'sts',
        'v4',
        session.get_credentials(),
        session.events
    )

    params = {
        'method': 'GET',
        'url': 'https://sts.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15',
        'body': {},
        'headers': {
            'x-k8s-aws-id': cluster_name
        },
        'context': {}
    }

    signed_url = signer.generate_presigned_url(
        params,
        region_name=session.region_name,
        expires_in=300,
        operation_name=''
    )
    base64_url = base64.urlsafe_b64encode(signed_url.encode('utf-8')).decode('utf-8')
    return 'k8s-aws-v1.' + re.sub(r'=*', '', base64_url)

def lambda_handler(_event, _context):
    "Lambda handler"
    if cluster_name in cluster_cache:
        cluster = cluster_cache[cluster_name]
    else:
        cluster = get_cluster_info()
        cluster_cache[cluster_name] = cluster

    kubeconfig = {
        'apiVersion': 'v1',
        'clusters': [{
          'name': 'cluster1',
          'cluster': {
            'certificate-authority-data': cluster["ca"],
            'server': cluster["endpoint"]}
        }],
        'contexts': [{'name': 'context1', 'context': {'cluster': 'cluster1', "user": "user1"}}],
        'current-context': 'context1',
        'kind': 'Config',
        'preferences': {},
        'users': [{'name': 'user1', "user" : {'token': get_bearer_token()}}]
    }

    config.load_kube_config_from_dict(config_dict=kubeconfig)
    v1_api = client.CoreV1Api()
    ret = v1_api.list_namespaced_pod("default")
    return f"There are {len(ret.items)} pods in the default namespace."

print(lambda_handler(None, None))

验证步骤 #

1. 先测试IAM服务账号权限：调用sts.get_caller_identity()和eks.describe_cluster()确认能正常返回数据
2. 生成令牌后，用curl测试集群访问：

curl -H "Authorization: Bearer 生成的令牌" https://你的EKS端点/api/v1/namespaces/default/pods --cacert <(echo 集群CA数据 | base64 -d)

内容的提问来源于stack exchange，提问作者Monu singh