k3awssign

A python lib for adding AWS signature version 4 to requests.

k3awssign is a component of pykit3 project: a python3 toolkit set.

Installation

pip install k3awssign

Quick Start

import k3awssign

access_key = 'your access key'
secret_key = 'your secret key'

signer = k3awssign.Signer(access_key, secret_key)

request = {
    'verb': 'PUT',
    'uri': '/test-bucket/test-key',
    'headers': {
        'Host': 'bscstorage.com',
        'Content-Length': 7,
    },
    'body': 'bla bla',
}

signer.add_auth(request, sign_payload=True)
# request['headers'] now contains AWS signature headers

API Reference

k3awssign

This lib is used to sign a request using aws signature version 4. You need to provide a python dict which represent your request(it typically contains verb, uri, args, headers, body), and your access key and your secret key. This lib will add signature to the request.

Signer

Bases: SigV4

Source code in k3awssign/awssign.py

class Signer(SigV4):
    def __init__(self, access_key, secret_key, region=None, service=None, default_expires=None):
        """

        :param access_key: the access key used to sign the request.
        :param secret_key: the secret key used to sign the request.
        :param region: the region name of the service, the default is 'us-east-1'.
        :param service: the service name, the default is 's3'.
        :param default_expires: the default expires time of a presigned url in seconds, the default is 60.
        """
        self._access_key = self._to_utf8(access_key)
        self._secret_key = self._to_utf8(secret_key)
        self._service = self._to_utf8(service) or "s3"
        self._region = self._to_utf8(region) or "us-east-1"
        self._default_expires = default_expires or 60

    def _get_request_date(self, request_date):
        if isinstance(request_date, (int, float)):
            dt = datetime.datetime.fromtimestamp(request_date, datetime.timezone.utc)
            request_date = dt.strftime(SIGV4_TIMESTAMP)
        elif isinstance(request_date, str):
            try:
                datetime.datetime.strptime(request_date, SIGV4_TIMESTAMP)
            except ValueError:
                raise RequestDateFormatError(
                    "request date: %s is not iso base formmat like 20161206T120102Z" % str(request_date)
                )
        else:
            datetime_now = datetime.datetime.now(datetime.timezone.utc)
            request_date = datetime_now.strftime(SIGV4_TIMESTAMP)

        return request_date

    def _clean_query_string(self, query_string):
        items = query_string.split("&")
        qs = []

        for item in items:
            arg_name = item.partition("=")[0]
            arg_name = self._unescape_plus(arg_name)

            if arg_name in VALID_AUTH_ARGS:
                continue

            qs.append(item)

        return "&".join(qs)

    def _clean_args(self, args):
        for arg_name in VALID_AUTH_ARGS:
            if arg_name in args:
                args.pop(arg_name)

        return args

    def _trimall(self, s):
        return " ".join(s.split())

    def _standardize_headers(self, headers, headers_not_to_sign):
        headers_not_to_sign_low = []
        for h_name in headers_not_to_sign:
            h_name = self._to_utf8(h_name)
            headers_not_to_sign_low.append(h_name.lower())

        stand_headers = {}

        for k, v in headers.items():
            k = self._to_utf8(k)
            v = self._to_utf8(v)

            low_name = k.lower().strip()
            stand_v = self._trimall(str(v))

            if low_name not in stand_headers:
                stand_headers[low_name] = []

            stand_headers[low_name].append(stand_v)

        for k, v in stand_headers.items():
            if len(v) == 1:
                stand_headers[k] = v[0]

        signed_header_names = []

        for h_name in stand_headers.keys():
            if h_name not in headers_not_to_sign_low:
                signed_header_names.append(str(h_name))

        signed_header_names.sort()
        return ";".join(signed_header_names), stand_headers

    def _validate_arg_value(self, arg_value):
        if arg_value is True or isinstance(arg_value, str):
            return

        if not isinstance(arg_value, list):
            raise InvalidRequestArgValueError("arg value: %s, must be string or list or True" % repr(arg_value))

        for value in arg_value:
            if not isinstance(value, str) and value is not True:
                raise InvalidRequestArgValueError("multi arg value: %s, must be string or True" % repr(value))

    def _validate_uri_and_args(self, uri, args):
        if not isinstance(uri, str) or not uri.startswith("/"):
            raise InvalidRequestUriError("uri: %s, must be a string and starts with /" % repr(uri))

        has_query_string = False
        if len(uri.split("?")) > 1:
            has_query_string = True
        if has_query_string and args is not None:
            raise InvalidRequestError("use both query string and args is not allowed")

        if args is None:
            return

        if not isinstance(args, dict):
            raise InvalidRequestArgError("args: %s, is not a dict" % repr(args))

        for arg_name, arg_value in args.items():
            if not isinstance(arg_name, str):
                raise InvalidRequestArgNameError("arg name: %s, is not a string" % repr(arg_name))

            self._validate_arg_value(arg_value)

    def _validate_headers(self, headers):
        if not isinstance(headers, dict):
            raise InvalidRequestHeadersError("headers: %s, is not a dict" % repr(headers))

        has_host = False
        for k, _ in headers.items():
            if not isinstance(k, str):
                raise InvalidRequestHeaerNameError("header name: %s, is not a string" % repr(k))

            if k.lower() == "host":
                has_host = True

        if has_host is not True:
            raise InvalidRequestHeadersError("absence of host header")

    def _validate_request(self, request):
        if not isinstance(request, dict):
            raise InvalidRequestError("request: %s, is not a dict" % repr(request))
        if not isinstance(request.get("verb"), str):
            raise InvalidRequestVerbError("absence of or invalid request verb")

        self._validate_uri_and_args(request.get("uri"), request.get("args"))

        self._validate_headers(request.get("headers"))

    def _modify_request_headers(self, request, query_auth, sign_payload, request_date):
        has_amz_date = False
        hashed_payload = None

        for k, v in list(request["headers"].items()):
            low_name = k.lower()

            if low_name == "authorization":
                request["headers"].pop(k)
            elif low_name == "x-amz-date":
                request["headers"].pop(k)
                has_amz_date = True
            elif low_name == "x-amz-content-sha256":
                hashed_payload = self._to_utf8(v)
                request["headers"].pop(k)

        if has_amz_date or not query_auth:
            request["headers"]["X-Amz-Date"] = request_date

        if hashed_payload is not None:
            request["headers"]["X-Amz-Content-SHA256"] = hashed_payload
            return hashed_payload

        if query_auth:
            return UNSIGNED_PAYLOAD

        if not sign_payload:
            request["headers"]["X-Amz-Content-SHA256"] = UNSIGNED_PAYLOAD
            return UNSIGNED_PAYLOAD

        if isinstance(request.get("body"), str) and len(request["body"]) > 0:
            hashed_payload = self._make_sha256(self._to_utf8(request["body"]))
        else:
            hashed_payload = EMPTY_PAYLOAD_HASH

        request["headers"]["X-Amz-Content-SHA256"] = hashed_payload
        return hashed_payload

    def _get_credential_scope(self, request_date, signing_date=None):
        credential_date = signing_date or request_date[:8]

        if not isinstance(credential_date, str):
            raise InvalidSigningDateError("invalid signing date: %s, not of string type" % repr(credential_date))

        credential_scope = "/".join([credential_date, self._region, self._service, CREDENTIALS_SUFFIX])

        return credential_scope

    def add_auth(self, request, **argkv):
        """
        The method used to sign a request.
        :param request: a python dict which used to represent your request.
        It may contents the following fields:
        verb:
        the request method, such as 'GET', 'PUT'. Required.
        uri: the url encoded uri, it can contains query string only when you did not specify `args` in `request`.
        Required.
        args: a python dict contains the request parameters, it should not be url encoded. You can not use both
        args and query string in `uri` at the same time.
        headers: a python dict contains request headers. It must contains the 'Host' header.
        body: a string contains the request payload. If you do not want to sign the payload or you have set
        'X-Amz-ContentSHA256' header in `headers`, you can omit this field.
        :param argkv: following keyword arguments are allowed.
        query_auth:
        set to `True` if you want to add the signature to the query string.
        The default is `False`, mean add the signature in the header.

        sign_payload:
        set to `True` if you want to sign the payload.
        The default is `False`.

        headers_not_to_sign:
        a list of header names, used to indicate which headers are
        not need to be signed. Optional.

        request_date:
        timestamp or a iso base format date string, used to specify
        a custom request date, instead of using current time as request date.
        Optional.

        expires:
        specify the signature expire time in seconds.
        It will overwrite the value of `default_expires`. Optional.

        signing_date:
        is a 8 digital date string like '20170131', used to specify a
        custom signing date. Optional.
        """
        self._validate_request(request)

        query_auth = argkv.get("query_auth", False)
        sign_payload = argkv.get("sign_payload", False)
        headers_not_to_sign = argkv.get("headers_not_to_sign", [])

        if not isinstance(headers_not_to_sign, list):
            raise InvalidHeadersNotToSignError("argument headers_not_to_sign is not a list")

        request_date = self._get_request_date(argkv.get("request_date"))

        credential_scope = self._get_credential_scope(request_date, self._to_utf8(argkv.get("signing_date")))

        credential = self._access_key + "/" + credential_scope

        hashed_payload = self._modify_request_headers(request, query_auth, sign_payload, request_date)

        signed_headers, stand_headers = self._standardize_headers(request["headers"], headers_not_to_sign)

        utf8_uri = self._to_utf8(request["uri"])
        origin_uri_path, delimiter, origin_query_string = utf8_uri.partition("?")

        if delimiter != "?":
            origin_query_string = None

        ctx = {
            "verb": self._to_utf8(request["verb"]),
            "uri": self._escape(self._unescape_plus(origin_uri_path), "/~"),
            "algorithm": ALGORITHM,
            "request_date": request_date,
            "credential_scope": credential_scope,
            "signed_headers": signed_headers,
            "hashed_payload": hashed_payload,
        }

        if origin_query_string is not None:
            cleaned_origin_query_string = self._clean_query_string(origin_query_string)
            args = self._query_string_to_args(cleaned_origin_query_string)
        else:
            args = self._clean_args(request.get("args", {}))
            query_string_from_args = self._args_to_query_string(args)

        auth_args = {}
        if query_auth:
            amz_expires = str(argkv.get("expires") or self._default_expires)
            amz_expires = self._to_utf8(amz_expires)
            auth_args = {
                "X-Amz-Algorithm": ALGORITHM,
                "X-Amz-Credential": credential,
                "X-Amz-Date": request_date,
                "X-Amz-Expires": amz_expires,
                "X-Amz-SignedHeaders": ctx["signed_headers"],
            }

            for k, v in auth_args.items():
                args[k] = v

        encoded_args = self._uri_encode_args(args)

        ctx["canonical_query_string"] = self._build_canonical_query_string(encoded_args)

        ctx["canonical_headers"] = self._build_canonical_headers(signed_headers, stand_headers)

        ctx["canonical_request"] = self._build_canonical_request(ctx)

        ctx["hashed_canonical_request"] = self._make_sha256(ctx["canonical_request"])

        ctx["string_to_sign"] = self._build_string_to_sign(ctx)

        ctx["signing_key"] = self._derive_signing_key(self._secret_key, credential_scope)

        ctx["signature"] = self._calc_signature(ctx["signing_key"], ctx["string_to_sign"])

        if origin_query_string is not None:
            qs = origin_query_string
        else:
            qs = query_string_from_args

        if query_auth:
            if len(qs) > 0:
                qs += "&"
            qs += self._args_to_query_string(auth_args)
            qs += "&X-Amz-Signature=" + ctx["signature"]

            request["uri"] = origin_uri_path + "?" + qs
        else:
            if len(qs) != 0:
                request["uri"] = origin_uri_path + "?" + qs
            else:
                request["uri"] = origin_uri_path

            auth_parts = ["AWS4-HMAC-SHA256 Credential=%s" % credential]
            auth_parts.append("SignedHeaders=%s" % ctx["signed_headers"])
            auth_parts.append("Signature=%s" % ctx["signature"])
            request["headers"]["Authorization"] = ", ".join(auth_parts)

        return ctx

    def add_post_auth(self, fields, **argkv):
        """
        :param fields: a python dict which contains form fields.

        It may contents the following fields:
        Policy: is python dict, describing what is permitted in the request.
        After this function call, it will be replaced by it's base64
        encoded version.

        key: the key of the object to upload.
        It also support some other fields, more infomation at
        [here](http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectPOST.html)
        This method will add some signature related fields to this dict.
        :param argkv: following keyword arguments are allowed.
        request_date:
        the same as in `add_auth`.
        signing_date:
        the same as in `add_auth`.
        """
        request_date = self._get_request_date(self._to_utf8(argkv.get("request_date")))

        credential_scope = self._get_credential_scope(request_date, self._to_utf8(argkv.get("signing_date")))

        credential = self._access_key + "/" + credential_scope

        ctx = {
            "algorithm": ALGORITHM,
            "request_date": request_date,
            "credential_scope": credential_scope,
            "policy": fields.get("Policy"),
        }

        policy_json_str = json.dumps(ctx["policy"])
        ctx["string_to_sign"] = base64.b64encode(policy_json_str.encode("utf-8"))

        ctx["signing_key"] = self._derive_signing_key(self._secret_key, credential_scope)

        ctx["signature"] = self._calc_signature(ctx["signing_key"], ctx["string_to_sign"])
        fields["Policy"] = ctx["string_to_sign"].decode()
        fields["X-Amz-Algorithm"] = ALGORITHM
        fields["X-Amz-Credential"] = credential
        fields["X-Amz-Date"] = request_date
        fields["X-Amz-Signature"] = ctx["signature"]

        return ctx

