用于生成预签名 URL 的示例 Python 代码 - Amazon IoT Core
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅 中国的 Amazon Web Services 服务入门 (PDF)

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

用于生成预签名 URL 的示例 Python 代码

以下代码显示了一个使用 Python 作为编程语言来生成预签名 URL 的示例。

先决条件

要使用 Python 编程语言来生成请求,您必须:

  • 已在您的计算机上安装 Python。您可以运行以下命令,也可以下载并运行 Python 安装程序

    sudo apt install python3
  • Python 请求库。您可以运行以下命令或下载请求库,该库在示例脚本中用于发出 Web 请求。

    pip install requests
  • 一个访问密钥,由环境变量中名为 AWS_ACCESS_KEY_IDAWS_SECRET_ACCESS_KEY 的访问密钥 ID 和秘密访问密钥构成。或者,您也可以在凭证文件中保存这些值,然后从这些文件中读取。

    注意

    作为最佳实践,我们建议您不要在代码中嵌入凭证。有关更多信息,请参阅《Amazon Account Management 参考指南》中的 Amazon 账户最佳实践

    $ export AWS_ACCESS_KEY_ID=My_Access_Key $ export AWS_SECRET_ACCESS_KEY=My_Secret_Key # Session token is required only if you use temporary access key starting with "ASIA" $ export AWS_SESSION_TOKEN=My_Session_token
  • 在您的账户中创建了网络分析器配置。要运行此脚本,请将网络分析器配置的名称指定为变量。此示例指的是名为 My_Network_Analyzer_Config 的配置。

    注意

    如果不指定配置名称,将使用默认配置 NetworkAnalyzerConfig_Default

示例 Python 代码

Python 代码生成预签名 URL,WebSocket库可以使用该网址向服务发送请求。此函数创建一个规范请求,再创建用于计算签名的待签名字符串,然后将签名添加到 HTTP 请求以创建预签名 URL。然后,您可以使用WebSocket库请求预签名的 URL。

要运行脚本 generate_presigned_url.py,请运行以下命令(如果要从脚本所在的同一路径运行该脚本)。

python generate_presigned_url.py

下面显示的是 generate_presigned_url.py 脚本的内容。

generate_presigned_url.py 的内容

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 # Version 4 signing example """ Sample Python code to generate the pre-signed URL. You can change the parameters in this code to your own values, such as the variables that are required for the request URL, the network analyzer configuration name, and Region. """ # ------------------------------------------------------------------ # Step 1. Import the required libraries and define the functions # sign and getSignatureKey that will be used to derive a signing key. # ------------------------------------------------------------------ import sys, os, base64, datetime, hashlib, hmac, urllib.pars import requests # pip install requests def sign(key, msg): return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest() def getSignatureKey(key, dateStamp, regionName, serviceName): kDate = sign(("AWS4" + key).encode("utf-8"), dateStamp) kRegion = sign(kDate, regionName) kService = sign(kRegion, serviceName) kSigning = sign(kService, "aws4_request") return kSigning # ------------------------------------------------------------------ # Step 2. Define the variables required for the request URL. Replace # values for the variables, such as region, with your own values. # ------------------------------------------------------------------ method = "GET" service = "iotwireless" region = "us-east-1" # Host and endpoint information. host = "api.iotwireless." + region + ".amazonaws.com" endpoint = "wss://" + host # Create a date for headers and the credential string. t = datetime.datetime.utcnow() amz_date = t.strftime("%Y%m%dT%H%M%SZ") # For date stamp, the date without time is used in credential scope. datestamp = t.strftime("%Y%m%d") # ----------------------------------------------------------------------- # Step 3. Create the canonical URI and canonical headers for the request. # ----------------------------------------------------------------------- canonical_uri = "/start-network-analyzer-stream" configuration_name = "My_Network_Analyzer_Config" canonical_headers = "host:" + host + "\n" signed_headers = "host" algorithm = "AWS4-HMAC-SHA256" credential_scope = datestamp + "/" + region + "/" + service + "/" + "aws4_request" # ----------------------------------------------------------------------- # Step 4. Read the credentials that are required for the request # from environment variables or configuration file. # ----------------------------------------------------------------------- # IMPORTANT: Best practice is NOT to embed credentials in code. access_key = os.environ.get("AWS_ACCESS_KEY_ID") secret_key = os.environ.get("AWS_SECRET_ACCESS_KEY") token = os.environ.get("AWS_SESSION_TOKEN") if access_key is None or secret_key is None: print("No access key is available.") sys.exit() if access_key.startswith("ASIA") and token is None: print("Detected temporary credentials. You must specify a token.") sys.exit() # ---------------------------------------------------------------------- # Step 5. Create the canonical query string. Query string values must be # URI-encoded and sorted by name. Query headers must in alphabetical order. # ---------------------------------------------------------------------- canonical_querystring = "X-Amz-Algorithm=" + algorithm canonical_querystring += "&X-Amz-Credential=" + \ urllib.parse.quote(access_key + "/" + credential_scope, safe="-_.~") canonical_querystring += "&X-Amz-Date=" + amz_date canonical_querystring += "&X-Amz-Expires=300" if access_key.startswith("ASIA"): # percent encode the token and double encode "=" canonical_querystring += "&X-Amz-Security-Token=" + \ urllib.parse.quote(token, safe="-_.~").replace("=", "%253D") canonical_querystring += "&X-Amz-SignedHeaders=" + signed_headers canonical_querystring += "&configuration-name=" + configuration_name # ---------------------------------------------------------------------- # Step 6. Create a hash of the payload. # ---------------------------------------------------------------------- payload_hash = hashlib.sha256(("").encode("utf-8")).hexdigest() # ------------------------------------------------------------------ # Step 7. Combine the elements, which includes the query string, the # headers, and the payload hash, to form the canonical request. # ------------------------------------------------------------------ canonical_request = method + "\n" + canonical_uri + "\n" + canonical_querystring \ + "\n" + canonical_headers + "\n" + signed_headers + "\n" + payload_hash # ---------------------------------------------------------------------- # Step 8. Create the metadata string to store the information required to # calculate the signature in the following step. # ---------------------------------------------------------------------- string_to_sign = algorithm + "\n" + amz_date + "\n" + \ credential_scope + "\n" + hashlib.sha256(canonical_request.encode("utf-8")).hexdigest() # ---------------------------------------------------------------------- # Step 9. Calculate the signature by using a signing key that"s obtained # from your secret key. # ---------------------------------------------------------------------- # Create the signing key from your secret key. signing_key = getSignatureKey(secret_key, datestamp, region, service) # Sign the string_to_sign using the signing key. signature = hmac.new(signing_key, (string_to_sign).encode("utf-8"), hashlib.sha256).hexdigest() # ---------------------------------------------------------------------- # Step 10. Create the request URL using the calculated signature and by # combining it with the canonical URI and the query string. # ---------------------------------------------------------------------- canonical_querystring += "&X-Amz-Signature=" + signature request_url = endpoint + canonical_uri + "?" + canonical_querystring print("\n-----------PRESIGNED URL-----------") print(request_url)

后续步骤

现在,您可以在WebSocket库中使用请求 URL 向服务发出请求并查看消息。

要安装用于 Python 的WebSocket库,请运行以下命令。有关如何在 Python 中使用WebSocket客户端的信息,请参阅带有低级 API 选项的 Python WebSocket 客户端。

pip install websocket-client

安装客户端并提出请求后,您将看到指示请求状态的消息和状态代码。有关更多信息,请参阅WebSocket消息和状态码