mcpField
A
aws

aws

github.com/vercel-labs测试与质量2.0k225 次安装6 天前更新
代码仓库

关于 aws

aws 是用于测试应用的 AWS API 模拟器,以内存状态模拟 S3、SQS、Cognito、IAM 和 STS,并兼容 AWS SDK 的相关请求格式。

主要功能

  1. 01模拟 S3、SQS、Cognito、IAM 和 STS
  2. 02兼容 AWS SDK 请求
  3. 03内存保存服务状态
  4. 04通过种子配置初始化数据
  5. 05提供状态检查器

适用场景

  1. 01测试 S3 文件上传与下载
  2. 02验证 SQS 消息处理流程
  3. 03测试 Cognito 注册与认证
  4. 04检查 IAM 和 STS API 集成

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/vercel-labs/emulate --skill aws

安装后在 Claude Code 中输入/aws确认已加载。

详细介绍

AWS 模拟器

S3、SQS、Cognito、IAM 和 STS 模拟,支持与 AWS SDK 兼容的 S3 路径。S3 上传和下载会保留任意二进制负载,包括原始字节长度和 ETag。SQS 支持 AWS JSON 1.0 和 XML Query 协议;IAM 和 STS 使用 XML Query。所有状态均保存在内存中。

启动

默认情况下,本地监听器绑定到 127.0.0.1。若要从容器或其他机器访问,请在 CLI 中传入 --host 0.0.0.0,或在 createEmulator 中传入 hostname: '0.0.0.0'。配置 --base-url 或 baseUrl,以提供这些客户端可访问的对外 URL。

# AWS only
npx emulate --service aws

# Default port (when run alone)
# http://localhost:4000

也可以通过编程方式启动:

import { createEmulator } from 'emulate'

const aws = await createEmulator({ service: 'aws', port: 4006 })
// aws.url === 'http://localhost:4006'

身份验证

以 Authorization: Bearer <token> 的形式传递令牌。范围权限使用 s3:*、sqs:*、iam:*、sts:* 模式。

curl http://localhost:4006/ \
  -H "Authorization: Bearer test_token_admin"

将应用指向模拟器

环境变量
AWS_EMULATOR_URL=http://localhost:4006
AWS SDK v3
import { S3Client } from '@aws-sdk/client-s3'

const s3 = new S3Client({
  endpoint: process.env.AWS_EMULATOR_URL,
  region: 'us-east-1',
  credentials: {
    accessKeyId: 'AKIAIOSFODNN7EXAMPLE',
    secretAccessKey: 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
  },
  forcePathStyle: true,
})
import { SQSClient } from '@aws-sdk/client-sqs'

const sqs = new SQSClient({
  endpoint: `${process.env.AWS_EMULATOR_URL}/sqs`,
  region: 'us-east-1',
  credentials: {
    accessKeyId: 'AKIAIOSFODNN7EXAMPLE',
    secretAccessKey: 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
  },
})
import { IAMClient } from '@aws-sdk/client-iam'

const iam = new IAMClient({
  endpoint: `${process.env.AWS_EMULATOR_URL}/iam`,
  region: 'us-east-1',
  credentials: {
    accessKeyId: 'AKIAIOSFODNN7EXAMPLE',
    secretAccessKey: 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
  },
})

种子配置

aws:
  region: us-east-1
  s3:
    buckets:
      - name: my-app-bucket
      - name: my-app-uploads
        region: eu-west-1
  sqs:
    queues:
      - name: my-app-events
      - name: my-app-dlq
        visibility_timeout: 60
      - name: my-app-orders.fifo
        fifo: true
  iam:
    users:
      - user_name: developer
        create_access_key: true
      - user_name: readonly-user
    roles:
      - role_name: lambda-execution-role
        description: Role for Lambda function execution
        assume_role_policy: '{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}'

默认种子(始终创建):S3 存储桶 emulate-default、SQS 队列 emulate-default-queue、IAM 用户 admin 及其访问密钥对(AKIAIOSFODNN7EXAMPLE / wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY)。

API 端点

S3

