By using AWS re:Post, you agree to the Terms of Use

POST request to presigned AWS Lambda Function URL with unsigned payload

0

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:

  1. 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

1 Answers
0

Hi First your role needs lambda invoke function permissions

Explaining the sigv4 usage https://docs.aws.amazon.com/general/latest/gr/sigv4-signed-request-examples.html#sig-v4-examples-post

answered a month ago
  • Hi Omid,

    Yes, the role that is used for creating the sigv4 does have lambda:invokefunction permissions. The role actually has power user access for the AWS account so I don't think the issue is in regards to the IAM permissions of the role.

You are not logged in. Log in to post an answer.

A good answer clearly answers the question and provides constructive feedback and encourages professional growth in the question asker.

Guidelines for Answering Questions