使用 Amazon CloudFront 和 Lambda @Edge 完成 Amazon S3 平滑迁移

背景简介

随着国内企业发展全球业务的进一步深入，考虑到亚马逊云科技全球云资源广泛分布的优势，很多国内企业发展海外业务时选择亚马逊云科技做为其重要的海外 IT 基础设施平台。本文的企业在前期已经将其海外业务系统迁移部署到亚马逊云科技海外区域，客户在海外有用数万台 IoT 设备，且数量伴随业务增长还在持续增加。由于历史原因，客户的 IoT 设备用其他云厂商的 SDK 或 Rest API 上传数据文件到对象存储 OSS，且设备不便更改现有配置，故而保留原始方案，设备数据依旧保持上报到其他云厂商。导致日常运维人员需要将数据从其他云平台复制到 Amazon S3，运维角度来看，需要更加便捷、高效的方案。

Amazon S3 在全球范围内有着良好的覆盖率与可用性，客户业务系统迁移至亚马逊云科技后，鉴于其全球业务的需要，希望将数据上报一同迁移到 Amazon S3。因为客户在全球拥有数万台设备，修改代码会耗费大量人力物力、开发测试周期长，且会带来故障风险。所以为避免风险，且尽快完成 IoT 设备端上传目标的修改，本文设计的方案是避免修改 IoT 设备代码，仅通过配置下发即可完成迁移。

本文的方案同样适用于其他大量客户的类似场景，即业务系统已经在亚马逊云科技，客户希望将设备端上传数据的目标存储从友商的对象存储 OSS 迁移至 Amazon S3 对象存储，但是 IoT 端又不便修改原代码的情况。

方案与架构设计

方案的主要目的是确保用于将文件上传到友商对象存储的 IoT 设备能够无缝地上传到 Amazon S3。此外，为了提高性能和可靠性，通过将 CloudFront 置于 Amazon S3 之前来引入内容分发网络（CDN），避免直接暴露 S3 链接。目标是仅通过修改定义设备使用的端点和凭据的配置文件来处理所有这些问题。由于 Amazon S3 及其 SDK 和 API 被业内作为经典案例或标准方案使用，所以我们认为在云端进行一定改动，即可完成此适配。

架构设计

涉及的服务

Amazon CloudFront：CloudFront 可加快分发静态和动态 Web 内容（例如，.html、.css、.php、图像和媒体文件）的过程。当用户请求内容时，CloudFront 通过可提供低延迟和高性能的全球边缘站点网络交付相应内容。本项目中使用 Cloudfront distribution 做为 S3 桶的接入点（endpoint）。对于不同的 S3 桶，需要创建不同的 distribution 对应。
Amazon Lambda：Lambda 在可用性高的计算基础设施上运行您的代码，执行计算资源的所有管理工作，其中包括服务器和操作系统维护、容量预置和弹性伸缩和记录。使用 Lambda，您只需在 Lambda 支持的一种语言运行时系统中提供代码。本项目中有两个 Lambda function，第一个 Lambda function 做为 API Gateway 的授权方（Authorizer），验证请求的合法性；第二个 Lambda function 可以向 S3 桶发起分片上传请求并拿到 Upload ID。
Amazon Lambda@Edge：Lambda@Edge 是 AWS Lambda 的扩展。Lambda@Edge 是一项计算服务，可用于执行函数以自定义 Amazon CloudFront 提供的内容。您可以在某个 AWS 区域，比如美国东部（弗吉尼亚州北部）的 Lambda 控制台中编写 Node.js 或 Python 函数。本项目中使用 Lambda@Edge 对 SDK 发起的请求进行分类，把发起分片上传的请求路由到 API Gateway。
Amazon API Gateway：API Gateway 是一项 AWS 服务，用于创建、发布、维护、监控和保护任意规模的 REST、HTTP 和 WebSocket API。API 开发人员可以创建能够访问 AWS 或其他 Web 服务以及存储在 AWS 云中的数据的 API。作为 API Gateway API 开发人员，您可以创建 API 以在您自己的客户端应用程序中使用。或者，您可以将您的 API 提供给第三方应用程序开发人员。本项目中使用 API Gateway 管理发起分片上传的请求。
Amazon S3：Amazon Simple Storage Service（Amazon S3）是一项对象存储服务，在可扩展性、数据可用性、安全性和能效方面业界领先。数百万不同规模和行业的客户可以为几乎任何应用场景存储、管理、分析和保护任意数量的数据，例如数据湖、云原生应用程序和移动应用程序。借助高成本效益的存储类和易于使用的管理功能，您可以优化成本、组织并分析数据，以及配置精细调整过的访问控制，从而满足特定的业务和合规性要求。本项目中使用 S3 替换友商的对象存储服务，保存用户数据。

