如何创建 AWS chime presigned URL 来为前端生成网络套接字 url?

How to create AWS chime presigned URL to generate web socket url for the Front end?

我正在尝试生成预签名的 Web 套接字 URL 以获取前端 AWS 钟声的实时消息通知,如图所示 here . I planning to get this deployed as separate back end API using lambda. I followed exactly as shown here 但我收到“未授权”错误在前端连接到这个生成的URL。任何人都可以帮助我在 python 中生成这个预签名的 URL 需要做什么吗?我想我使用了错误的 host/service 铃声参数。

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