AWS v4 signature sample

v4_signature_sample.py

import datetime
import hashlib
import hmac
import os
import requests

import requests_aws4auth
import aws_requests_auth.aws_auth
import botocore.auth
import botocore.awsrequest
import botocore.credentials
import botocore.httpsession


def list_bucket_requests_aws4auth():
    url = 'https://s3.{}.amazonaws.com/'.format(os.environ['AWS_REGION'])
    auth = requests_aws4auth.AWS4Auth(
        os.environ['AWS_ACCESS_KEY_ID'],
        os.environ['AWS_SECRET_ACCESS_KEY'],
        os.environ['AWS_REGION'],
        's3',
        session_token=os.environ['AWS_SESSION_TOKEN'],
    )
    response = requests.get(url, auth=auth)
    return response.text


def list_bucket_aws_requests_auth():
    host = 's3.{}.amazonaws.com'.format(os.environ['AWS_REGION'])
    url = 'https://s3.{}.amazonaws.com/'.format(os.environ['AWS_REGION'])
    auth = aws_requests_auth.aws_auth.AWSRequestsAuth(
        aws_access_key=os.environ['AWS_ACCESS_KEY_ID'],
        aws_secret_access_key=os.environ['AWS_SECRET_ACCESS_KEY'],
        aws_token=os.environ['AWS_SESSION_TOKEN'],
        aws_host=host,
        aws_region=os.environ['AWS_REGION'],
        aws_service='s3')
    response = requests.get(url, auth=auth)
    return response.text


def list_bucket_botocore():
    url = 'https://s3.{}.amazonaws.com/'.format(os.environ['AWS_REGION'])
    credentials = botocore.credentials.Credentials(
        os.environ['AWS_ACCESS_KEY_ID'],
        os.environ['AWS_SECRET_ACCESS_KEY'],
        os.environ['AWS_SESSION_TOKEN'])
    request = botocore.awsrequest.AWSRequest(method='GET', url=url)
    botocore.auth.S3SigV4Auth(
        credentials, 's3', os.environ['AWS_REGION']).add_auth(request)
    response = botocore.httpsession.URLLib3Session().send(request.prepare())
    return response.text


def list_bucket_diy():
    url = 'https://s3.{}.amazonaws.com/'.format(os.environ['AWS_REGION'])
    host = 's3.{}.amazonaws.com'.format(os.environ['AWS_REGION'])
    t = datetime.datetime.utcnow()
    amzdate = t.strftime('%Y%m%dT%H%M%SZ')
    datestamp = t.strftime('%Y%m%d')

    # task 1: create a canonical request
    method = 'GET'
    canonical_uri = '/'
    canonical_query_string = ''
    hashed_payload = hashlib.sha256(''.encode('utf-8')).hexdigest()
    canonical_headers = \
        'content-type:application/x-www-form-urlencoded; charset=utf-8\n' \
        + 'host:{}\n'.format(host) \
        + 'x-amz-content-sha256:{}\n'.format(hashed_payload) \
        + 'x-amz-date:{}\n'.format(amzdate) \
        + 'x-amz-security-token:{}\n'.format(os.environ['AWS_SESSION_TOKEN'])
    signed_headers = ';'.join([
        'content-type', 'host', 'x-amz-content-sha256', 'x-amz-date',
        'x-amz-security-token'])
    canonical_request = '\n'.join([
        method, canonical_uri, canonical_query_string, canonical_headers,
        signed_headers, hashed_payload])
    hashed_canonical_request = hashlib.sha256(
        canonical_request.encode('utf-8')).hexdigest()

    # task 2: create string to sign
    algorithm = 'AWS4-HMAC-SHA256'
    credential_scope = '/'.join([
        datestamp, os.environ['AWS_REGION'], 's3', 'aws4_request'])
    string_to_sign = '\n'.join([
        algorithm, amzdate, credential_scope, hashed_canonical_request])

    # task 3: calculate the signature
    def sign(key, msg):
        return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
    kSecret = os.environ['AWS_SECRET_ACCESS_KEY']
    kDate = sign(('AWS4' + kSecret).encode('utf-8'), datestamp)
    kRegion = sign(kDate, os.environ['AWS_REGION'])
    kService = sign(kRegion, 's3')
    kSigning = sign(kService, 'aws4_request')
    signature = hmac.new(
        kSigning, string_to_sign.encode('utf-8'), hashlib.sha256
    ).hexdigest()

    # task 4: add signing information to the request
    authorization = \
        '{} Credential={}/{}, SignedHeaders={}, Signature={}'.format(
            algorithm, os.environ['AWS_ACCESS_KEY_ID'], credential_scope,
            signed_headers, signature)
    headers = {
        'Content-Type': 'application/x-www-form-urlencoded; charset=utf-8',
        'x-amz-date': amzdate,
        'x-amz-content-sha256': hashed_payload,
        'Authorization': authorization,
        'X-Amz-Security-Token': os.environ['AWS_SESSION_TOKEN'],
    }

    response = requests.get(url, headers=headers)
    return response.text


if __name__ == '__main__':
    print('*** requests_aws4auth ***')
    print(list_bucket_requests_aws4auth())

    print('\n*** aws_requests_auth ***')
    print(list_bucket_aws_requests_auth())

    print('\n*** botocore ***')
    print(list_bucket_botocore())

    print('\n*** diy ***')
    print(list_bucket_diy())