smithy-aws-core

import asyncio import os import datetime from smithy_aws_core.config import AwsUserConfig from smithy_aws_core.signing import AwsSigner from smithy_python.types import Credentials, UnresolvedUrl from smithy_python.http import HttpRequest async def main(): # In a real application, consider using aiobotocore.session.get_session() # for robust credential discovery. For this quickstart, we use environment variables. aws_access_key_id = os.environ.get("AWS_ACCESS_KEY_ID", "AKIAIOSFODNN7EXAMPLE") aws_secret_access_key = os.environ.get("AWS_SECRET_ACCESS_KEY", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY") aws_session_token = os.environ.get("AWS_SESSION_TOKEN", None) if aws_access_key_id == "AKIAIOSFODNN7EXAMPLE" or aws_secret_access_key == "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY": print("Warning: Using dummy AWS credentials. Set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables for real usage.") credentials = Credentials( access_key_id=aws_access_key_id, secret_access_key=aws_secret_access_key, token=aws_session_token ) config = AwsUserConfig( region="us-east-1", credentials=credentials, endpoint_url=UnresolvedUrl("https://s3.us-east-1.amazonaws.com") # Example S3 endpoint ) print(f"\nCreated AWS User Config for region: {config.region}") # Demonstrate an AWS Signer signer = AwsSigner(service_id="s3", config=config) # 's3' is the service ID for S3 print(f"Created AWS Signer for service: {signer.service_id}") # Prepare a dummy HTTP request to be signed request = HttpRequest( method="GET", url="https://s3.us-east-1.amazonaws.com/my-bucket", # Example URL, typically resolved by client headers={ "Host": "s3.us-east-1.amazonaws.com", "User-Agent": "smithy-python-quickstart/1.0" }, body=b"" ) # Sign the request # Note: The 'url' in HttpRequest must be a concrete string, not UnresolvedUrl, for signing. # In a real client, endpoint resolution would convert UnresolvedUrl to a string. signed_request = await signer.sign(request, credentials, datetime.datetime.now(datetime.timezone.utc)) print(f"\n--- Signed Request Details ---") print(f"Method: {signed_request.method}") print(f"URL: {signed_request.url}") print(f"Headers: {signed_request.headers}") if 'Authorization' in signed_request.headers: print(f"Authorization header present (truncated): {signed_request.headers['Authorization'][:50]}...") else: print("Authorization header NOT found.") asyncio.run(main())