in AWSIoTPythonSDK/core/protocol/connection/cores.py [0:0]
def createWebsocketEndpoint(self, host, port, region, method, awsServiceName, path):
# Return the endpoint as unicode string in 3.x
# Gather all the facts
amazonDate = self._createAmazonDate()
amazonDateSimple = amazonDate[0] # Unicode in 3.x
amazonDateComplex = amazonDate[1] # Unicode in 3.x
allKeys = self._checkIAMCredentials() # Unicode in 3.x
if not self._hasCredentialsNecessaryForWebsocket(allKeys):
raise wssNoKeyInEnvironmentError()
else:
# Because of self._hasCredentialsNecessaryForWebsocket(...), keyID and secretKey should not be None from here
keyID = allKeys["aws_access_key_id"]
secretKey = allKeys["aws_secret_access_key"]
# amazonDateSimple and amazonDateComplex are guaranteed not to be None
queryParameters = "X-Amz-Algorithm=AWS4-HMAC-SHA256" + \
"&X-Amz-Credential=" + keyID + "%2F" + amazonDateSimple + "%2F" + region + "%2F" + awsServiceName + "%2Faws4_request" + \
"&X-Amz-Date=" + amazonDateComplex + \
"&X-Amz-Expires=86400" + \
"&X-Amz-SignedHeaders=host" # Unicode in 3.x
hashedPayload = hashlib.sha256(str("").encode('utf-8')).hexdigest() # Unicode in 3.x
# Create the string to sign
signedHeaders = "host"
canonicalHeaders = "host:" + host + "\n"
canonicalRequest = method + "\n" + path + "\n" + queryParameters + "\n" + canonicalHeaders + "\n" + signedHeaders + "\n" + hashedPayload # Unicode in 3.x
hashedCanonicalRequest = hashlib.sha256(str(canonicalRequest).encode('utf-8')).hexdigest() # Unicoede in 3.x
stringToSign = "AWS4-HMAC-SHA256\n" + amazonDateComplex + "\n" + amazonDateSimple + "/" + region + "/" + awsServiceName + "/aws4_request\n" + hashedCanonicalRequest # Unicode in 3.x
# Sign it
signingKey = self._getSignatureKey(secretKey, amazonDateSimple, region, awsServiceName)
signature = hmac.new(signingKey, (stringToSign).encode("utf-8"), hashlib.sha256).hexdigest()
# generate url
url = "wss://" + host + ":" + str(port) + path + '?' + queryParameters + "&X-Amz-Signature=" + signature
# See if we have STS token, if we do, add it
awsSessionTokenCandidate = allKeys.get("aws_session_token")
if awsSessionTokenCandidate is not None and len(awsSessionTokenCandidate) != 0:
aws_session_token = allKeys["aws_session_token"]
url += "&X-Amz-Security-Token=" + quote(aws_session_token.encode("utf-8")) # Unicode in 3.x
self._logger.debug("createWebsocketEndpoint: Websocket URL: " + url)
return url