问题描述
我正在尝试生成一个预签名的 Web 套接字 URL，以获取前端中 AWS chime 的实时消息通知，如图 here 所示。我计划使用 lambda 将其部署为单独的后端 API。我完全按照 here 所示进行操作，但是在前端连接到此生成的 URL 时出现“未经授权”错误。任何人都可以帮助我解决在 python 中生成这个预先签名的 URL 需要做些什么吗？我想我对 chime 使用了错误的主机/服务参数。
import json
import boto3
import urllib.parse
import requests
import uuid
import datetime
import sys,os,base64,datetime,hashlib,hmac
import os
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
def lambda_handler(event,context):
method = 'GET'
algorithm = 'AWS4-HMAC-SHA256'
service = 'chime'
host = client.get_messaging_session_endpoint()['Endpoint']['Url']
region = 'us-east-1'
#Getting the messaging endpoint using boto3
client = boto3.client('chime',region_name='us-east-1')
endpoint='wss://'+client.get_messaging_session_endpoint()['Endpoint']['Url']
user_id=event['querystringparameters'].get('userId')
session_id=event['querystringparameters'].get('sessionId')
user_id_arn=f'arn:aws:chime:us-east-1:11******:app-instance/03457-*****-412345-b3e4-123444/user/{user_id}'
access_key=os.environ['access_key']
secret_key=os.environ['secret_key']
#Following the steps as shown in the AWS documentation https://docs.aws.amazon.com/general/latest/gr/sigv4-signed-request-examples.html
t = datetime.datetime.utcNow()
amz_date = t.strftime('%Y%m%dT%H%M%sZ')
datestamp = t.strftime('%Y%m%d')
credential_scope=urllib.parse.quote(f'{access_key}/{datestamp}/us-east-1/chime/aws4_request',safe='')
canonical_uri = '/connect'
payload_hash = hashlib.sha256(('').encode('utf-8')).hexdigest()
canonical_headers = 'host:' + host + '\n'
signed_headers = 'host'
credential_scope=urllib.parse.quote(f'{access_key}/{datestamp}/us-east-1/chime/aws4_request',safe='')
canonical_querystring=''
canonical_querystring+='?X-Amz-Algorithm=AWS4-HMAC-SHA256'
canonical_querystring+=f'&X-Amz-Credential={credential_scope}'
canonical_querystring += '&X-Amz-Date=' + amz_date
canonical_querystring += '&X-Amz-SignedHeaders=' + signed_headers
canonical_querystring += '&X-Amz-Expires=3600'
canonical_querystring += '&sessionId=' + session_id
canonical_querystring += '&userArn=' + user_id_arn
canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
hashed_canonical_request=hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
string_to_sign = algorithm + '\n' + amz_date + '\n' + credential_scope + '\n' + hashed_canonical_request
signing_key = getSignatureKey(secret_key,datestamp,region,service)
signature = hmac.new(signing_key,(string_to_sign).encode("utf-8"),hashlib.sha256).hexdigest()
canonical_querystring += '&X-Amz-Signature=' + signature
request_url = endpoint + canonical_uri+canonical_querystring
return_dict={'wssUrl':request_url}
return return dict
解决方法
任何想知道问题是什么的人 - 我们必须将 user_id_arn 传递为 arn%3Aaws%3Achime%3Aus-east-1%3A123456789012%3Aapp-instance%2f5abcdefg-cc50-4a70-a88e-fd07351d3c2a%2Fuser%2Fcustom-user-id 而不是 arn:aws:chime:us-east-1:123456789012:app-instance/f5abcdefg-cc50-4a70-a88e-fd07351d3c2a/user/custom-user-id
完整的工作代码：
import json
import boto3
import urllib.parse
import requests
import uuid
import datetime
import sys,os,base64,datetime,hashlib,hmac
import os
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
def handler():
# TODO: Replace with your info
session_id= # <session_id>
user_id_arn= # <user_id_arn>
access_key= # <access_key>
secret_key= # <secret_key>
# Following the steps as shown in the AWS documentation https://docs.aws.amazon.com/general/latest/gr/sigv4-signed-request-examples.html
t = datetime.datetime.utcnow()
amz_date = t.strftime('%Y%m%dT%H%M%SZ')
datestamp = t.strftime('%Y%m%d')
# Getting the messaging endpoint using boto3
client = boto3.client('chime',region_name='us-east-1')
hostname = client.get_messaging_session_endpoint()['Endpoint']['Url']
method = 'GET'
service = 'chime'
region = 'us-east-1'
canonical_uri = '/connect'
canonical_headers = 'host:' + hostname + '\n'
signed_headers = 'host'
algorithm = 'AWS4-HMAC-SHA256'
credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
canonical_querystring = ''
canonical_querystring += 'X-Amz-Algorithm=AWS4-HMAC-SHA256'
canonical_querystring += '&X-Amz-Credential=' + urllib.parse.quote_plus(access_key + '/' + credential_scope)
canonical_querystring += '&X-Amz-Date=' + amz_date
canonical_querystring += '&X-Amz-Expires=3600'
canonical_querystring += '&X-Amz-Security-Token=' + urllib.parse.quote(session_token,safe='')
canonical_querystring += '&X-Amz-SignedHeaders=' + signed_headers
canonical_querystring += '&sessionId=' + urllib.parse.quote(session_id,safe='')
canonical_querystring += '&userArn=' + urllib.parse.quote(user_id_arn,safe='')
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
string_to_sign = algorithm + '\n' + amz_date + '\n' + credential_scope + '\n' + hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
signing_key = getSignatureKey(secret_key,datestamp,region,service)
signature = hmac.new(signing_key,(string_to_sign).encode('utf-8'),hashlib.sha256).hexdigest()
canonical_querystring += '&X-Amz-Signature=' + signature
request_url = 'wss://' + hostname + canonical_uri + '?' + canonical_querystring
return_dict={'wssUrl':request_url}
return return_dict