一个完整的 AWS 无服务器架构教程

字段名	类型	说明
currentVersion	string	当前固件版本
targetVersion	string	目标固件版本（若为空则无更新）
deviceType	string	设备类型（用于选择固件）
lastCheck	string	最后一次检查时间（ISO8601）
updateStatus	string	`pending` / `downloaded` / `applied` / `failed`
firmwareKey	string	S3 中的固件对象键（如 `firmware/device-type-a/v1.2.3.bin`）

创建步骤：

进入 DynamoDB 控制台
点击“创建表”
表名：OtaDevices
分区键：deviceId（类型：字符串）
其他保持默认，点击“创建表”

示例数据：

{
"deviceId": "device-001",
"currentVersion": "v1.1.0",
"targetVersion": "v1.2.3",
"deviceType": "sensor-x1",
"firmwareKey": "firmware/sensor-x1/v1.2.3.bin",
"updateStatus": "pending"
}

四、创建 Lambda 函数

我们将创建一个 Lambda 函数，处理 /check 和 /report 逻辑。

4.1 创建 Lambda

进入 Lambda 控制台
点击“创建函数” → “从头开始创作”
名称：OtaHandler
运行时：Python 3.12（或 Node.js 18.x，本教程用 Python）
权限：创建新角色（稍后手动添加权限）

4.2 编写代码（Python）

import json
import boto3
import os
from datetime import datetime, timedelta
from botocore.exceptions import ClientError
# 初始化客户端
dynamodb = boto3.resource('dynamodb')
s3_client = boto3.client('s3')
table = dynamodb.Table(os.environ['DYNAMODB_TABLE'])
BUCKET_NAME = os.environ['S3_BUCKET']
def lambda_handler(event, context):
print("Event:", json.dumps(event))
# 获取 HTTP 方法和路径
http_method = event['httpMethod']
resource_path = event['resource']
# 简单鉴权：检查 header 中的 token（实际应使用更安全机制，如 JWT、API Key）
auth_token = event['headers'].get('Authorization') or event['headers'].get('authorization')
if not auth_token or not auth_token.startswith("Bearer "):
return {
'statusCode': 401,
'body': json.dumps({'error': 'Missing or invalid Authorization header'})
}
device_id = auth_token.split(" ")[1]  # 假设 token 是 "Bearer <deviceId>"try:if resource_path == '/check' and http_method == 'GET':return handle_check(device_id)elif resource_path == '/report' and http_method == 'POST':body = json.loads(event['body']) if event.get('body') else {}return handle_report(device_id, body)else:return {'statusCode': 404, 'body': json.dumps({'error': 'Not found'})}except Exception as e:print("Error:", str(e))return {'statusCode': 500, 'body': json.dumps({'error': 'Internal error'})}def handle_check(device_id):# 1. 查询设备信息response = table.get_item(Key={'deviceId': device_id})if 'Item' not in response:return {'statusCode': 404, 'body': json.dumps({'error': 'Device not found'})}item = response['Item']# 2. 检查是否有待更新if not item.get('targetVersion') or item.get('currentVersion') == item.get('targetVersion'):return {'statusCode': 200,'body': json.dumps({'updateAvailable': False})}# 3. 生成预签名 URL（有效期 10 分钟）firmware_key = item['firmwareKey']try:presigned_url = s3_client.generate_presigned_url('get_object',Params={'Bucket': BUCKET_NAME, 'Key': firmware_key},ExpiresIn=600  # 10 分钟)except ClientError as e:return {'statusCode': 500, 'body': json.dumps({'error': 'Failed to generate URL'})}# 4. 更新 lastCheck 时间table.update_item(Key={'deviceId': device_id},UpdateExpression="SET lastCheck = :t",ExpressionAttributeValues={':t': datetime.utcnow().isoformat()})return {'statusCode': 200,'body': json.dumps({'updateAvailable': True,'version': item['targetVersion'],'downloadUrl': presigned_url,'firmwareKey': firmware_key})}def handle_report(device_id, report_data):# 报告内容示例：{"status": "success", "message": "Applied v1.2.3"}status = report_data.get('status', 'unknown')message = report_data.get('message', '')# 更新设备状态table.update_item(Key={'deviceId': device_id},UpdateExpression="SET updateStatus = :s, currentVersion = :v, lastReport = :t",ExpressionAttributeValues={':s': status,':v': report_data.get('version', 'unknown'),':t': datetime.utcnow().isoformat()})return {'statusCode': 200,'body': json.dumps({'message': 'Report received'})}