__init__(access_key, secret_key, region=None, service=None, default_expires=None)

:param access_key: the access key used to sign the request. :param secret_key: the secret key used to sign the request. :param region: the region name of the service, the default is 'us-east-1'. :param service: the service name, the default is 's3'. :param default_expires: the default expires time of a presigned url in seconds, the default is 60.

Source code in k3awssign/awssign.py

def __init__(self, access_key, secret_key, region=None, service=None, default_expires=None):
    """

    :param access_key: the access key used to sign the request.
    :param secret_key: the secret key used to sign the request.
    :param region: the region name of the service, the default is 'us-east-1'.
    :param service: the service name, the default is 's3'.
    :param default_expires: the default expires time of a presigned url in seconds, the default is 60.
    """
    self._access_key = self._to_utf8(access_key)
    self._secret_key = self._to_utf8(secret_key)
    self._service = self._to_utf8(service) or "s3"
    self._region = self._to_utf8(region) or "us-east-1"
    self._default_expires = default_expires or 60

add_auth(request, **argkv)

The method used to sign a request. :param request: a python dict which used to represent your request. It may contents the following fields: verb: the request method, such as 'GET', 'PUT'. Required. uri: the url encoded uri, it can contains query string only when you did not specify args in request. Required. args: a python dict contains the request parameters, it should not be url encoded. You can not use both args and query string in uri at the same time. headers: a python dict contains request headers. It must contains the 'Host' header. body: a string contains the request payload. If you do not want to sign the payload or you have set 'X-Amz-ContentSHA256' header in headers, you can omit this field. :param argkv: following keyword arguments are allowed. query_auth: set to True if you want to add the signature to the query string. The default is False, mean add the signature in the header.

