app公司网站建设百度收录文章

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

一、架构图

[Step Function] → [Lambda函数] → [Secrets Manager] → [企业Okta认证] → [内部API]↑              ↓重试策略       [VPC配置/NAT]

二、配置步骤

1. 创建专用域账号

   • 联系企业IT部门在Okta创建服务账号（如：[e-mail]）
   • 授予最小必要API访问权限

2. 配置AWS Secrets Manager

aws secretsmanager create-secret --name InternalApiCreds \
--secret-string '{"username":"[e-mail]","password":"P@ssw0rd!2023"}'

3. 创建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": "*"}]
}

4. 配置VPC网络

   • 创建私有子网
   • 配置NAT网关（出口流量）
   • 安全组设置（仅允许HTTPS出口）

5. 部署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访问失败，请检查日志"}

6. 配置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访问流程失败"}}
}

三、安全增强措施

1. KMS加密

aws kms create-key --description "API访问加密密钥"
aws secretsmanager update-secret --secret-id InternalApiCreds \
--kms-key-id alias/aws/secretsmanager

2. 网络隔离

# VPC配置示例
Resources:ApiAccessFunction:Type: AWS::Lambda::FunctionProperties:VpcConfig:SecurityGroupIds:- sg-0123456789abcdef0SubnetIds:- subnet-0123456789abcdef0- subnet-0123456789abcdef1

3. 凭证自动轮换

# 轮换Lambda示例（需要企业API支持）
def rotate_secret(event, context):# 1. 生成新密码# 2. 更新企业AD密码# 3. 更新Secrets Manager# 4. 验证新凭证pass

四、监控配置

1. CloudWatch警报指标：

   • Lambda错误率 > 1%
   • API调用延迟 > 5秒
   • Step Function失败次数

2. 日志筛选模式：

ERROR ApiAccessError
WARN RetryAttempt

五、部署流程

1. 基础设施部署

sam deploy \--template-file template.yaml \--stack-name api-access-system \--capabilities CAPABILITY_IAM

2. 环境变量配置

Environment:Variables:SECRET_NAME: InternalApiCredsOKTA_AUTH_URL: https://company.okta.com/oauth2/v1/tokenINTERNAL_API_URL: https://internal-api.company.com/v1/data

六、测试方案

1. 单元测试

# 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'

2. 集成测试

# 触发Step Function执行
aws stepfunctions start-execution \--state-machine-arn arn:aws:states:us-west-2:123456789012:stateMachine:ApiDataWorkflow

3. 混沌测试

# 模拟网络故障
aws lambda update-function-configuration \--function-name ApiAccessFunction \--vpc-config SubnetIds=subnet-99999999

以上方案实现了：

1. 安全的凭证管理（Secrets Manager + KMS）
2. 可靠的重试机制（Step Function重试策略）
3. 网络隔离（VPC配置）
4. 完善的监控告警
5. 自动化的部署流程

建议配合企业安全策略定期进行：

• 渗透测试
• 凭证轮换审计
• IAM策略审查
• 灾难恢复演练