4.3 配置环境变量

在 Lambda 函数的“配置” → “环境变量”中添加：

键	值
`DYNAMODB_TABLE`	`OtaDevices`
`S3_BUCKET`	`my-ota-firmware-bucket-12345`

4.4 添加 IAM 权限

编辑 Lambda 的执行角色（在“配置” → “权限”中点击角色名），附加以下策略：

{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"dynamodb:GetItem",
"dynamodb:UpdateItem"
],
"Resource": "arn:aws:dynamodb:YOUR_REGION:YOUR_ACCOUNT:table/OtaDevices"
},
{
"Effect": "Allow",
"Action": [
"s3:GetObject"
],
"Resource": "arn:aws:s3:::my-ota-firmware-bucket-12345/*"
}
]
}

替换 YOUR_REGION 和 YOUR_ACCOUNT（或直接使用控制台 ARN）

五、配置 API Gateway

5.2 创建资源和方法

创建 `/check` 资源

在“资源”中点击“操作” → “创建资源”
- 资源名称：check
- 资源路径：check
选中 /check，点击“操作” → “创建方法” → 选择 GET → 勾选“使用 Lambda 代理集成”
Lambda 函数：选择 OtaHandler
点击“保存”，确认权限

创建 `/report` 资源

同样创建资源 report
添加 POST 方法，同样指向 OtaHandler，启用 Lambda 代理集成

5.3 部署 API

点击“操作” → “部署 API”
新建阶段（如 prod）

记下调用 URL，例如：

https://abc123.execute-api.us-east-1.amazonaws.com/prod

六、测试流程

6.1 上传固件到 S3

aws s3 cp firmware/sensor-x1/v1.2.3.bin s3://my-ota-firmware-bucket-12345/firmware/sensor-x1/v1.2.3.bin

6.2 在 DynamoDB 插入测试设备

{
"deviceId": "device-001",
"currentVersion": "v1.1.0",
"targetVersion": "v1.2.3",
"deviceType": "sensor-x1",
"firmwareKey": "firmware/sensor-x1/v1.2.3.bin",
"updateStatus": "pending"
}

6.3 调用 /check

curl -H "Authorization: Bearer device-001" \
https://abc123.execute-api.us-east-1.amazonaws.com/prod/check

✅ 响应示例：

{
"updateAvailable": true,
"version": "v1.2.3",
"downloadUrl": "https://my-ota-firmware-bucket-12345.s3.amazonaws.com/...?X-Amz-Signature=..."
}

6.4 调用 /report

curl -X POST \
-H "Authorization: Bearer device-001" \
-H "Content-Type: application/json" \
-d '{"status": "success", "version": "v1.2.3"}' \
https://abc123.execute-api.us-east-1.amazonaws.com/prod/report

✅ 响应：{"message": "Report received"}

七、安全增强建议（生产环境）

鉴权升级：
- 使用 API Gateway 的 自定义授权方（Custom Authorizer） 或 Cognito
- 或使用 API Key + Usage Plan
输入验证：
- 在 Lambda 中校验 deviceId 格式
- 防止路径遍历（如 firmwareKey 不应包含 ../）
日志与监控：
- 启用 CloudWatch Logs
- 设置告警（如大量 401 请求）
S3 安全：
- 确保 S3 存储桶策略禁止公开读取
- 启用服务器端加密（SSE-S3）
Lambda 超时与内存：
- 设置合理超时（如 10 秒）
- 内存 128MB 足够

八、清理（避免产生费用）

删除 API Gateway
删除 Lambda 函数
清空并删除 S3 存储桶
删除 DynamoDB 表

✅ 至此，你已成功搭建一个基于 AWS 的 OTA 固件分发服务！

一个完整的 AWS 无服务器架构教程 - 教程

一、准备工作

二、创建 S3 存储桶（用于固件）

三、创建 DynamoDB 表

其他属性（可选，根据业务扩展）：

四、创建 Lambda 函数

4.2 编写代码（Python）

4.3 配置环境变量

五、配置 API Gateway

5.2 创建资源和方法

创建 /report 资源

六、测试流程

6.2 在 DynamoDB 插入测试设备

6.3 调用 /check

6.4 调用 /report

七、安全增强建议（生产环境）

创建 `/report` 资源