具体实现方案

本方案实现上传 API 的迁移（5 个 API，分别是PutObjectComman,InitiateMultipartUploadRequest,UploadPart,CompleteMultipartUpload,AbortMultipartUpload），分别对应普通上传和分片上传的需要。

步骤 1：在 Amazon S3 前面设置 CloudFront

添加 CloudFront 提供了额外的性能改进和安全性。配置涉及创建以 S3 存储桶为源的 CloudFront 分发。 CloudFront 不仅通过缓存内容提高了传输速度，还减少了 S3 存储桶的直接暴露，增强了安全性。

需要注意的是，此 S3 根据最小授权原则，仅开启指定存储桶到 CloudFront 策略即可。具体策略如下（仅赋予上传权限，不允许 Get、Delete、List 权限）：

<bucket name><CloudFront distribution ARN> 替换为您真实的内容

{
    "Version": "2008-10-17",
    "Id": "PolicyForCloudFrontPrivateContent",
    "Statement": [
        {
            "Sid": "AllowCloudFrontServicePrincipal",
            "Effect": "Allow",
            "Principal": {
                "Service": "cloudfront.amazonaws.com"
            },
            "Action": [
                "s3:PutObject",
                "s3:AbortMultipartUpload"
            ],
            "Resource": "arn:aws:s3:::<bucket name>/*",
            "Condition": {
                "StringEquals": {
                    "AWS:SourceArn": "<CloudFront distribution ARN>"
                }
            }
        }
    ]
}

