Background
I have a Lambda Function URL configured with AWS_IAM
authentication. The right resource policy is attached to the Lambda Function to allow the specified IAM role to invoke the function URL. This is validated by the fact that the role is able to successfully POST request to the function URL if there is no payload. For my use case, the payload is dynamic for every request made with the Lambda Function URL. My use case also has a constraint that the authorization has to be via request query parameters and not headers. I'm hoping there's a way to configure the AWS V4 signed URL so that the AWS-managed authorizer is able to authorize requests with dynamic payloads from valid IAM roles.
Problem
Once the request contains a payload of any sort, the request is rejected before reaching the Lambda Function and the response is:
{
"message": "The request signature we calculated does not match the signature you provided. Check your AWS Secret Access Key and signing method. Consult the service documentation for details."
}
Here's the code that's creating the signature:
import hmac
import hashlib
from collections import OrderedDict
from datetime import datetime
import urllib
import logging
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)
def create_sha256_sig(key:str, msg:str) -> str:
"""Returns SHA-256 signature using the provided key and message"""
return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
def create_sig_key(aws_secret_key, datestamp, region, service):
date_key = create_sha256_sig(('AWS4' + aws_secret_key).encode('utf-8'), datestamp)
region_key = create_sha256_sig(date_key, region)
service_key = create_sha256_sig(region_key, service)
return create_sha256_sig(service_key, 'aws4_request')
def create_aws_v4_sig(url, method, service, region, access_key, secret_key, security_token, payload="", unsign_payload=False):
o = urllib.parse.urlparse(url)
host = o.hostname
log.debug(f"Hostname: {host}")
endpoint = o.scheme + "://" + o.netloc + o.path
log.debug(f"Endpoint: {endpoint}")
t = datetime.utcnow()
amz_date = t.strftime('%Y%m%dT%H%M%SZ')
datestamp = t.strftime('%Y%m%d')
canonical_uri = '/'
log.debug(f"Security Token:\n{security_token}")
canonical_querystring = o.query
log.debug(f"Base query string: {canonical_querystring}")
credential_scope = datestamp + '/' + region + '/' + service + '/' + 'aws4_request'
if unsign_payload:
canonical_headers = 'host:' + host + '\n' + "x-amz-content-sha256:UNSIGNED-PAYLOAD" + "\n"
signed_headers = 'host;x-amz-content-sha256'
payload_hash = "UNSIGNED-PAYLOAD"
else:
canonical_headers = 'host:' + host + '\n'
signed_headers = 'host'
payload_hash = hashlib.sha256((payload).encode('utf-8')).hexdigest()
log.debug(f"Canonical headers:\n{canonical_headers}")
if canonical_querystring == "":
canonical_querystring += 'X-Amz-Algorithm=AWS4-HMAC-SHA256'
else:
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=30'
canonical_querystring += '&X-Amz-Security-Token=' + urllib.parse.quote_plus(security_token)
canonical_querystring += '&X-Amz-SignedHeaders=' + signed_headers
algorithm = 'AWS4-HMAC-SHA256'
canonical_request = method + '\n' + canonical_uri + '\n' + canonical_querystring + '\n' + canonical_headers + '\n' + signed_headers + '\n' + payload_hash
log.debug(f"Canonical request:\n{canonical_request}")
msg = algorithm + '\n' + amz_date + '\n' + credential_scope + '\n' + hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
sig_key = create_sig_key(secret_key, datestamp, region, service)
signature = hmac.new(sig_key, (msg).encode('utf-8'), hashlib.sha256).hexdigest()
canonical_querystring += '&X-Amz-Signature=' + signature
sign_url = endpoint + "?" + canonical_querystring
return sign_url
Here's a snippet for calling the function:
def test_request(tf_out):
signed_url = create_aws_v4_sig(
# Lambda functions URL without any transformations
tf_out["lambda_function_url"],
"POST",
"lambda",
os.environ["AWS_REGION"],
os.environ["AWS_ACCESS_KEY_ID"],
os.environ["AWS_SECRET_ACCESS_KEY"],
os.environ["AWS_SECURITY_TOKEN"],
unsign_payload=True
)
log.debug(f"Signed URL: {signed_url}")
log.info("Sending request")
response = requests.post(signed_url, data=json.dumps({"foo": "bar"}))
log.debug(f"Response: {response.json()}")
assert response.status_code == 200
Attempts:
- Using the AWS S3 Guide as a reference to see if Lambda shares the same authorization functionality, I put the literal string
UNSIGNED-PAYLOAD
at the bottom of the canonical request like so:
POST
/
X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=<mask>%2F20220719%2Fus-west-2%2Flambda%2Faws4_request&X-Amz-Date=20220719T012942Z&X-Amz-Expires=30&X-Amz-Security-Token=<mask>&X-Amz-SignedHeaders=host;x-amz-content-sha256
host:<host>.lambda-url.us-west-2.on.aws
host
UNSIGNED-PAYLOAD
Unfortunately the same response specified within the "Problem" section is returned
Hi Has your role lambda:invokefunction rights?
Hi @marshall-m, I'm having the same problem. Could you find a way to solve this?