当前位置: 首页 > news >正文

实现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到企业网络

三、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响应时间指标

相关文章:

  • [每周一更]-(第140期):sync.Pool 使用详解:性能优化的利器
  • Python制作简易PDF查看工具PDFViewerV1.0查找功能优化
  • 【文件操作与IO】详细解析文件操作与IO (二)
  • 零、HarmonyOS应用开发者基础学习总览
  • Cursor新版0.49.x发布
  • 开源Midjourney替代方案:企业级AI绘画+PPT生成系统+AI源码
  • YOLO拓展-锚框(anchor box)详解
  • 深入理解C++ 中的vector容器
  • 机器学习核心算法全解析:从基础到进阶的 18 大算法模型
  • 点云数据处理开源C++方案
  • 神经网络的数学之旅:从输入到反向传播
  • 在串口通信中使用共享指针(`std::shared_ptr`)
  • 用 R 语言打造交互式叙事地图:讲述黄河源区生态变化的故事
  • MCP认证难题破解:常见技术难题实战分析与解决方案
  • 额外篇 非递归之美:归并排序与快速排序的创新实现
  • 基于Redis的3种分布式ID生成策略
  • JAVA文件I/O
  • 大数据平台简介
  • 《Operating System Concepts》阅读笔记:p738-p747
  • Java从入门到“放弃”(精通)之旅——数组的定义与使用⑥
  • 江西望仙谷回应“游客凌晨等不到接驳车”:已限流,接驳车运行时间延长
  • 消费持续升温,这个“五一”假期有何新亮点?
  • 美国多地爆发集会抗议特朗普政府多项政策
  • 《探秘海昏侯国》数字沉浸特展亮相首届江西文化旅游产业博览交易会
  • 强制性国家标准《危险化学品企业安全生产标准化通用规范》发布
  • 辽宁辽阳市白塔区一饭店火灾事故举行新闻发布会,现场为遇难者默哀