如何仅以Okta为OIDC提供商生成JWT并获取AWS STS临时凭证

前置概念纠正 #

1. 不是「登录Okta门户生成JWT」，而是用户登录你集成了Okta OIDC的业务应用/自定义登录页时，Okta会返回符合OIDC标准的ID Token/Access Token（JWT格式）。Okta门户是管理员操作后台，不是终端用户的认证入口。
2. AWS STS的AssumeRoleWithWebIdentity原生支持Okta这类标准OIDC提供商的JWT，完全不需要Cognito中转，你的核心思路是对的。

步骤1：Okta OIDC应用配置 #

1. 登录Okta管理控制台，创建一个OIDC - Web Application（如果是纯前端应用，选Single-Page Application）。
2. 配置应用的登录回调URL（前端跳转认证的话必填；如果是后端直接调用Okta API拿token，可暂时留空）。
3. 进入Sign On标签页，找到「OpenID Connect ID Token」配置：
   • 勾选需要包含在JWT里的动态属性，比如用户角色（groups）、租户ID（多租户SaaS可自定义custom:tenant_id属性）。
   • 设置固定的Audience（受众），比如https://your-aws-saas-app.com，后续AWS信任策略会用到这个值。
4. 记录两个关键信息：
   • Okta OIDC issuer URL（格式：https://<你的Okta域名>.com/oauth2/default）
   • 应用的Client ID

步骤2：AWS IAM角色配置 #

1. 登录AWS控制台，进入IAM服务，创建新角色：
   • 信任实体类型选Web identity。
   • 点击「Add provider」，类型选OpenID Connect，输入Okta的issuer URL，点击获取指纹，再填入刚才设置的Audience值，完成身份提供商创建。
   • 回到角色创建页面，选择刚创建的Okta OIDC提供商，匹配对应的受众（aud）。
   • 给角色添加基础权限策略（权限要略宽于实际需求，后续用会话策略缩小范围），比如允许访问S3、DynamoDB等核心服务。
2. 编辑角色的信任策略，确保内容如下（替换占位符）：

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Principal": {
           "Federated": "arn:aws:iam::<你的AWS账号ID>:oidc-provider/<你的Okta域名>.com/oauth2/default"
         },
         "Action": "sts:AssumeRoleWithWebIdentity",
         "Condition": {
           "StringEquals": {
             "<你的Okta域名>.com/oauth2/default:aud": "https://your-aws-saas-app.com"
           }
         }
       }
     ]
   }
   

3. 记录这个角色的ARN：arn:aws:iam::<你的AWS账号ID>:role/<你的角色名>

步骤3：Lambda函数实现（Python示例） #

1. 创建Lambda函数，运行时选Python 3.10+，给Lambda的执行角色添加sts:AssumeRoleWithWebIdentity权限。
2. 代码示例（建议把敏感配置放到Lambda环境变量，不要硬编码）：

   import boto3
   import jwt
   import json
   import os
   from botocore.exceptions import ClientError
   from jwt import PyJWKClient
   
   # 从环境变量读取配置
   OKTA_ISSUER = os.environ['OKTA_ISSUER']
   ROLE_ARN = os.environ['ROLE_ARN']
   OKTA_AUDIENCE = os.environ['OKTA_AUDIENCE']
   
   sts_client = boto3.client('sts')
   
   def lambda_handler(event, context):
       # 从POST请求中提取JWT
       try:
           request_body = json.loads(event['body'])
           jwt_token = request_body.get('id_token')
           if not jwt_token:
               return {'statusCode': 400, 'body': json.dumps('Missing JWT token')}
       except:
           return {'statusCode': 400, 'body': json.dumps('Invalid request format')}
   
       # 验证JWT有效性（可选但推荐，减少无效STS调用）
       try:
           jwks_client = PyJWKClient(f"{OKTA_ISSUER}/v1/keys")
           signing_key = jwks_client.get_signing_key_from_jwt(jwt_token)
           payload = jwt.decode(
               jwt_token,
               signing_key.key,
               algorithms=["RS256"],
               audience=OKTA_AUDIENCE,
               issuer=OKTA_ISSUER
           )
       except jwt.InvalidTokenError as e:
           return {'statusCode': 401, 'body': json.dumps(f'Invalid JWT: {str(e)}')}
   
       # 从JWT中提取动态属性
       user_role = payload.get('groups', ['viewer'])[0]
       tenant_id = payload.get('custom:tenant_id', 'default-tenant')
   
       # 生成会话策略，动态缩小权限
       session_policy = {
           "Version": "2012-10-17",
           "Statement": [
               {
                   "Effect": "Allow",
                   "Action": [
                       "s3:ListBucket", "s3:GetObject" if user_role == 'admin' else "s3:GetObject"
                   ],
                   "Resource": [
                       f"arn:aws:s3:::{tenant_id}-app-bucket",
                       f"arn:aws:s3:::{tenant_id}-app-bucket/*"
                   ]
               }
           ]
       }
   
       # 调用STS获取临时凭证
       try:
           response = sts_client.assume_role_with_web_identity(
               RoleArn=ROLE_ARN,
               RoleSessionName=f"{tenant_id}-{payload['sub']}",
               WebIdentityToken=jwt_token,
               Policy=json.dumps(session_policy),
               DurationSeconds=3600
           )
           # 返回临时凭证给客户端
           return {
               'statusCode': 200,
               'body': json.dumps({
                   'AccessKeyId': response['Credentials']['AccessKeyId'],
                   'SecretAccessKey': response['Credentials']['SecretAccessKey'],
                   'SessionToken': response['Credentials']['SessionToken'],
                   'Expiration': response['Credentials']['Expiration'].isoformat()
               })
           }
       except ClientError as e:
           return {'statusCode': 500, 'body': json.dumps(f'STS Error: {str(e)}')}
   

3. 注意：需要给Lambda添加pyjwt和python-jose依赖，可通过Lambda层部署。

步骤4：API Gateway配置 #

1. 创建REST API，添加POST方法，集成到上述Lambda函数。
2. 配置请求模板，确保能正确解析application/json格式的请求体。
3. 可选：启用API密钥认证，防止未授权调用。

步骤5：客户端捕获JWT并调用API #

1. 前端用Okta Sign-In Widget实现登录：
   • 引入Widget库，配置Client ID、Okta域名、回调URL。
   • 登录成功后，通过oktaAuth.token.getWithoutPrompt()或登录回调获取ID Token。
   • 以{"id_token": "<获取到的JWT>"}为请求体，调用API Gateway的POST端点。
2. 后端应用可直接用Okta OAuth2客户端库获取token，再调用API。

测试验证 #

1. 用户登录你的应用，完成Okta认证后拿到JWT。
2. 调用API Gateway端点，获取临时凭证。
3. 用AWS CLI测试权限：

   # 配置临时凭证
   aws configure set aws_access_key_id <临时AccessKeyId>
   aws configure set aws_secret_access_key <临时SecretAccessKey>
   aws configure set aws_session_token <临时SessionToken>
   
   # 验证权限（比如查看租户专属桶）
   aws s3 ls s3://<tenant-id>-app-bucket
   

   确认只能执行会话策略允许的操作。

内容的提问来源于stack exchange，提问作者Piyush Pandey