AWS Signature Version 4 Signing for Requests
requests-aws-sign is a Python package that enables AWS Signature Version 4 (SigV4) request signing using the popular `requests` library. It provides the `AWSV4Sign` class which extends `requests.auth.AuthBase` to handle the intricate SigV4 signing process for HTTP requests to AWS services. The current version is 0.1.6, and it appears to be in a maintenance state, with the last release in July 2020.
Warnings
- gotcha AWS SigV4 signing requires specific parameters: valid AWS credentials, the correct AWS region, and the exact AWS service name (e.g., 's3', 'es', 'execute-api'). Incorrect values for any of these parameters will result in 'SignatureDoesNotMatch' or 'IncompleteSignature' errors from AWS services.
- gotcha When using AWS Security Token Service (STS) temporary credentials (e.g., from an assumed role or EC2 instance profile), the `AWSV4Sign` class needs the `session_token` in addition to `access_key` and `secret_key`. Failure to include the `X-Amz-Security-Token` header (which the library adds if `session_token` is provided) will lead to authentication failures like 'ExpiredTokenException'.
- gotcha This library primarily focuses on signing individual `requests` calls. For making multiple signed requests to the same AWS service efficiently, it's generally best practice to use a `requests.Session` object. The `AWSV4Sign` object can be assigned to a session's `auth` attribute, but the library does not provide a specialized `Session` subclass itself.
- breaking While `requests-aws-sign` itself is designed for SigV4, older AWS SDKs or custom implementations might still use Signature Version 2 (SigV2). AWS has deprecated SigV2 for new S3 buckets created after June 24, 2020, and strongly encourages migration to SigV4 for all services due to enhanced security. If you are integrating with an existing system that uses SigV2, this library will not be compatible.
Install
-
pip install requests-aws-sign
Imports
- AWSV4Sign
from requests_aws_sign import AWSV4Sign
Quickstart
import requests
from requests_aws_sign import AWSV4Sign
from boto3 import session
import os
# NOTE: For a real application, ensure AWS credentials are set via environment
# variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN)
# or AWS config files, so boto3 can find them.
# For this example, we'll try to get them, but you might need to set them.
session_boto3 = session.Session()
credentials = session_boto3.get_credentials() if session_boto3.get_credentials() else None
if not credentials or not credentials.access_key or not credentials.secret_key:
print("Warning: AWS credentials not found. Using dummy credentials. This request will likely fail.")
print("Please configure AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, and optionally AWS_SESSION_TOKEN environment variables.")
access_key = os.environ.get('AWS_ACCESS_KEY_ID', 'AKIAIOSFODNN7EXAMPLE')
secret_key = os.environ.get('AWS_SECRET_ACCESS_KEY', 'wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY')
session_token = os.environ.get('AWS_SESSION_TOKEN', None)
# Mocking a credentials object if boto3 couldn't find them
class MockCredentials:
def __init__(self, access_key, secret_key, token):
self.access_key = access_key
self.secret_key = secret_key
self.token = token
credentials = MockCredentials(access_key, secret_key, session_token)
region = session_boto3.region_name or 'us-east-1' # Default to us-east-1 if boto3 can't determine
service = 'es' # Example service, e.g., 's3', 'execute-api', 'es'
# This URL is an example and likely won't work without a real Elasticsearch domain
# Replace with a real AWS service endpoint you have access to
url = f"https://{service}-domain-example.{region}.es.amazonaws.com/"
auth = AWSV4Sign(credentials, region, service)
try:
response = requests.get(url, auth=auth, timeout=5) # Added timeout
response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx)
print(f"Successfully signed and sent request to {url}")
print(f"Status Code: {response.status_code}")
# print(response.text) # Uncomment to see response body
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
if hasattr(e, 'response') and e.response is not None:
print(f"Response status code: {e.response.status_code}")
print(f"Response headers: {e.response.headers}")
print(f"Response body: {e.response.text}")