0

我正在尝试生成一个预签名的 Web 套接字 URL，以获取前端中 AWS chime 的实时消息通知，如此处所示。我计划使用 lambda 将其部署为单独的后端 API。我完全按照此处所示操作，但在前端连接到此生成的 URL 时出现“未经授权”错误。任何人都可以帮助我在 python 中生成这个预签名的 URL 需要做什么？我想我使用了错误的主机/服务参数来进行铃声。

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
4

1 回答 1

0

任何想知道问题出在哪里的人——我们必须传递 user_id_arnarn%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
于 2021-05-02T17:55:44.913 回答