From 7ec73c3affbb9e21bcc9ff688c1510d35ab51cd3 Mon Sep 17 00:00:00 2001 From: Nate Prewitt Date: Tue, 9 Dec 2025 17:20:33 -0700 Subject: [PATCH] Migrate examples directory from original project --- packages/aws-sdk-signers/examples/README.md | 165 ++++++++++++++++++ packages/aws-sdk-signers/examples/__init__.py | 2 + .../examples/aiohttp_signer.py | 71 ++++++++ .../aws-sdk-signers/examples/curl_signer.py | 68 ++++++++ .../examples/requests_signer.py | 77 ++++++++ 5 files changed, 383 insertions(+) create mode 100644 packages/aws-sdk-signers/examples/README.md create mode 100644 packages/aws-sdk-signers/examples/__init__.py create mode 100644 packages/aws-sdk-signers/examples/aiohttp_signer.py create mode 100644 packages/aws-sdk-signers/examples/curl_signer.py create mode 100644 packages/aws-sdk-signers/examples/requests_signer.py diff --git a/packages/aws-sdk-signers/examples/README.md b/packages/aws-sdk-signers/examples/README.md new file mode 100644 index 000000000..f6df76ca7 --- /dev/null +++ b/packages/aws-sdk-signers/examples/README.md @@ -0,0 +1,165 @@ +# Example Signers for aws-sdk-signers + +## Requests +We utilize the `AuthBase` construct provided by Requests to apply our signature +to each request. Our `SigV4Auth` class takes two arguments +[`SigV4SigningProperties`](https://github.com/smithy-lang/smithy-python/blob/9c0225b2810b3f68a84aa074e9b4e728a3043721/packages/aws-sdk-signers/src/aws_sdk_signers/signers.py#L44-L50) +and an [`AWSCredentialIdentity`](https://github.com/smithy-lang/smithy-python/blob/9c0225b2810b3f68a84aa074e9b4e728a3043721/packages/aws-sdk-signers/src/aws_sdk_signers/_identity.py#L10-L15). +These will be used across requests as "immutable" input. This is currently an +intentional design decision to work with Requests auth design. We'd love to +hear feedback on how you feel about the current approach, we recommend checking +the AIOHTTP section below for an alternative design. + +### Requests Sample +```python +from os import environ + +import requests + +from examples import requests_signer +from aws_sdk_signers import SigV4SigningProperties, AWSCredentialIdentity + +SERVICE="lambda" +REGION="us-west-2" + +# A GET request to this URL performs a "ListFunctions" invocation. +# Full API documentation can be found here: +# https://docs.aws.amazon.com/lambda/latest/api/API_ListFunctions.html +URL='https://lambda.us-west-2.amazonaws.com/2015-03-31/functions/' + +def get_credentials_from_env(): + """You will need to pull credentials from some source to use the signer. + This will auto-populate an AWSCredentialIdentity when credentials are + available through the env. + + You may also consider using another SDK to assume a role or pull + credentials from another source. + """ + return AWSCredentialIdentity( + access_key_id=environ["AWS_ACCESS_KEY_ID"], + secret_access_key=environ["AWS_SECRET_ACCESS_KEY"], + session_token=environ.get("AWS_SESSION_TOKEN"), + ) + +# Set up our properties and identity +identity = get_credentials_from_env() +properties = SigV4SigningProperties(region=REGION, service=SERVICE) + +# Configure the auth class for signing +sigv4_auth = requests_signer.SigV4Auth(properties, identity) + +r = requests.get(URL, auth=sigv4_auth) +``` + +## AIOHTTP +For AIOHTTP, we don't have a concept of a Request object, or option to subclass an +existing auth mechanism. Instead, we'll take parameters you normally pass to a Session +method and use them to generate signing headers before passing them on to AIOHTTP. + +This signer will be configured the same way as Requests and provides an Async signing +interface to be used alongside AIOHTTP. This is still a work in progress and will likely +have some amount of iteration to improve performance and ergonomics as we collect feedback. + +### AIOHTTP Sample +```python +import asyncio +from collections.abc import AsyncIterable, Mapping +from os import environ + +import aiohttp + +from examples import aiohttp_signer +from aws_sdk_signers import SigV4SigningProperties, AWSCredentialIdentity + + +SERVICE="lambda" +REGION="us-west-2" + +# A GET request to this URL performs a "ListFunctions" invocation. +# Full API documentation can be found here: +# https://docs.aws.amazon.com/lambda/latest/api/API_ListFunctions.html +URL='https://lambda.us-west-2.amazonaws.com/2015-03-31/functions/' + +def get_credentials_from_env(): + """You will need to pull credentials from some source to use the signer. + This will auto-populate an AWSCredentialIdentity when credentials are + available through the env. + + You may also consider using another SDK to assume a role or pull + credentials from another source. + """ + return AWSCredentialIdentity( + access_key_id=environ["AWS_ACCESS_KEY_ID"], + secret_access_key=environ["AWS_SECRET_ACCESS_KEY"], + session_token=environ.get("AWS_SESSION_TOKEN"), + ) + +# Set up our signing_properties and identity +identity = get_credentials_from_env() +properties = SigV4SigningProperties(region=REGION, service=SERVICE) + +signer = aiohttp_signer.SigV4Signer(properties, identity) + +async def make_request( + method: str, + url: str, + headers: Mapping[str, str], + body: AsyncIterable[bytes] | None, +) -> None: + # For more robust applications, you'll likely want to reuse this session. + async with aiohttp.ClientSession() as session: + signing_headers = await signer.generate_signature(method, url, headers, body) + headers.update(signing_headers) + async with session.request(method, url, headers=headers, data=body) as response: + print("Status:", response.status) + print("Content-Type:", response.headers['content-type']) + + body_content = await response.text() + print(body_content) + +asyncio.run(make_request("GET", URL, {}, None)) +``` + +## Curl Signer +For curl, we're generating a string to be used in a terminal or invoked subprocess. +This currently only supports known arguments like defining the method, headers, +and a request body. We can expand this to support arbitrary curl arguments in +a future version if there's demand. + +### Curl Sample +```python +from examples import curl_signer +from aws_sdk_signers import SigV4SigningProperties, AWSCredentialIdentity + +from os import environ + + +SERVICE="lambda" +REGION="us-west-2" + +# A GET request to this URL performs a "ListFunctions" invocation. +# Full API documentation can be found here: +# https://docs.aws.amazon.com/lambda/latest/api/API_ListFunctions.html +URL='https://lambda.us-west-2.amazonaws.com/2015-03-31/functions/' + + +properties = SigV4SigningProperties(region=REGION, service=SERVICE) +identity = AWSCredentialIdentity( + access_key_id=environ["AWS_ACCESS_KEY_ID"], + secret_access_key=environ["AWS_SECRET_ACCESS_KEY"], + session_token=environ["AWS_SESSION_TOKEN"] +) + +# Our curl signer doesn't need state so we +# can call classmethods directly on the signer. +signer = curl_signer.SigV4Curl +curl_cmd = signer.generate_signed_curl_cmd( + properties=properties, + identity=identity, + method="GET", + url=URL, + headers={}, + body=None, +) +print(curl_cmd) +``` diff --git a/packages/aws-sdk-signers/examples/__init__.py b/packages/aws-sdk-signers/examples/__init__.py new file mode 100644 index 000000000..04f8b7b76 --- /dev/null +++ b/packages/aws-sdk-signers/examples/__init__.py @@ -0,0 +1,2 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 diff --git a/packages/aws-sdk-signers/examples/aiohttp_signer.py b/packages/aws-sdk-signers/examples/aiohttp_signer.py new file mode 100644 index 000000000..b0a603e04 --- /dev/null +++ b/packages/aws-sdk-signers/examples/aiohttp_signer.py @@ -0,0 +1,71 @@ +""" +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +SPDX-License-Identifier: Apache-2.0 + +Sample signer using aiohttp. +""" + +import typing +from collections.abc import Mapping +from urllib.parse import urlparse + +from aws_sdk_signers import AsyncSigV4Signer, AWSRequest, Field, Fields, URI + +if typing.TYPE_CHECKING: + from aws_sdk_signers import AWSCredentialIdentity, SigV4SigningProperties + +SIGNING_HEADERS = ( + "Authorization", + "Date", + "X-Amz-Date", + "X-Amz-Security-Token", + "X-Amz-Content-SHA256", +) + + +class SigV4Signer: + """Minimal Signer implementation to be used with AIOHTTP.""" + + def __init__( + self, + properties: "SigV4SigningProperties", + identity: "AWSCredentialIdentity", + ): + self._properties = properties + self._identity = identity + self._signer = AsyncSigV4Signer() + + async def generate_signature( + self, + method: str, + url: str, + headers: Mapping[str, str], + body: typing.AsyncIterable[bytes] | None, + ) -> Mapping[str, str]: + """Generate signature headers for applying to request.""" + url_parts = urlparse(url) + uri = URI( + scheme=url_parts.scheme, + host=url_parts.hostname, + port=url_parts.port, + path=url_parts.path, + query=url_parts.query, + fragment=url_parts.fragment, + ) + fields = Fields([Field(name=k, values=[v]) for k, v in headers.items()]) + awsrequest = AWSRequest( + destination=uri, + method=method, + body=body, + fields=fields, + ) + signed_request = await self._signer.sign( + properties=self._properties, + request=awsrequest, + identity=self._identity, + ) + return { + header: signed_request.fields[header].as_string() + for header in SIGNING_HEADERS + if header in signed_request.fields + } diff --git a/packages/aws-sdk-signers/examples/curl_signer.py b/packages/aws-sdk-signers/examples/curl_signer.py new file mode 100644 index 000000000..3e9cf22dd --- /dev/null +++ b/packages/aws-sdk-signers/examples/curl_signer.py @@ -0,0 +1,68 @@ +""" +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +SPDX-License-Identifier: Apache-2.0 + +Sample signer using Requests. +""" + +import typing +from collections.abc import Iterable, Mapping +from urllib.parse import urlparse + +from aws_sdk_signers import AWSRequest, Field, Fields, SigV4Signer, URI + +if typing.TYPE_CHECKING: + from aws_sdk_signers import AWSCredentialIdentity, SigV4SigningProperties + + +class SigV4Curl: + """Generates a curl command with a SigV4 signature applied.""" + + signer = SigV4Signer() + + @classmethod + def generate_signed_curl_cmd( + cls, + properties: "SigV4SigningProperties", + identity: "AWSCredentialIdentity", + method: str, + url: str, + headers: Mapping[str, str], + body: Iterable[bytes] | None, + ) -> str: + url_parts = urlparse(url) + uri = URI( + scheme=url_parts.scheme, + host=url_parts.hostname, + port=url_parts.port, + path=url_parts.path, + query=url_parts.query, + fragment=url_parts.fragment, + ) + fields = Fields([Field(name=k, values=[v]) for k, v in headers.items()]) + awsrequest = AWSRequest( + destination=uri, + method=method, + body=body, + fields=fields, + ) + signed_request = cls.signer.sign( + properties=properties, + request=awsrequest, + identity=identity, + ) + return cls._construct_curl_cmd(request=signed_request) + + @classmethod + def _construct_curl_cmd(self, request: AWSRequest) -> str: + cmd_list = ["curl"] + cmd_list.append(f"-X {request.method.upper()}") + for header in request.fields: + cmd_list.append(f'-H "{header.name}: {header.as_string()}"') + if request.body is not None: + # Forcing bytes to a utf-8 string, if we need arbitrary bytes for the + # terminal we should add an option to write to file and use that + # in the command. + cmd_list.append(f"-d {b''.join(list(request.body)).decode()}") + cmd_list.append(request.destination.build()) + return " ".join(cmd_list) diff --git a/packages/aws-sdk-signers/examples/requests_signer.py b/packages/aws-sdk-signers/examples/requests_signer.py new file mode 100644 index 000000000..4b47039ea --- /dev/null +++ b/packages/aws-sdk-signers/examples/requests_signer.py @@ -0,0 +1,77 @@ +""" +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +SPDX-License-Identifier: Apache-2.0 + +Sample signer using Requests. +""" + +import typing +from urllib.parse import urlparse + +from aws_sdk_signers import AWSRequest, Field, Fields, SigV4Signer, URI +from requests import PreparedRequest +from requests.auth import AuthBase + +if typing.TYPE_CHECKING: + from aws_sdk_signers import AWSCredentialIdentity, SigV4SigningProperties + +SIGNING_HEADERS = ( + "Authorization", + "Date", + "X-Amz-Date", + "X-Amz-Security-Token", + "X-Amz-Content-SHA256", +) + + +class SigV4Auth(AuthBase): + """Attaches SigV4Authentication to the given Request object.""" + + def __init__( + self, + properties: "SigV4SigningProperties", + identity: "AWSCredentialIdentity", + ): + self._properties = properties + self._identity = identity + self._signer = SigV4Signer() + + def __eq__(self, other): + return self.properties == getattr(other, "properties", None) + + def __ne__(self, other): + return not self == other + + def __call__(self, r): + self.sign_request(r) + return r + + def sign_request(self, r: PreparedRequest): + request = self.convert_to_awsrequest(r) + signed_request = self._signer.sign( + properties=self._properties, + request=request, + identity=self._identity, + ) + for header in SIGNING_HEADERS: + if header in signed_request.fields: + r.headers[header] = signed_request.fields[header].as_string() + return r + + def convert_to_awsrequest(self, r: PreparedRequest) -> AWSRequest: + url_parts = urlparse(r.url) + uri = URI( + scheme=url_parts.scheme, + host=url_parts.hostname, + port=url_parts.port, + path=url_parts.path, + query=url_parts.query, + fragment=url_parts.fragment, + ) + fields = Fields([Field(name=k, values=[v]) for k, v in r.headers.items()]) + return AWSRequest( + destination=uri, + method=r.method, + body=r.body, + fields=fields, + )