如何轻量化的在亚马逊云科技中国区安全使用 Transfer Family SFTP

程序（公网部署）
 │  持有：X.509 客户端证书 + 私钥（唯一长期凭据）
 │
 │ ① X.509 证书 → IAM Roles Anywhere
 ▼
IAM Roles Anywhere（Trust Anchor = 自签 CA）
 │  返回：短时 AWS 临时凭据（AccessKey/Secret/SessionToken，1小时过期）
 │
 │ ② 临时凭据 → Secrets Manager
 ▼
Secrets Manager（KMS 加密，每 4 小时自动轮换）
 │  返回：当前有效的 SFTP 密码
 │
 │ ③ 用户名 + 密码 → SFTP
 ▼
Transfer Family SFTP Server
 │  VPC Internet-Facing 端点 + Elastic IP（固定公网 IP 用于白名单）
 │  安全组：仅允许客户端出口 IP 的 22 端口
 │  身份提供者：Custom Identity Provider（Lambda）
 │      - 校验密码（同时接受 AWSCURRENT 和 AWSPREVIOUS 版本）
 │      - 校验客户端源 IP
 │      - 返回 IAM 角色 + S3 逻辑目录映射
 ▼
Amazon S3（后端存储，KMS 加密）

aws s3api create-bucket \
  --bucket sftp-poc- \
  --create-bucket-configuration LocationConstraint=cn-north-1 \
  --profile china --region cn-north-1

aws s3api put-public-access-block \
  --bucket sftp-poc- \
  --public-access-block-configuration \
  BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true \
  --profile china --region cn-north-1

aws kms create-key \
  --description "SFTP POC - Secrets Manager encryption" \
  --profile china --region cn-north-1

aws kms create-alias \
  --alias-name alias/sftp-secret-key \
  --target-key-id  \
  --profile china --region cn-north-1

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {"Service": "transfer.amazonaws.com"},
    "Action": "sts:AssumeRole"
  }]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket", "s3:GetBucketLocation"],
      "Resource": "arn:aws-cn:s3:::sftp-poc-"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:GetObject", "s3:DeleteObject"],
      "Resource": "arn:aws-cn:s3:::sftp-poc-/*"
    }
  ]
}

aws secretsmanager create-secret \
  --name sftp/prod/app1 \
  --kms-key-id alias/sftp-secret-key \
  --secret-string '{"Password":""}' \
  --profile china --region cn-north-1

import json
import os
import hmac
import boto3

sm = boto3.client("secretsmanager")

SECRET_ID = os.environ["SECRET_ID"]
ROLE_ARN = os.environ["ROLE_ARN"]
HOME_BUCKET = os.environ["HOME_BUCKET"]
ALLOWED_IPS = os.environ.get("ALLOWED_IPS", "").split(",")

DENY = {}

def _get_pw(stage):
    try:
        r = sm.get_secret_value(SecretId=SECRET_ID, VersionStage=stage)
        return json.loads(r["SecretString"]).get("Password", "")
    except Exception:
        return ""

def handler(event, context):
    # 日志（不记录密码）
    print(json.dumps({
        "username": event.get("username"),
        "protocol": event.get("protocol"),
        "sourceIp": event.get("sourceIp")
    }))

    if event.get("protocol") != "SFTP":
        return DENY

    # 源 IP 白名单校验
    src = event.get("sourceIp", "")
    if ALLOWED_IPS and ALLOWED_IPS[0] and src not in ALLOWED_IPS:
        return DENY

    # 强制密码登录
    presented = event.get("password", "")
    if not presented:
        return DENY

    # 同时接受当前和上一版本密码（覆盖轮换过渡期）
    for stage in ("AWSCURRENT", "AWSPREVIOUS"):
        valid = _get_pw(stage)
        if valid and hmac.compare_digest(presented, valid):
            return {
                "Role": ROLE_ARN,
                "HomeDirectoryType": "LOGICAL",
                "HomeDirectoryDetails": json.dumps([
                    {"Entry": "/", "Target": f"/{HOME_BUCKET}/users/{event['username']}"}
                ]),
            }

    return DENY

# 1. 创建安全组（仅允许客户端 IP 访问 22 端口）
aws ec2 create-security-group \
  --group-name sftp-sg \
  --description "SFTP - allow client IP only" \
  --vpc-id  \
  --profile china --region cn-north-1

aws ec2 authorize-security-group-ingress \
  --group-id  \
  --ip-permissions '[{"IpProtocol":"tcp","FromPort":22,"ToPort":22,
    "IpRanges":[{"CidrIp":"/32"}]}]' \
  --profile china --region cn-north-1

# 2. 分配 EIP（每个子网一个）
aws ec2 allocate-address --domain vpc --profile china --region cn-north-1

