实现AWS Lambda函数安全地请求企业内部API返回数据

需要编写一个Lambda函数在AWS云上运行，它需要访问企业内部的API获取JSON格式的数据，企业有网关和防火墙，API有公司的okta身份认证，通过公司的域账号来授权访问，现在需要创建一个专用的域账号，让Lambda函数访问Secret Manager，来获取账号密码，然后通过配置访问公司内部API的数据，请写出所有的开发配置步骤，以及完成这一功能的所有的Python源代码，需要确保安全性和可靠性。

---

一、架构设计

1. Lambda函数部署在AWS公有云
2. 通过AWS Secrets Manager存储Okta认证凭证
3. Lambda通过HTTPS访问企业API网关
4. 使用Okta的OAuth 2.0客户端凭证模式获取访问令牌
5. 所有通信使用TLS加密

---

二、配置步骤

1. 创建专用域账号

• 在企业Okta控制台：
  • 创建新应用（OAuth 2.0客户端凭证类型）
  • 生成Client ID和Client Secret
  • 配置API访问权限范围（Scopes）

2. 配置AWS Secrets Manager

aws secretsmanager create-secret --name prod/okta_api_creds \
--secret-string '{"client_id": "YOUR_CLIENT_ID","client_secret": "YOUR_CLIENT_SECRET","token_url": "https://yourcompany.okta.com/oauth2/v1/token","api_endpoint": "https://internal-api.yourcompany.com/v1/data"
}'

3. 配置Lambda执行角色

IAM策略示例：

{"Version": "2012-10-17","Statement": [{"Effect": "Allow","Action": "secretsmanager:GetSecretValue","Resource": "arn:aws:secretsmanager:region:account-id:secret:prod/okta_api_creds-*"},{"Effect": "Allow","Action": "logs:CreateLogGroup","Resource": "arn:aws:logs:region:account-id:*"},{"Effect": "Allow","Action": ["logs:CreateLogStream","logs:PutLogEvents"],"Resource": ["arn:aws:logs:region:account-id:log-group:/aws/lambda/your-function-name:*"]}]
}

4. 网络配置（可选）

如果企业API在私有网络：

• 部署Lambda到VPC私有子网
• 配置安全组出站规则
• 设置VPC对等连接或VPN到企业网络，具体来说，可以使用VPC对等连接连接到已有的可以连接到公司VPN网络，也可以在VPC网络里创建EC2实例安装可以连接到公司VPN的第三方连接软件实现连接，Lambda调用EC2上的Python代码实现，还可以使用AWS Site-to-Site VPN组件直接连接VPN实现，在Lambda函数中实现业务逻辑。

---

三、Python源代码

1. Lambda主处理程序

import os
import json
import boto3
from botocore.exceptions import ClientError
import requests
from requests.exceptions import RequestException# 初始化客户端
secrets_client = boto3.client('secretsmanager')
session = requests.Session()def get_secret():"""安全获取Secret Manager中的凭证"""try:response = secrets_client.get_secret_value(SecretId='prod/okta_api_creds')return json.loads(response['SecretString'])except ClientError as e:if e.response['Error']['Code'] == 'ResourceNotFoundException':raise Exception("Secret not found") from eelif e.response['Error']['Code'] == 'AccessDeniedException':raise Exception("Access denied to secret") from eelse:raise Exception(f"Secret retrieval failed: {str(e)}") from edef get_okta_token(client_id, client_secret, token_url):"""获取Okta访问令牌"""headers = {'Accept': 'application/json'}data = {'grant_type': 'client_credentials','client_id': client_id,'client_secret': client_secret,'scope': 'api.access'}try:response = session.post(token_url,headers=headers,data=data,timeout=10)response.raise_for_status()return response.json()['access_token']except RequestException as e:raise Exception(f"Token request failed: {str(e)}") from edef call_internal_api(api_url, token):"""调用内部API"""headers = {'Authorization': f'Bearer {token}','Content-Type': 'application/json'}try:response = session.get(api_url,headers=headers,timeout=15)response.raise_for_status()return response.json()except RequestException as e:raise Exception(f"API call failed: {str(e)}") from edef lambda_handler(event, context):try:# 获取认证凭证secret = get_secret()# 获取访问令牌token = get_okta_token(secret['client_id'],secret['client_secret'],secret['token_url'])# 调用内部APIapi_data = call_internal_api(secret['api_endpoint'], token)return {'statusCode': 200,'body': api_data}except Exception as e:# 记录完整错误到CloudWatchprint(f"Error: {str(e)}")return {'statusCode': 500,'body': json.dumps({'error': 'Internal Server Error'})}

2. 安全增强措施

1. Secret加密：

   aws secretsmanager update-secret --secret-id prod/okta_api_creds \
   --kms-key-id alias/aws/secretsmanager
   

2. 网络防护：

   # 在requests会话中增加安全配置
   session = requests.Session()
   adapter = requests.adapters.HTTPAdapter(max_retries=3,pool_connections=10,pool_maxsize=10
   )
   session.mount('https://', adapter)
   

3. 证书验证增强：

   # 在敏感请求中启用证书固定
   response = session.get(api_url, headers=headers,verify='/path/to/cert.pem')
   

---

四、部署与验证

1. 部署流程

# 安装依赖
pip install requests boto3 -t ./dependencies# 打包Lambda
zip -r lambda.zip *.py dependencies/# 创建Lambda函数
aws lambda create-function \--function-name InternalApiAccess \--runtime python3.9 \--handler lambda_function.lambda_handler \--role arn:aws:iam::account-id:role/lambda-execution-role \--zip-file fileb://lambda.zip \--environment Variables={AWS_NODEJS_CONNECTION_REUSE_ENABLED=1} \--timeout 30

2. 测试验证

1. 在AWS控制台执行Lambda测试事件
2. 验证CloudWatch日志：

   aws logs tail /aws/lambda/InternalApiAccess --follow
   

3. 执行端到端测试：

   # 测试脚本示例
   import lambda_function
   print(lambda_function.lambda_handler({}, None))
   

---

五、运维监控

1. 安全审计配置

# 启用Secret轮换
aws secretsmanager rotate-secret --secret-id prod/okta_api_creds \
--rotation-lambda-arn arn:aws:lambda:region:account-id:function:secret-rotation-function \
--rotation-rules '{"AutomaticallyAfterDays": 90}'

2. CloudWatch警报配置

• 监控指标：
  • Lambda错误率 > 1%
  • 调用延迟 > 5秒
  • 并发执行数异常增长

3. 安全防护

1. 启用Lambda代码签名
2. 配置VPC流量镜像
3. 启用AWS Shield Advanced

---

六、故障处理预案

故障场景	处理方案
Okta认证失败	1. 检查Secret Manager内容 2. 验证Okta应用配置 3. 检查网络连通性
API连接超时	1. 检查安全组规则 2. 验证企业防火墙白名单 3. 测试DNS解析
凭证泄露	1. 立即轮换Secret 2. 审计Lambda执行日志 3. 撤销旧客户端凭证

---

该方案实现了：

1. 凭证与代码分离（通过Secrets Manager）
2. 最小权限原则（IAM策略）
3. 端到端加密传输
4. 完善的错误处理机制
5. 完整的审计跟踪能力

建议定期执行以下维护：

• 每季度轮换客户端凭证
• 每月审计Lambda执行日志
• 每周验证备份恢复流程
• 持续监控API响应时间指标