requests-aws-sign

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}")