Authenticate for AWS S3 request

s3authentication.py

from datetime import datetime
from hashlib import sha256
import hmac
import requests
from os import environ as env


# templates for creating the auth header
Canonical_Request = '{HTTPMethod}\n{CanonicalURI}\n{CanonicalQueryString}\n{CanonicalHeaders}\n\n{SignedHeaders}\n{HashedPayload}'
Scope = '{DATE}/{REGION}/{SERVICE_NAME}/{KIND}'
Authorization_Header = '{SIGNING_ALGORITHM} Credential={ACCESS_KEY_ID}/{SCOPE},SignedHeaders={signedHeaders},Signature={signature}'
String_To_Sign = '{SIGNING_ALGORITHM}\n{ISODATE}\n{SCOPE}\n{hexed_request}'
# templates for http endpoint
Bucket_Host = '{BUCKET}.s3.{REGION}.amazonaws.com' # {BUCKET}.s3.amazonaws.com
Object_URL = '{PROTOCOL}://{BUCKET_HOST}{PATH}{QUERY}'

# config
config = {
    'ACCESS_KEY_ID': env["ACCESS_KEY_ID"],
    'ACCESS_KEY': env["ACCESS_KEY"],
    'SIGNING_ALGORITHM': 'AWS4-HMAC-SHA256',
    'SERVICE_NAME': 's3',
    'DATE': datetime.utcnow().strftime('%Y%m%d'),
    'ISODATE': datetime.utcnow().strftime("%Y%m%dT%H%M%SZ"),
    'PREFIX': 'AWS4',
    'KIND': 'aws4_request',
    'SCOPE': None,
    'PROTOCOL': 'http',
    'REGION': 'eu-central-1',
    'BUCKET': 'lazydog',
    'PATH': '/buckle.txt',
    'QUERY': '?acl=',
}

# add scope and bucket host to config for convenience
config["SCOPE"] = Scope.format(**config)
config["BUCKET_HOST"] = Bucket_Host.format(**config)

# create signing key
key = str.encode('{PREFIX}{ACCESS_KEY}'.format(**config))
DateKey = hmac.new(key, str.encode('{DATE}'.format(**config)), sha256).digest()
DateRegionKey = hmac.new(DateKey, str.encode('{REGION}'.format(**config)), sha256).digest()
DateRegionServiceKey = hmac.new(DateRegionKey, str.encode('{SERVICE_NAME}'.format(**config)), sha256).digest()
SigningKey =  hmac.new(DateRegionServiceKey, str.encode('{KIND}'.format(**config)), sha256).digest()

# empty payload for this get request
hashed_payload = sha256(b'').hexdigest()

# create headers
headers  = {
    'host': Bucket_Host.format(**config),
    'x-amz-content-sha256': hashed_payload,
    'x-amz-date': config["ISODATE"],
}

ch = []
sh = []
for k, v in headers.items():
    ch.append(f'{k}:{v}')
    sh.append(k)
    
canonical_headers = '\n'.join(ch) 
signed_headers = ';'.join(sh)

# build canonical request with current facts
# make sure proper formatted strings are passed here
# this is a good point to write some kind of adapter
# to transform a a raw requests into the canonical form
canonical_request = Canonical_Request.format(**{
    'HTTPMethod': 'GET',
    'CanonicalURI':  config["PATH"],
    'CanonicalQueryString': config['QUERY'][1:],
    'CanonicalHeaders': canonical_headers,
    'SignedHeaders': signed_headers,
    'HashedPayload': hashed_payload,
})

# create the string to sign
string_to_sign = String_To_Sign.format(**{
    **config,
    'hexed_request': sha256(str.encode(canonical_request)).hexdigest(),
})

# sign the string to obtain signature
signature = hmac.new(SigningKey,  str.encode(string_to_sign), sha256).hexdigest()

# create auth headers using signature and signed headers
authorization_header = Authorization_Header.format(**{
    **config,
    'signedHeaders': signed_headers,
    'signature': signature,
})

headers["Authorization"] = authorization_header

# check how things look
print(f"canonical request:\n{canonical_request}\n")
print(f"string to sign:\n{string_to_sign}\n")
print(f'auth header:\n{authorization_header}\n')

# finally make request
object_url = Object_URL.format(**config)
req = requests.Request('GET', url=object_url, headers=headers)
prepared = req.prepare()
s = requests.session()
resp = s.send(prepared)
print(f'response:\n{resp.content}\n')