S3 路由使用与真实 AWS S3 线路格式匹配的根路径。也支持旧版带 /s3/ 前缀的路径。

# List all buckets
curl http://localhost:4006/ \
  -H "Authorization: Bearer $TOKEN"

# Create bucket
curl -X PUT http://localhost:4006/my-bucket \
  -H "Authorization: Bearer $TOKEN"

# Delete bucket (must be empty)
curl -X DELETE http://localhost:4006/my-bucket \
  -H "Authorization: Bearer $TOKEN"

# Head bucket (check existence, get region)
curl -I http://localhost:4006/my-bucket \
  -H "Authorization: Bearer $TOKEN"

# List objects (with prefix, delimiter, pagination)
curl "http://localhost:4006/my-bucket?prefix=uploads/&delimiter=/&max-keys=100" \
  -H "Authorization: Bearer $TOKEN"

# Put object
curl -X PUT http://localhost:4006/my-bucket/path/to/file.txt \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: text/plain" \
  -H "x-amz-meta-author: test" \
  --data-binary "file contents"

# Get object
curl http://localhost:4006/my-bucket/path/to/file.txt \
  -H "Authorization: Bearer $TOKEN"

# Head object (metadata only)
curl -I http://localhost:4006/my-bucket/path/to/file.txt \
  -H "Authorization: Bearer $TOKEN"

# Delete object
curl -X DELETE http://localhost:4006/my-bucket/path/to/file.txt \
  -H "Authorization: Bearer $TOKEN"

# Copy object
curl -X PUT http://localhost:4006/dest-bucket/copy.txt \
  -H "Authorization: Bearer $TOKEN" \
  -H "x-amz-copy-source: /source-bucket/original.txt"
SQS

当前 AWS SDK 客户端使用上文所示的 /sqs 端点。SQS 接受发送到 POST /sqs/ 的 AWS JSON 1.0 请求,其 Content-Type 为 application/x-amz-json-1.0,X-Amz-Target 为 AmazonSQS.<Action>,并返回 JSON 结果和错误。仍支持带有 Action 表单参数的 XML Query 请求。队列操作也接受返回的 QueueUrl 作为端点。

# List queues using AWS JSON 1.0
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/x-amz-json-1.0" \
  -H "X-Amz-Target: AmazonSQS.ListQueues" \
  -d '{}'
# Create queue
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "Action=CreateQueue&QueueName=my-queue"

# Create queue with attributes
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "Action=CreateQueue&QueueName=my-queue&Attribute.1.Name=VisibilityTimeout&Attribute.1.Value=30"

# List queues
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ListQueues"

# List queues with prefix filter
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ListQueues&QueueNamePrefix=my-"

# Get queue URL
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=GetQueueUrl&QueueName=my-queue"

# Get queue attributes
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=GetQueueAttributes&QueueUrl=<queue_url>"

# Send message
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=SendMessage&QueueUrl=<queue_url>&MessageBody=Hello+World"

# Send message with attributes
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=SendMessage&QueueUrl=<queue_url>&MessageBody=Hello&MessageAttribute.1.Name=type&MessageAttribute.1.Value.DataType=String&MessageAttribute.1.Value.StringValue=greeting"

# Receive messages
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ReceiveMessage&QueueUrl=<queue_url>&MaxNumberOfMessages=5"

# Delete message
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=DeleteMessage&QueueUrl=<queue_url>&ReceiptHandle=<receipt_handle>"

# Purge queue
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=PurgeQueue&QueueUrl=<queue_url>"

# Delete queue
curl -X POST http://localhost:4006/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=DeleteQueue&QueueUrl=<queue_url>"
IAM

所有 IAM 操作都使用 POST /iam/,并将 Action 作为 form-urlencoded 参数传递。

# Create user
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=CreateUser&UserName=new-user"

# Get user
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=GetUser&UserName=new-user"

# List users
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ListUsers"

# Delete user
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=DeleteUser&UserName=new-user"

# Create access key
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=CreateAccessKey&UserName=developer"

# List access keys
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ListAccessKeys&UserName=developer"

# Delete access key
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=DeleteAccessKey&UserName=developer&AccessKeyId=AKIA..."

