gistfile1.txt

import hashlib
import hmac
import datetime
import os
import sys
import urllib.request
import urllib.parse
import xml.etree.ElementTree as ET

AWS_ACCESS_KEY_ID     = os.environ["AWS_ACCESS_KEY_ID"]
AWS_SECRET_ACCESS_KEY = os.environ["AWS_SECRET_ACCESS_KEY"]
AWS_REGION            = os.environ.get("AWS_REGION", "eu-west-1")
BUCKET_NAME           = os.environ["AWS_BUCKET_NAME"]
ENDPOINT              = os.environ.get("AWS_ENDPOINT", "https://s3.amazonaws.com")


# --- SigV4 Helpers ---

def sha256_hex(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def hmac_sha256(key: bytes, msg: str) -> bytes:
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()

def get_signing_key(secret_key: str, date: str, region: str, service: str) -> bytes:
    k_date    = hmac_sha256(("AWS4" + secret_key).encode("utf-8"), date)
    k_region  = hmac_sha256(k_date, region)
    k_service = hmac_sha256(k_region, service)
    k_signing = hmac_sha256(k_service, "aws4_request")
    return k_signing

def sigv4_headers(
    method: str,
    path: str,
    body: bytes,
    region: str,
    service: str = "s3",
    extra_headers: dict = None,
) -> dict:
    now        = datetime.datetime.utcnow()
    amzdate    = now.strftime("%Y%m%dT%H%M%SZ")
    datestamp  = now.strftime("%Y%m%d")
    host       = urllib.parse.urlparse(ENDPOINT).netloc

    headers = {
        "host":                 host,
        "x-amz-date":          amzdate,
        "x-amz-content-sha256": sha256_hex(body),
    }
    if extra_headers:
        headers.update({k.lower(): v for k, v in extra_headers.items()})

    # Canonical request
    signed_headers_str = ";".join(sorted(headers))
    canonical_headers  = "".join(f"{k}:{headers[k]}\n" for k in sorted(headers))
    canonical_request  = "\n".join([
        method,
        path,
        "",                         # query string (empty here)
        canonical_headers,
        signed_headers_str,
        headers["x-amz-content-sha256"],
    ])

    # String to sign
    credential_scope = f"{datestamp}/{region}/{service}/aws4_request"
    string_to_sign   = "\n".join([
        "AWS4-HMAC-SHA256",
        amzdate,
        credential_scope,
        sha256_hex(canonical_request.encode("utf-8")),
    ])

    # Signature
    signing_key = get_signing_key(AWS_SECRET_ACCESS_KEY, datestamp, region, service)
    signature   = hmac.new(signing_key, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    authorization = (
        f"AWS4-HMAC-SHA256 Credential={AWS_ACCESS_KEY_ID}/{credential_scope}, "
        f"SignedHeaders={signed_headers_str}, "
        f"Signature={signature}"
    )

    # Return only the headers urllib needs (exclude 'host', it's set automatically)
    return {
        "Authorization":          authorization,
        "x-amz-date":             amzdate,
        "x-amz-content-sha256":   headers["x-amz-content-sha256"],
        **({k: v for k, v in (extra_headers or {}).items()}),
    }


# --- S3 Operations ---

def s3_request(method: str, key: str, body: bytes = b"", extra_headers: dict = None):
    path     = f"/{BUCKET_NAME}/{key}"
    url      = f"{ENDPOINT}{path}"
    headers  = sigv4_headers(method, path, body, AWS_REGION, extra_headers=extra_headers)
    req      = urllib.request.Request(url, data=body or None, headers=headers, method=method)
    with urllib.request.urlopen(req) as resp:
        return resp.status, resp.read()


def list_bucket():
    path    = f"/{BUCKET_NAME}/"
    url     = f"{ENDPOINT}{path}"
    headers = sigv4_headers("GET", path, b"", AWS_REGION)
    req     = urllib.request.Request(url, headers=headers, method="GET")
    with urllib.request.urlopen(req) as resp:
        return resp.status, resp.read()


if __name__ == "__main__":
    if len(sys.argv) == 1:
        status, body = list_bucket()
        ns = {"s3": "http://s3.amazonaws.com/doc/2006-03-01/"}
        root = ET.fromstring(body)
        objects = root.findall("s3:Contents", ns)
        if not objects:
            print("(empty bucket)")
        else:
            for obj in objects:
                key      = obj.findtext("s3:Key", namespaces=ns)
                size     = int(obj.findtext("s3:Size", namespaces=ns))
                modified = obj.findtext("s3:LastModified", namespaces=ns)
                print(f"{modified}  {size:>12,}  {key}")
    else:
        key = sys.argv[1]
        dest = sys.argv[2] if len(sys.argv) > 2 else os.path.basename(key)
        status, body = s3_request("GET", key)
        with open(dest, "wb") as f:
            f.write(body)
        print(f"GET {key} → {status}, saved to {dest}")