实现AWS Step Function安全地请求企业内部API返回数据
需要编写一个Step Function在AWS云上运行，它需要访问企业内部的API获取JSON格式的数据，企业有网关和防火墙，API有公司的okta身份认证，通过公司的域账号来授权访问，现在需要创建一个专用的域账号，让Step Function访问Secret Manager，来获取账号密码，然后通过配置访问公司内部API的数据，请写出所有的开发配置步骤，以及完成这一功能的所有的Python源代码，需要确保安全性和可靠性。
一、架构图
[Step Function] → [Lambda函数] → [Secrets Manager] → [企业Okta认证] → [内部API]↑ ↓重试策略 [VPC配置/NAT]
二、配置步骤
-
创建专用域账号
- 联系企业IT部门在Okta创建服务账号（如：api-access@company.com）
- 授予最小必要API访问权限
-
配置AWS Secrets Manager
aws secretsmanager create-secret --name InternalApiCreds \
--secret-string '{"username":"api-access@company.com","password":"P@ssw0rd!2023"}'
- 创建IAM角色
// Lambda执行策略
{"Version": "2012-10-17","Statement": [{"Effect": "Allow","Action": "secretsmanager:GetSecretValue","Resource": "arn:aws:secretsmanager:region:account-id:secret:InternalApiCreds-xxxxx"},{"Effect": "Allow","Action": ["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"],"Resource": "*"}]
}
-
配置VPC网络
- 创建私有子网
- 配置NAT网关（出口流量）
- 安全组设置（仅允许HTTPS出口）
-
部署Lambda函数
import os
import json
import boto3
import requests
from botocore.exceptions import ClientError
from requests.exceptions import RequestExceptionclass ApiAccessError(Exception):"""自定义异常基类"""passclass SecretRetrievalError(ApiAccessError):"""密钥获取异常"""passclass OktaAuthError(ApiAccessError):"""认证失败异常"""passclass ApiConnectionError(ApiAccessError):"""API连接异常"""passdef get_secret():"""从Secrets Manager获取凭证"""secret_name = os.environ['SECRET_NAME']region = os.environ['AWS_REGION']client = boto3.client(service_name='secretsmanager',region_name=region)try:response = client.get_secret_value(SecretId=secret_name)secret = json.loads(response['SecretString'])return secret['username'], secret['password']except (ClientError, KeyError, json.JSONDecodeError) as e:raise SecretRetrievalError(f"密钥获取失败: {str(e)}")def get_okta_token(username: str, password: str) -> str:"""获取Okta访问令牌"""auth_url = os.environ['OKTA_AUTH_URL']payload = {'grant_type': 'password','username': username,'password': password,'scope': 'api_access'}headers = {'Accept': 'application/json','Content-Type': 'application/x-www-form-urlencoded'}try:response = requests.post(auth_url,data=payload,headers=headers,timeout=10)response.raise_for_status()return response.json()['access_token']except (RequestException, KeyError) as e:raise OktaAuthError(f"认证失败: {str(e)}")def fetch_api_data(token: str) -> dict:"""调用内部API"""api_url = os.environ['INTERNAL_API_URL']headers = {'Authorization': f'Bearer {token}','Content-Type': 'application/json'}try:response = requests.get(api_url,headers=headers,timeout=15)response.raise_for_status()return response.json()except RequestException as e:raise ApiConnectionError(f"API调用失败: {str(e)}")def lambda_handler(event, context):try:# 获取凭证username, password = get_secret()# 获取访问令牌access_token = get_okta_token(username, password)# 调用APIapi_response = fetch_api_data(access_token)return {'status': 'SUCCESS','data': api_response}except ApiAccessError as e:# 记录详细错误日志但不暴露敏感信息print(f"操作失败: {str(e)}")return {'status': 'ERROR','error_type': type(e).__name__,'error_message': "API访问失败，请检查日志"}
- 配置Step Function
{"Comment": "企业API数据获取流程","StartAt": "RetrieveApiData","States": {"RetrieveApiData": {"Type": "Task","Resource": "arn:aws:lambda:us-west-2:123456789012:function:ApiAccessFunction","Retry": [{"ErrorEquals": ["Lambda.ServiceException", "Lambda.SdkClientException"],"IntervalSeconds": 2,"MaxAttempts": 3,"BackoffRate": 2},{"ErrorEquals": ["ApiConnectionError"],"IntervalSeconds": 5,"MaxAttempts": 2,"BackoffRate": 1.5}],"Catch": [{"ErrorEquals": ["States.ALL"],"Next": "ErrorHandler"}],"Next": "ProcessData"},"ProcessData": {"Type": "Pass","End": true},"ErrorHandler": {"Type": "Fail","Error": "ApiAccessFailed","Cause": "API访问流程失败"}}
}
三、安全增强措施
- KMS加密
aws kms create-key --description "API访问加密密钥"
aws secretsmanager update-secret --secret-id InternalApiCreds \
--kms-key-id alias/aws/secretsmanager
- 网络隔离
# VPC配置示例
Resources:ApiAccessFunction:Type: AWS::Lambda::FunctionProperties:VpcConfig:SecurityGroupIds:- sg-0123456789abcdef0SubnetIds:- subnet-0123456789abcdef0- subnet-0123456789abcdef1
- 凭证自动轮换
# 轮换Lambda示例（需要企业API支持）
def rotate_secret(event, context):# 1. 生成新密码# 2. 更新企业AD密码# 3. 更新Secrets Manager# 4. 验证新凭证pass
四、监控配置
-
CloudWatch警报指标：
- Lambda错误率 > 1%
- API调用延迟 > 5秒
- Step Function失败次数
-
日志筛选模式：
ERROR ApiAccessError
WARN RetryAttempt
五、部署流程
- 基础设施部署
sam deploy \--template-file template.yaml \--stack-name api-access-system \--capabilities CAPABILITY_IAM
- 环境变量配置
Environment:Variables:SECRET_NAME: InternalApiCredsOKTA_AUTH_URL: https://company.okta.com/oauth2/v1/tokenINTERNAL_API_URL: https://internal-api.company.com/v1/data
六、测试方案
- 单元测试
# test_api_access.py
import pytest
from unittest.mock import patch@patch('requests.post')
def test_okta_auth_success(mock_post):mock_post.return_value.status_code = 200mock_post.return_value.json.return_value = {'access_token': 'test123'}token = get_okta_token('user', 'pass')assert token == 'test123'
- 集成测试
# 触发Step Function执行
aws stepfunctions start-execution \--state-machine-arn arn:aws:states:us-west-2:123456789012:stateMachine:ApiDataWorkflow
- 混沌测试
# 模拟网络故障
aws lambda update-function-configuration \--function-name ApiAccessFunction \--vpc-config SubnetIds=subnet-99999999
以上方案实现了：
- 安全的凭证管理（Secrets Manager + KMS）
- 可靠的重试机制（Step Function重试策略）
- 网络隔离（VPC配置）
- 完善的监控告警
- 自动化的部署流程
建议配合企业安全策略定期进行：
- 渗透测试
- 凭证轮换审计
- IAM策略审查
- 灾难恢复演练