# Create role
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=CreateRole&RoleName=my-role&AssumeRolePolicyDocument={}"

# Get role
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=GetRole&RoleName=my-role"

# List roles
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ListRoles"

# Delete role
curl -X POST http://localhost:4006/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=DeleteRole&RoleName=my-role"
Cognito

使用端点为 /cognito-idp 的 CognitoIdentityProviderClient,以及端点为 /cognito-identity 的 CognitoIdentityClient。两者均支持 AWS JSON 1.1,并可与官方 AWS SDK v3 配合使用:

import { CognitoIdentityProviderClient, InitiateAuthCommand } from "@aws-sdk/client-cognito-identity-provider";

const cognito = new CognitoIdentityProviderClient({
  endpoint: "http://localhost:4007/cognito-idp",
  region: "us-east-1",
  credentials: { accessKeyId: "local-access-key", secretAccessKey: "local-secret" },
});
const result = await cognito.send(new InitiateAuthCommand({
  ClientId: "emulateclient",
  AuthFlow: "USER_PASSWORD_AUTH",
  AuthParameters: { USERNAME: "developer", PASSWORD: "LocalPass123!" },
}));

系统会创建默认用户池 us-east-1_emulate、公共应用客户端 emulateclient 和身份池 us-east-1:00000000-0000-4000-8000-000000000001,但不会创建用户。为上面的示例创建一个已确认的账户:

aws:
  cognito:
    user_pools:
      - id: us-east-1_emulate
        confirmation_code: "123456"
        users:
          - username: developer
            password: LocalPass123!
            attributes:
              email: [email protected]
              email_verified: "true"
              name: Developer

用户池还支持 name 和 clients,其中 clients 可包含 client_id、可选的 client_secret、精确的 redirect_uris 和允许的 scopes。用户支持 status(CONFIRMED、UNCONFIRMED 或 FORCE_CHANGE_PASSWORD)和 enabled。密码至少需要八个字符,并以加盐哈希形式存储。机密客户端在使用 SDK 流程时需要提供 Cognito 的 HMAC SecretHash 或 SECRET_HASH。

  • SignUp、ConfirmSignUp、ResendConfirmationCode、ForgotPassword 和 ConfirmForgotPassword 用于模拟账户确认和恢复。验证码在十分钟后过期,并使用用户池的 confirmation_code fixture,默认值为 123456。系统不会发送电子邮件或短信。可在 /_inspector?tab=cognito 查看有效验证码;已创建但尚未确认的用户可通过 ResendConfirmationCode 请求验证码。
  • InitiateAuth 支持 USER_PASSWORD_AUTH、REFRESH_TOKEN_AUTH 和 REFRESH_TOKEN。AdminInitiateAuth 还支持 ADMIN_USER_PASSWORD_AUTH。通过管理员创建的临时密码会返回 NEW_PASSWORD_REQUIRED,可通过 RespondToAuthChallenge 或 AdminRespondToAuthChallenge 完成。挑战会话在三分钟后过期,并与用户和客户端绑定。
  • GetUser、ChangePassword、GlobalSignOut 和 RevokeToken 使用已签发的令牌。AdminCreateUser(带有 TemporaryPassword)、AdminGetUser、AdminConfirmSignUp、AdminSetUserPassword、AdminEnableUser、AdminDisableUser、AdminDeleteUser 和分页的 ListUsers 支持测试设置和管理。

令牌是带有 Cognito token_use 声明的 RS256 JWT。ID 令牌通过 aud 标识应用客户端;访问令牌使用 client_id。ID 令牌和访问令牌在一小时后过期;刷新令牌在三十天后过期。登出、撤销、停用、删除和密码重置会使受影响的会话失效。用户、会话、身份和签名密钥都会纳入快照、恢复和重置。

