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

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

news 来源：原创 2025/8/13 14:55:24

需要编写一个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策略审查
灾难恢复演练