sign_payload: set to True if you want to sign the payload. The default is False.

headers_not_to_sign: a list of header names, used to indicate which headers are not need to be signed. Optional.

request_date: timestamp or a iso base format date string, used to specify a custom request date, instead of using current time as request date. Optional.

expires: specify the signature expire time in seconds. It will overwrite the value of default_expires. Optional.

signing_date: is a 8 digital date string like '20170131', used to specify a custom signing date. Optional.

Source code in k3awssign/awssign.py

def add_auth(self, request, **argkv):
    """
    The method used to sign a request.
    :param request: a python dict which used to represent your request.
    It may contents the following fields:
    verb:
    the request method, such as 'GET', 'PUT'. Required.
    uri: the url encoded uri, it can contains query string only when you did not specify `args` in `request`.
    Required.
    args: a python dict contains the request parameters, it should not be url encoded. You can not use both
    args and query string in `uri` at the same time.
    headers: a python dict contains request headers. It must contains the 'Host' header.
    body: a string contains the request payload. If you do not want to sign the payload or you have set
    'X-Amz-ContentSHA256' header in `headers`, you can omit this field.
    :param argkv: following keyword arguments are allowed.
    query_auth:
    set to `True` if you want to add the signature to the query string.
    The default is `False`, mean add the signature in the header.

    sign_payload:
    set to `True` if you want to sign the payload.
    The default is `False`.

    headers_not_to_sign:
    a list of header names, used to indicate which headers are
    not need to be signed. Optional.

    request_date:
    timestamp or a iso base format date string, used to specify
    a custom request date, instead of using current time as request date.
    Optional.

    expires:
    specify the signature expire time in seconds.
    It will overwrite the value of `default_expires`. Optional.

    signing_date:
    is a 8 digital date string like '20170131', used to specify a
    custom signing date. Optional.
    """
    self._validate_request(request)

    query_auth = argkv.get("query_auth", False)
    sign_payload = argkv.get("sign_payload", False)
    headers_not_to_sign = argkv.get("headers_not_to_sign", [])

    if not isinstance(headers_not_to_sign, list):
        raise InvalidHeadersNotToSignError("argument headers_not_to_sign is not a list")

    request_date = self._get_request_date(argkv.get("request_date"))

    credential_scope = self._get_credential_scope(request_date, self._to_utf8(argkv.get("signing_date")))

    credential = self._access_key + "/" + credential_scope

    hashed_payload = self._modify_request_headers(request, query_auth, sign_payload, request_date)

    signed_headers, stand_headers = self._standardize_headers(request["headers"], headers_not_to_sign)

    utf8_uri = self._to_utf8(request["uri"])
    origin_uri_path, delimiter, origin_query_string = utf8_uri.partition("?")

    if delimiter != "?":
        origin_query_string = None

    ctx = {
        "verb": self._to_utf8(request["verb"]),
        "uri": self._escape(self._unescape_plus(origin_uri_path), "/~"),
        "algorithm": ALGORITHM,
        "request_date": request_date,
        "credential_scope": credential_scope,
        "signed_headers": signed_headers,
        "hashed_payload": hashed_payload,
    }

    if origin_query_string is not None:
        cleaned_origin_query_string = self._clean_query_string(origin_query_string)
        args = self._query_string_to_args(cleaned_origin_query_string)
    else:
        args = self._clean_args(request.get("args", {}))
        query_string_from_args = self._args_to_query_string(args)

    auth_args = {}
    if query_auth:
        amz_expires = str(argkv.get("expires") or self._default_expires)
        amz_expires = self._to_utf8(amz_expires)
        auth_args = {
            "X-Amz-Algorithm": ALGORITHM,
            "X-Amz-Credential": credential,
            "X-Amz-Date": request_date,
            "X-Amz-Expires": amz_expires,
            "X-Amz-SignedHeaders": ctx["signed_headers"],
        }

        for k, v in auth_args.items():
            args[k] = v

    encoded_args = self._uri_encode_args(args)

    ctx["canonical_query_string"] = self._build_canonical_query_string(encoded_args)

    ctx["canonical_headers"] = self._build_canonical_headers(signed_headers, stand_headers)

    ctx["canonical_request"] = self._build_canonical_request(ctx)

    ctx["hashed_canonical_request"] = self._make_sha256(ctx["canonical_request"])

    ctx["string_to_sign"] = self._build_string_to_sign(ctx)

    ctx["signing_key"] = self._derive_signing_key(self._secret_key, credential_scope)

    ctx["signature"] = self._calc_signature(ctx["signing_key"], ctx["string_to_sign"])

    if origin_query_string is not None:
        qs = origin_query_string
    else:
        qs = query_string_from_args

    if query_auth:
        if len(qs) > 0:
            qs += "&"
        qs += self._args_to_query_string(auth_args)
        qs += "&X-Amz-Signature=" + ctx["signature"]

        request["uri"] = origin_uri_path + "?" + qs
    else:
        if len(qs) != 0:
            request["uri"] = origin_uri_path + "?" + qs
        else:
            request["uri"] = origin_uri_path

        auth_parts = ["AWS4-HMAC-SHA256 Credential=%s" % credential]
        auth_parts.append("SignedHeaders=%s" % ctx["signed_headers"])
        auth_parts.append("Signature=%s" % ctx["signature"])
        request["headers"]["Authorization"] = ", ".join(auth_parts)

    return ctx