本地颁发者为 http://localhost:4007/cognito-idp/USER_POOL_ID。发现文档和公钥分别通过 ISSUER/.well-known/openid-configuration 和 ISSUER/.well-known/jwks.json 提供。配置令牌验证器时,请使用此本地颁发者和 JWKS。ISSUER/oauth2/authorize 会显示本地用户选择器,ISSUER/oauth2/token 会交换授权码或刷新令牌,ISSUER/oauth2/userInfo 会返回具有相应作用域的个人资料声明。授权码在五分钟后过期,只能使用一次,并与客户端、精确的重定向 URI 和可选的 S256 PKCE 挑战绑定。客户端身份验证支持 client_secret_basic 或 client_secret_post。默认客户端允许 http://localhost:3000/callback。

身份池支持 GetId 和 GetCredentialsForIdentity。请在 Logins 中通过 cognito-idp.REGION.amazonaws.com/USER_POOL_ID 提供 ID 令牌。默认身份池接受默认用户池和应用客户端。自定义的 aws.cognito.identity_pools 条目可包含 id、可选的 name、providers: [{ user_pool_id, client_id }] 和 allow_unauthenticated_identities(默认值为 false)。对于相同的已验证主体,身份保持稳定。凭证在一小时后过期;系统会检查登录签名、过期时间、受众和身份所有权。

系统不会模拟 SRP、MFA、外部身份提供方、Lambda 触发器、用户池和客户端管理 API、ListUsers 筛选条件以及身份池角色选择。管理员请求遵循 AWS 模拟器的宽松凭证行为:不会评估 SigV4 签名和 IAM 策略。身份池凭证是生成的测试凭证,不会进行角色或策略评估。

STS

所有 STS 操作均使用 POST /sts/,并将 Action 作为 form-urlencoded 参数。

# Get caller identity
curl -X POST http://localhost:4006/sts/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=GetCallerIdentity"

# Assume role
curl -X POST http://localhost:4006/sts/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=AssumeRole&RoleArn=arn:aws:iam::123456789012:role/my-role&RoleSessionName=my-session"
Inspector
# HTML dashboard (shows S3, SQS, IAM state)
curl http://localhost:4006/_inspector?tab=s3
curl http://localhost:4006/_inspector?tab=sqs
curl http://localhost:4006/_inspector?tab=iam

常见模式

上传和获取对象
TOKEN="test_token_admin"
BASE="http://localhost:4006"

# Create bucket
curl -X PUT $BASE/my-data \
  -H "Authorization: Bearer $TOKEN"

# Upload file
curl -X PUT $BASE/my-data/config.json \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  --data-binary '{"key": "value"}'

# Download file
curl $BASE/my-data/config.json \
  -H "Authorization: Bearer $TOKEN"
发送和接收 SQS 消息
TOKEN="test_token_admin"
BASE="http://localhost:4006"

# Get queue URL
QUEUE_URL=$(curl -s -X POST $BASE/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=GetQueueUrl&QueueName=emulate-default-queue" | grep -oP '<QueueUrl>\K[^<]+')

# Send message
curl -X POST $BASE/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=SendMessage&QueueUrl=$QUEUE_URL&MessageBody=Hello+from+emulate"

# Receive messages
curl -X POST $BASE/sqs/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=ReceiveMessage&QueueUrl=$QUEUE_URL&MaxNumberOfMessages=1"
创建带访问密钥的 IAM 用户
TOKEN="test_token_admin"
BASE="http://localhost:4006"

# Create user
curl -X POST $BASE/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=CreateUser&UserName=ci-user"

# Generate access key
curl -X POST $BASE/iam/ \
  -H "Authorization: Bearer $TOKEN" \
  -d "Action=CreateAccessKey&UserName=ci-user"

与内置模拟器搭配使用的自定义模拟器

使用 npx emulate init --custom inventory 搭建第三方 API 模拟器并进行测试。使用 emulate 中的 defineConfig 在 emulate.config.ts 中注册该模拟器,并与 { emulator: "aws" } 等内置条目并列。运行 npx emulate start --watch 以重新加载导入内容,并在打印出的 /_emulate URL 中检查自定义状态。成功重新加载后,运行会重置为初始数据。现有的扁平种子配置仍然可用;--config 用于指定文件。有关编写和测试第三方 API 模拟器的信息,请参阅 https://emulate.dev/docs/custom-emulators。

评论

支持 Markdown