実用性は微妙です。

やったこと

IAMの認証情報をヘッダに格納してHTTPリクエスト。普通はSDKとか(GraphQLの場合はGraphQLクライアントが)よしなにやってくれるためあまり意識しないはずなんですが、署名がどうなっているか見たことなかったので。
内容は、署名計算に関してはほぼ公式ドキュメントのまんま。GraphQLのクエリをリクエストに入れるところとかが慣れず困ったので次回コピペできるようにメモ。

得たもの

IAM認証情報を署名に入れるときってこんな情報が入っていてこんな手順なのか。HTTPメソッドとかサービスとかも正しく指定しないと署名エラーが返ってくる。
リクエストボディはJSON

準備

AppSyncとスキーマを適当に用意する(IAM認証)。今回は疎通確認がメインなので内容は省略。適当にamplifyで作った。コンソールからmutationして一個だけデータ入ってる。
appsync:GraphQLの権限を持ったIAMユーザー。

クライアント環境

デフォルトのAmazon Linux 2。

$ python --version
Python 2.7.18
$ cat /etc/system-release
Amazon Linux release 2 (Karoo)

注意点

そもそも本当にこの記事の内容が必要かどうか。冒頭に書いた通り普通はSDKとかが認証は簡単にやってくれるし、ない場合でもNode.jsなら楽に署名が作れるSignerというものがあり、Lambdaから送るサンプルが公開されている。署名計算とかを肩代わりしてくれるライブラリとかを探す方が楽。
クライアントがEC2なのにIAMユーザーを使うのはバッドプラクティスです。やったことないですが、多分IAMロール+AWS SDKで一時的な認証情報が取得できるので、そちらを使う方が良いと思われる。
Pythonのバージョン(公式ドキュメント参照)

処理内容

import sys, os, base64, datetime, hashlib, hmac 
import requests
import json

method = 'POST'
region = 'ap-northeast-1'
endpoint = 'https://hogehogege.appsync-api.ap-northeast-1.amazonaws.com/graphql'
host = 'hogehogege.appsync-api.ap-northeast-1.amazonaws.com'
service = 'appsync'

AWS_ACCESS_KEY_ID = 'XXXXXXX'
AWS_SECRET_ACCESS_KEY = 'YYYYYYYY'

## 何かしらのGraphQLオペレーション
query = 'query MyQuery {'
query += 'listTodos {'
query +=  'nextToken '
query +=    'items {'
query +=      'name'
query +=    '}'
query +=  '}'
query += '}'

body = json.dumps({'query':query})
print(body)

## task3で使う署名用の関数定義
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

t = datetime.datetime.utcnow()
amzdate = t.strftime('%Y%m%dT%H%M%SZ')
datestamp = t.strftime('%Y%m%d')

# task1
canonical_uri = '/graphql' 
canonical_querystring = ''
canonical_headers = 'host:' + host + '\n' + 'x-amz-date:' + amzdate + '\n'

signed_headers = 'host;x-amz-date'

payload_hash = hashlib.sha256(body.encode('utf-8')).hexdigest()

canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash

# task2
algorithm = 'AWS4-HMAC-SHA256'
credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
string_to_sign = algorithm + '\n' +  amzdate + '\n' +  credential_scope + '\n' +  hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()

# task3
signing_key = getSignatureKey(AWS_SECRET_ACCESS_KEY, datestamp, region, service)
signature = hmac.new(signing_key, (string_to_sign).encode('utf-8'), hashlib.sha256).hexdigest()

# task4
authorization_header = algorithm + ' ' + 'Credential=' + AWS_ACCESS_KEY_ID + '/' + credential_scope + ', ' +  'SignedHeaders=' + signed_headers + ', ' + 'Signature=' + signature
print(authorization_header)
headers = {'x-amz-date':amzdate, 'Authorization':authorization_header}

# request
r = requests.post(endpoint, data=body, headers=headers)

print('\nRESPONSE++++++++++++++++++++++++++++++++++++')
print(r.content)

処理結果

成功する例


$ python call_appsync.py 
{"query": "query MyQuery {listTodos {nextToken items {name}}}"}
AWS4-HMAC-SHA256 Credential=accecc_key_id/20210307/ap-northeast-1/appsync/aws4_request, SignedHeaders=host;x-amz-date, Signature=piyopiyopiyo

RESPONSE++++++++++++++++++++++++++++++++++++
{"data":{"listTodos":{"nextToken":null,"items":[{"name":"first to do"}]}}}

雑に途中でbodyとAuthorizationヘッダが出力されています。

失敗する例

クエリをミスった場合(nameでなくてnamにした)。レスポンスではエラー内容見られるけど、CloudWatch Logsではエラーの詳細な内容わからないんですね(AppSync側でログを有効化&詳細なコンテンツを含める設定済み)。

(省略)
RESPONSE++++++++++++++++++++++++++++++++++++
{"data":null,"errors":[{"path":null,"locations":[{"line":1,"column":54,"sourceName":null}],"message":"Validation error of type FieldUndefined: Field 'nam' in type 'Todo' is undefined @ 'listTodos/items/nam'"}]}

CloudWatch Logsの内容抜粋。
クエリの内容はわかるが、ステータスが200になるよう。

GraphQL Query: query MyQuery {listTodos {nextToken items {nam}}}, Operation: null, Variables: {}
...
{
    "logType": "RequestSummary",
    "requestId": "zzzzzzzz",
    "graphQLAPIId": "lollollol",
    "statusCode": 200,
    "latency": 85699000
}
...
Response Headers: {Content-Type=application/json; charset=UTF-8}