AWS4-HMAC-SHA256 request

aws4_hmac_sha256.py

import urllib.parse
import hashlib
import hmac
from datetime import datetime


def sign_request(
    method: str,
    headers: dict,
    path: str,
    query: dict,
    payload: str,
    access_key: str,
    secret_key: str,
    region: str = "ap-northeast-2",
    service: str = "execute-api",
) -> dict[str, str]:
    """
    :param method: GET, POST, PUT, DELETE
    :param headers:
    :param path: /post/product, /favorite
    :param query:
        {"category": "it,etc", "last_post_datetime": "2023-02-10T15:00:00.000000"}
    :param payload:
    :param access_key:
    :param secret_key:
    :param region:
    :param service:
    :return: Authorization header
    """

    # 필수 헤드 x-amz-date 추가
    amz_date = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    headers.update({"x-amz-date": amz_date})

    payload_hash = hashlib.sha256(payload.encode("utf-8")).hexdigest()
    # payload가 있으면 x-amz-content-sha256 헤더 추가
    if len(payload) > 0:
        headers.update({"x-amz-content-sha256": payload_hash})

    canonical_querystring = get_canonical_querystring(query)

    # 필요한 헤더는 host, x-amz-date
    # key 기준으로 알파벳 정렬 해야함
    sorted_headers = sorted(
        [
            (k, v)
            for k, v in headers.items()
            if k in ["host", "x-amz-date", "x-amz-content-sha256"]
        ],
        key=lambda x: x[0],
    )
    canonical_headers = "\n".join(f"{k}:{v}" for k, v in sorted_headers) + "\n"
    signed_headers = ";".join(k for k, _ in sorted_headers)

    canonical_uri = path
    canonical_request = f"{method}\n{canonical_uri}\n{canonical_querystring}\n{canonical_headers}\n{signed_headers}\n{payload_hash}"

    algorithm = "AWS4-HMAC-SHA256"
    date_stamp = datetime.utcnow().strftime("%Y%m%d")
    credential_scope = f"{date_stamp}/{region}/{service}/aws4_request"
    string_to_sign = f'{algorithm}\n{amz_date}\n{credential_scope}\n{hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()}'

    signing_key = get_signature_key(secret_key, date_stamp, region, service)
    signature = hmac.new(
        signing_key, string_to_sign.encode("utf-8"), hashlib.sha256
    ).hexdigest()
    authorization_header = f"{algorithm} Credential={access_key}/{credential_scope}, SignedHeaders={signed_headers}, Signature={signature}"
    return {"Authorization": authorization_header, "x-amz-date": amz_date}


def get_canonical_querystring(query):
    if len(query) == 0:
        canonical_querystring = ""
    else:
        sorted_parameters = sorted(query.items(), key=lambda x: x[0])
        encoded_query_parameters = [
            urllib.parse.quote(x[0], safe="~")
            + "="
            + urllib.parse.quote(x[1], safe="~")
            for x in sorted_parameters
        ]
        canonical_querystring = "&".join(encoded_query_parameters)
    return canonical_querystring


def get_signature_key(key, date_stamp, regionName, serviceName):
    k_date = sign(("AWS4" + key).encode("utf-8"), date_stamp)
    k_region = sign(k_date, regionName)
    k_service = sign(k_region, serviceName)
    k_signing = sign(k_service, "aws4_request")
    return k_signing


def sign(key, msg):
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()