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

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

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

先决条件

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

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

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

    pip install requests
  • 一个 WebSocket 库,在使用脚本生成请求后,可使用此库请求预签名 URL。要安装此库,请运行以下命令。有关如何将 WebSocket 客户端与 python 结合使用的信息,请参阅带低级别 API 选项的适用于 Python 的 WebSocket 客户端

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

    注意

    作为最佳实践,我们建议您不要在代码中嵌入凭证。有关更多信息,请参阅亚马逊云科技一般参考中的管理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

示例 python 代码

python 代码生成预签名 URL,WebSocket 库可使用该 URL 向服务发送请求。此函数创建一个规范请求,再创建用于计算签名的待签名字符串,然后将签名添加到 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 # Amazon 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 Amazon Web Services 区域. """ # ------------------------------------------------------------------ # 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 Amazon 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 and combine the elements to form # the canonical request. # ---------------------------------------------------------------------- payload_hash = hashlib.sha256(("").encode("utf-8")).hexdigest() canonical_request = method + "\n" + canonical_uri + "\n" + canonical_querystring \ + "\n" + canonical_headers + "\n" + signed_headers + "\n" + payload_hash # ---------------------------------------------------------------------- # Step 9. 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 10. Calculate the signature by using a signing key that's obtained # from your secret key. # ---------------------------------------------------------------------- # Create the signing key from your Amazon 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 11. 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 向服务发出请求并观察消息。有关更多信息,请参阅WebSocket 消息和状态码