# 3. 创建 SFTP Server（先不带 EIP）
aws transfer create-server \
  --endpoint-type VPC \
  --endpoint-details '{"SubnetIds":["",""],
    "VpcId":"","SecurityGroupIds":[""]}' \
  --identity-provider-type AWS_LAMBDA \
  --identity-provider-details '{"Function":"",
    "SftpAuthenticationMethods":"PASSWORD"}' \
  --protocols '["SFTP"]' \
  --profile china --region cn-north-1

# 4. 停止 → 挂载 EIP → 启动（EIP 只能在 OFFLINE 状态挂载）
aws transfer stop-server --server-id  ...
aws transfer update-server --server-id  \
  --endpoint-details '{"AddressAllocationIds":["",""],
    "SubnetIds":["",""]}' ...
aws transfer start-server --server-id  ...

import json
import secrets
import string
import boto3

sm = boto3.client("secretsmanager")
SECRET_ID = "sftp/prod/app1"

def handler(event, context):
    alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
    new_pw = "".join(secrets.choice(alphabet) for _ in range(32))
    sm.put_secret_value(
        SecretId=SECRET_ID,
        SecretString=json.dumps({"Password": new_pw})
    )
    print(f"Rotated password for {SECRET_ID}")
    return {"statusCode": 200}

aws events put-rule \
  --name sftp-rotate-every-4h \
  --schedule-expression "rate(4 hours)" \
  --profile china --region cn-north-1

# CA（必须包含完整 v3 扩展，Roles Anywhere 验证严格）
cat > ca.cnf  client_ext.cnf 注册 Trust Anchor 和 Profile：

__CODE_START__
# 注册 CA 为 Trust Anchor
aws rolesanywhere create-trust-anchor \
  --name sftp-ca \
  --source '{"sourceType":"CERTIFICATE_BUNDLE",
    "sourceData":{"x509CertificateData":""}}' \
  --profile china --region cn-north-1

aws rolesanywhere enable-trust-anchor --trust-anchor-id  ...

# 创建角色（信任 rolesanywhere 服务，限定 Trust Anchor）
# 权限：仅允许 GetSecretValue + KMS Decrypt

# 创建 Profile
aws rolesanywhere create-profile \
  --name sftp-profile \
  --role-arns '["arn:aws-cn:iam:::role/sftp-secrets-reader-role"]' \
  --duration-seconds 3600 \
  --profile china --region cn-north-1

aws rolesanywhere enable-profile --profile-id  ...

# 注册 CA 为 Trust Anchor
aws rolesanywhere create-trust-anchor \
  --name sftp-ca \
  --source '{"sourceType":"CERTIFICATE_BUNDLE",
    "sourceData":{"x509CertificateData":"<CA-CERT-PEM>"}}' \
  --profile china --region cn-north-1

aws rolesanywhere enable-trust-anchor --trust-anchor-id <ID> ...

# 创建角色（信任 rolesanywhere 服务，限定 Trust Anchor）
# 权限：仅允许 GetSecretValue + KMS Decrypt

# 创建 Profile
aws rolesanywhere create-profile \
  --name sftp-profile \
  --role-arns '["arn:aws-cn:iam::<ACCOUNT>:role/sftp-secrets-reader-role"]' \
  --duration-seconds 3600 \
  --profile china --region cn-north-1

aws rolesanywhere enable-profile --profile-id <ID> ...

#!/usr/bin/env python3
"""完整链路测试：证书 → 临时凭据 → 取密码 → SFTP"""
import json, subprocess, boto3, paramiko

# 1. X.509 证书 → Roles Anywhere 换取临时凭据
result = subprocess.run([
    "./aws_signing_helper", "credential-process",
    "--certificate", "pki/client.crt",
    "--private-key", "pki/client.key",
    "--trust-anchor-arn", "
",
    "--profile-arn", "
",
    "--role-arn", "",
    "--region", "cn-north-1"
], capture_output=True, text=True, check=True)
creds = json.loads(result.stdout)
print(f"临时凭据获取成功，过期时间：{creds['Expiration']}")

# 2. 临时凭据 → Secrets Manager 取密码
session = boto3.Session(
    aws_access_key_id=creds["AccessKeyId"],
    aws_secret_access_key=creds["SecretAccessKey"],
    aws_session_token=creds["SessionToken"],
    region_name="cn-north-1"
)
sm = session.client("secretsmanager")
password = json.loads(
    sm.get_secret_value(SecretId="sftp/prod/app1")["SecretString"]
)["Password"]
print(f"密码已获取（长度={len(password)}）")

# 3. 密码 → 连接 SFTP
transport = paramiko.Transport(("", 22))
transport.connect(username="app1", password=password)
sftp = paramiko.SFTPClient.from_transport(transport)

sftp.open("/test.txt", "w").write("Hello SFTP!\n")
print("上传成功")
print(f"目录列表：{sftp.listdir('/')}")
sftp.remove("/test.txt")
print("全部验证通过！")

临时凭据获取成功，过期时间：2026-06-25T15:05:22Z
密码已获取（长度=32）
上传成功
目录列表：['test.txt']
全部验证通过！