I am trying to generate a pre-signed web socket URL to get real time messaging notification for the AWS chime in the frontend as shown here . I planning to get this deployed as separate back end API using lambda. I followed exactly as shown here but I am getting "unauthorized" error when connecting to this generated URL in the front end. Can any one help me with what needs to be done to generate this pre-signed URL in python? I think I am using the wrong host/service parameters for 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
Any one who is wondering what the issue was- we have to pass the user_id_arn as arn%3Aaws%3Achime%3Aus-east-1%3A123456789012%3Aapp-instance%2f5abcdefg-cc50-4a70-a88e-fd07351d3c2a%2Fuser%2Fcustom-user-id instead of arn:aws:chime:us-east-1:123456789012:app-instance/f5abcdefg-cc50-4a70-a88e-fd07351d3c2a/user/custom-user-id
Full working code:
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