如何使用 IAM Auth 将 python websocket 连接到 AppSync 订阅？答案

【问题标题】：How to connect a python websocket to an AppSync subscription, with IAM Auth?如何使用 IAM Auth 将 python websocket 连接到 AppSync 订阅？
【发布时间】：2020-07-30 05:10:11
【问题描述】：

我希望 python 打开 AWS AppSync 订阅并接收更新。

官方博客对如何使用 IAM 对 websocket 进行身份验证一无所知。
https://aws.amazon.com/blogs/mobile/appsync-websockets-python/

它遵循this document about signing http requests，它没有提到 websockets，因此没有提供任何线索来包含、排除或改变什么来启动 websocket。我担心在发生任何事情之前我需要相当长的时间来消化、适应和测试。

IAM_header = {
    "accept": "application/json, text/javascript",
    "content-encoding": "amz-1.0",
    "content-type": "application/json; charset=UTF-8",
    "host": <HOST>,
    "x-amz-date": <ISO_UTC_TIMESTAMP>,,
    "X-Amz-Security-Token": <IAM SECURITY TOKEN>,
    "Authorization": <IAM SIGNATURE>
}

我想以前没有人这样做过并且可以分享一些见解或代码吗？

IAM_SIGNATURE
在创建 websocket 的上下文中，我们正在签名的“request_parameters”是什么？

到目前为止，我已将详细信息输入 requests_aws4auth 并将其中的标头填充到 websocket 中。这项工作收到了这样的回复：

消息

此请求的规范字符串应该是\n'POST\n/graphql/connect\n\nhost:iutysrbt2qxx7ekx2xi.appsync-api .AP-东南-2.amazonaws.com \ NX-AMZ-内容-SHA256：e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934852b855 \ NX-AMZ-日期：20200730T044539Z \ NX-AMZ-安全令牌：FwoGZpgV9hfLzGbDiPLCK1HMsXYJapqB299GReKYAALLKPOgifkFMi2VZsiycpLbxp7ZDGGO4NfSgEaULYm9GMMkGIj1Ck / 0YcckBNawWpisa6NxdMs = \ n \ n主机; X-amz- content-sha256;x-amz-date;x-amz-security-token\n44136fa355b3678a1b4fc21fe77e8310c060f61caaff8a'\n\nString-to-Sign应该是\n'AWS4-HMAC-SHA256\n20200730T044539Z \n20200730/ap-southeast-2/appsync/aws4_request\n29a5ecebf83602958c569f1ae18b5fa8353dd04833a833885'\n"}]},"type":"connection_error"}

所以我需要弄清楚“Canonical String”、“String-to-Sign”以及如何将它们填充到 Authorization 标头中。

【问题讨论】：

标签： python amazon-web-services websocket amazon-iam aws-appsync

【解决方案1】：

graphql-python/gql 自 version 3.0.0rc0 起支持 AWS AppSync。

它支持实时端点上的查询、变异甚至订阅。

它支持IAM、api key和JWT认证方式。

文档可通过here获取

以下是使用 IAM 身份验证的订阅示例：

import asyncio
import os
import sys

from gql import Client, gql
from gql.transport.appsync_websockets import AppSyncWebsocketsTransport

# Uncomment the following lines to enable debug output
# import logging
# logging.basicConfig(level=logging.DEBUG)


async def main():

    # Should look like:
    # https://XXXXXXXXXXXXXXXXXXXXXXXXXX.appsync-api.REGION.amazonaws.com/graphql
    url = os.environ.get("AWS_GRAPHQL_API_ENDPOINT")

    if url is None:
        print("Missing environment variables")
        sys.exit()

    transport = AppSyncWebsocketsTransport(url=url)

    async with Client(transport=transport) as session:

        subscription = gql(
            """
subscription onCreateMessage {
  onCreateMessage {
    message
  }
}
"""
        )

        print("Waiting for messages...")

        async for result in session.subscribe(subscription):
            print(result)


asyncio.run(main())

【讨论】：