注意：CloudFront 默认不会传递部分请求头和 Query 参数，需要您在 CloudFront 中配置(CloudFront 的行为编辑页面）。请参考下图配置：

步骤 2：CloudFront 设置CloudFront Function，在请求侧增加 Lambda @Edge，通过 API Gateway 特殊处理部分请求

验证 V1 签名的 CloudFront Function 代码：

注意，CloudFront Function 运行时仅完全支持 ES5.1 语法，部分支持 ES6-ES12 的新特性，如采用不支持的语法代码执行过程会报错（直接 return 503）。具体支持的语法请参考 https://docs.aws.amazon.com/AmazonCloudFront/latest/DeveloperGuide/functions-javascript-runtime-20.html

const crypto = require("crypto");

const ACCESS_KEY_ID = "ACCESS_KEY_ID"; // 您的AK 不一定是真实的 要保证和请求发起端一致

const SECRET_ACCESS_KEY = "SECRET_ACCESS_KEY"; // 您的SK 不一定是真实的 要保证和请求发起端一致

function handler(event) {
  var request = event.request;
  var headers = request.headers;
  // Retrieve the Authorization header
  if (!headers.authorization) {
    return generateErrorResponse(401, "Authorization header missing");
  }

  var authorization = headers.authorization.value;
  // Check if Authorization header starts with "OSS "
  if (!authorization.startsWith("OSS ")) {
    return generateErrorResponse(401, "Invalid Authorization header format");
  }
  // Extract AccessKeyId and Signature from Authorization header
  var authParts = authorization.substring(4).split(":");
  var accessKeyId = authParts[0];

  if (authParts.length !== 2 || accessKeyId !== ACCESS_KEY_ID) {
    return generateErrorResponse(401, "Invalid Authorization header format");
  }

  var providedSignature = authParts[1];
  // Retrieve Date header
  if (!headers.date) {
    return generateErrorResponse(400, "Date header missing");
  }
  var date = headers.date.value;
  // Validate Date format (should be GMT)
  if (!isValidGMTDate(date)) {
    return generateErrorResponse(
      400,
      "Invalid Date format. Expected GMT format"
    );
  }
  // 大于5分钟请求进行拦截
  const requestDate = new Date(date);
  if (new Date() - requestDate > 300000) {
    return generateErrorResponse(
      400,
      "Invalid Date format. Expected GMT format"
    );
  }

  // Retrieve optional headers
  var contentMD5 = headers["content-md5"] ? headers["content-md5"].value : "";
  var contentType = headers["content-type"]
    ? headers["content-type"].value
    : "";

  // Construct CanonicalizedOSSHeaders
  var canonicalizedOSSHeaders = getCanonicalizedOSSHeaders(headers);

  // Construct CanonicalizedResource (Assuming the request.uri contains the resource path)
  const bucketName = headers.host.value.split(".")[0];

  const params = getURLParams(event.request.querystring);

  var canonicalizedResource = `/${bucketName}${decodeURIComponent(
    event.request.uri
  )}${params}`;

  // Construct StringToSign
  var stringToSign = [
    request.method.toUpperCase(),
    contentMD5 || "",
    contentType || "",
    date,
    canonicalizedOSSHeaders + canonicalizedResource,
  ].join("\n");

  // Retrieve AccessKeySecret corresponding to AccessKeyId
  var accessKeySecret = SECRET_ACCESS_KEY;
  if (!accessKeySecret) {
    return generateErrorResponse(403, "Invalid AccessKeyId");
  }

  // Compute expected Signature
  var expectedSignature = computeSignature(accessKeySecret, stringToSign);
  // Compare signatures
  if (providedSignature !== expectedSignature) {
    return generateErrorResponse(403, "Signature mismatch");
  }
  // Authorization successful, proceed with the request
  return request;
}

function getURLParams(paramObject) {
  if (!paramObject) return "";
  let keyList = Object.keys(paramObject);
  if (!keyList || keyList.length === 0) return "";
  let result = "?";
  keyList = keyList.sort();
  keyList.forEach((item, index) => {
    if (index === 0) {
      result = result + item;
    } else {
      result = result + "&" + item;
    }
    if (paramObject[item].value) {
      result = result + "=" + paramObject[item].value;
    }
  });
  console.log("getURLParams");
  console.log(result);
  return result;
}

function hmacSha1(key, data) {
  return crypto.createHmac("sha1", key).update(data).digest();
}
function base64Encode(data) {
  return Buffer.from(data).toString("base64");
}

// Helper function to compute Signature using HMAC-SHA1 and base64 encoding
function computeSignature(accessKeySecret, stringToSign) {
  const signature = hmacSha1(accessKeySecret, stringToSign);
  return base64Encode(signature);
}

// Helper function to construct CanonicalizedOSSHeaders
function getCanonicalizedOSSHeaders(headers) {
  var ossHeaders = [];

  for (var headerName in headers) {
    if (headerName.startsWith("x-oss-")) {
      ossHeaders.push(headerName.toLowerCase());
    }
  }

  ossHeaders.sort();

  var canonicalizedHeaders = "";
  ossHeaders.forEach(function (headerName) {
    canonicalizedHeaders +=
      headerName + ":" + headers[headerName].value.trim() + "\n";
  });

  return canonicalizedHeaders;
}

// Helper function to validate GMT date format
function isValidGMTDate(dateString) {
  var gmtRegex =
    /^[A-Za-z]{3},\s\d{2}\s[A-Za-z]{3}\s\d{4}\s\d{2}:\d{2}:\d{2}\sGMT$/;
  return gmtRegex.test(dateString);
}

// Helper function to generate error response
function generateErrorResponse(statusCode, message) {
  return {
    statusCode: statusCode.toString(),
    statusDescription: message,
    headers: {
      "content-type": { value: "text/plain" },
    },
    body: message,
  };
}

// 测试数据

const data = {
  version: "1.0",
  context: {
    distributionDomainName: "<distributionName>.cloudfront.net",
    distributionId: "<distributionId>",
    eventType: "viewer-response",
    requestId: "ZrtBMn3adU5rfBf487PLKHZGeYV2rx1-e-zQXGTBuSDnl0Az813OzA==",
  },
  viewer: { ip: "demoIp" },
  request: {
    method: "POST",
    uri: "/example%2FexampleMp4.mp4",
    querystring: { uploads: { value: "" } },
    headers: {
      "user-agent": {
        value: "xxxx",
      },
      "cloudfront-viewer-http-version": { value: "1.1" },
      date: { value: "Tue, 03 Sep 2024 10:21:17 GMT" },
      "cloudfront-is-tablet-viewer": { value: "false" },
      "cloudfront-is-smarttv-viewer": { value: "false" },
      "cloudfront-is-android-viewer": { value: "false" },
      "cloudfront-viewer-asn": { value: "16509" },
      host: { value: "<distributionName>.cloudfront.net" },
      accept: { value: "*/*" },
      "cloudfront-viewer-country": { value: "SG" },
      "cloudfront-is-mobile-viewer": { value: "false" },
      "cloudfront-viewer-city": { value: "Singapore" },
      "content-type": { value: "video/mp4" },
      "cloudfront-forwarded-proto": { value: "http" },
      "cloudfront-viewer-address": { value: "xxxx:5677" },
      "cloudfront-is-desktop-viewer": { value: "true" },
      "accept-encoding": { value: "identity" },
      "content-length": { value: "0" },
      "cloudfront-is-ios-viewer": { value: "false" },
      authorization: { value: "OSS ACCESS_KEY_ID:4RTCTg9Xa+A6PlrZ+YliGMbxhbY=" },
    },
    cookies: {},
  },
  response: {
    statusCode: 200,
    statusDescription: "OK",
    headers: {
      server: { value: "CloudFront" },
      date: { value: "Tue, 03 Sep 2024 10:21:21 GMT" },
      "x-amz-apigw-id": { value: "xxx" },
      "x-amzn-trace-id": {
        value:
          "Root=1-66d6e31f-39844ed46c9a4aec67b6af8a;Parent=0c0916df5d6528e0;Sampled=0;lineage=2:146c3b21:0",
      },
      via: {
        value:
          "1.1 3c724fc8704aec61a7bab068ccd978fe.cloudfront.net (CloudFront)",
      },
      "content-type": { value: "application/xml" },
      "content-length": { value: "375" },
    },
    cookies: {},
  },
};

如何去除/兼容？

由于各友商可能对响应正文有修改，所以在此列出较为特殊的返回值，如无特殊说明则表示默认兼容。

Amazon S3 CreateMultipartUploadCommand API：

<?xml version="1.0" encoding="UTF-8"?>
  <InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
  <Bucket>demo-bucket</Bucket>
  <Key>example/exampleMp4.mp4</Key>
  <UploadId>SyKw2ih1NT_YWR7n4E92OGkm_ykauFQpQA43w4vLhE7dH3v8h5Plk7A2V2a6nbhCIfVa3CTOGKAx6f7eBClhBD8dZqXo0lkpwcxxBXb4Om6sZsWg3Elmh1OYMOkFl71DTWqEpnGQl7.yzP1KQ3UvO_6eaJTvIv3ipgKQF8rQttk-</UploadId>
</InitiateMultipartUploadResult>

注意，Amazon S3 在此 API 会多返回一个 xmlns，部分友商 SDK 没有对此兼容，所以需要其他手段在云端将 xmlns 部分去除。

已知 CloudFront 的响应的 Body 部分在 Lambda @Edge 或 Function 拿不到，所以需要增加路由，对 CreateMultipartUploadCommand API 进行特殊处理。此 API 最终是通过/upload 路径发起 POST 请求，所以对此请求做特殊转发，将其转发至指定 API Gateway，API Gateway 后接入 Lambda，处理响应内容，按照所需结果处理返回值。其他特殊响应的 API 也可以按此逻辑处理。

设置鉴权

本方案面对的是互联网上的 IoT 设备，所以采取了一些安全措施防止匿名访问：

S3 桶为私有桶，对 CloudFront 增加权限。CloudFront 只能通过 OAC 访问。
S3 桶对 CloudFront 仅开放 PutObject 和 AbortMultipartUpload 权限。
在 CloudFront 侧的 Viewer request 事件中对请求进行签名验证。需要用户提供伪造的 OSS key 和 secret，在 CloudFront Function 中对友商 SDK 生成的签名进行验签。Amazon S3 API 在 CloudFront 侧的 Viewer request 事件中友商 SDK SignerV1 签名验证功能（需要在代码中配置 AK SK）。其余 API 会进行屏蔽。
API gateway 后面的 Lambda authorizer 也会使用相同的方法对请求进行验签。
虚拟的 OSS AK SK 需要配置在 Secrets Manager 中。

步骤 3：终端和凭证修改

向 IoT 设备下发 AK，SK，Endpoint。注意 Endpoint 可以按照友商格式下发。例如：

import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# 从环境变量中获取访问凭证。运行本代码示例之前，请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
# 或者 auth = oss2.Auth('your_ak','your_sk')
print(auth)
# yourEndpoint填写Bucket所在地域对应的Endpoint。
# 填写Bucket名称。
bucket = oss2.Bucket(auth, 'https://cloudfront.net', 'distribution_name')

步骤 4：测试和验证

在设备上更新端点和凭证后，我们进行了一系列测试，以验证 IoT 设备是否成功将文件上传到 S3 存储桶。一切都按预期进行，文件通过 CloudFront 到达 S3 存储桶，无需对设备代码进行任何更改。

需要注意的是，考虑到此过程中 AK、SK 为您管理，并非云服务商管控，所以存在更换 AK、SK 情形。因此，在 CloudFront Function 代码中，采用 CloudFront Key Value Pairs 存储多个 AK、SK。根据请求着签名中的 AK 来采用指定 SK。以防在更换 AK、SK 过程中造成系统中断。

总结

本文编写背景是亚马逊云科技帮助客户进行了友商对象存储迁移到 Amazon S3 的技术总结。这家公司比较特殊，有很多 IoT 设备，IoT 设备里使用友商 SDK 上传文件。我们在不修改代码、仅修改下发 Endpoint 配置和 AK SK 配置的情况下，让 IoT 设备把文件传到 Amazon S3 上。

这个 Amazon S3 前面我们加了一个 CloudFront，同时添加了 Lambda @Edge 进行请求转发和响应处理，通过 CloudFront 地址进行传输。同时通过 CloudFront Function 增加了 AK、SK 签名验证的安全限制。最大程度帮助我们的客户进行平滑迁移。

亚马逊AWS官方博客