add_post_auth(fields, **argkv)

:param fields: a python dict which contains form fields.

It may contents the following fields: Policy: is python dict, describing what is permitted in the request. After this function call, it will be replaced by it's base64 encoded version.

key: the key of the object to upload. It also support some other fields, more infomation at here This method will add some signature related fields to this dict. :param argkv: following keyword arguments are allowed. request_date: the same as in add_auth. signing_date: the same as in add_auth.

Source code in k3awssign/awssign.py

def add_post_auth(self, fields, **argkv):
    """
    :param fields: a python dict which contains form fields.

    It may contents the following fields:
    Policy: is python dict, describing what is permitted in the request.
    After this function call, it will be replaced by it's base64
    encoded version.

    key: the key of the object to upload.
    It also support some other fields, more infomation at
    [here](http://docs.aws.amazon.com/AmazonS3/latest/API/RESTObjectPOST.html)
    This method will add some signature related fields to this dict.
    :param argkv: following keyword arguments are allowed.
    request_date:
    the same as in `add_auth`.
    signing_date:
    the same as in `add_auth`.
    """
    request_date = self._get_request_date(self._to_utf8(argkv.get("request_date")))

    credential_scope = self._get_credential_scope(request_date, self._to_utf8(argkv.get("signing_date")))

    credential = self._access_key + "/" + credential_scope

    ctx = {
        "algorithm": ALGORITHM,
        "request_date": request_date,
        "credential_scope": credential_scope,
        "policy": fields.get("Policy"),
    }

    policy_json_str = json.dumps(ctx["policy"])
    ctx["string_to_sign"] = base64.b64encode(policy_json_str.encode("utf-8"))

    ctx["signing_key"] = self._derive_signing_key(self._secret_key, credential_scope)

    ctx["signature"] = self._calc_signature(ctx["signing_key"], ctx["string_to_sign"])
    fields["Policy"] = ctx["string_to_sign"].decode()
    fields["X-Amz-Algorithm"] = ALGORITHM
    fields["X-Amz-Credential"] = credential
    fields["X-Amz-Date"] = request_date
    fields["X-Amz-Signature"] = ctx["signature"]

    return ctx

License

The MIT License (MIT) - Copyright (c) 2015 Zhang Yanpo (张炎泼)