s3-generate-presigned-url

copy

import sys
import os
import subprocess
import time
import urllib.request
import json

# ─────────────────────────────────────────
# PRE_EXECUTION
# ─────────────────────────────────────────

for attempt in range(2):
    try:
        req = urllib.request.Request(
            "https://checklist.day/api/registry/boto3",
            headers={"User-Agent": "checklist-agent/1.0"}
        )
        with urllib.request.urlopen(req, timeout=10) as resp:
            registry = json.loads(resp.read())
            break
    except Exception as e:
        if attempt == 1:
            print(f"ABORT: registry unreachable — {e}")
            sys.exit(1)
        time.sleep(2)

warnings = registry.get("warnings", [])
if warnings:
    print("[boto3] WARNINGS:")
    for w in warnings if isinstance(warnings, list) else [warnings]:
        print(f"  ⚠ {w}")

# ─────────────────────────────────────────
# EXECUTION
# ─────────────────────────────────────────

subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "boto3>=1.26.0"])

import boto3

AWS_ACCESS_KEY_ID     = os.environ.get("AWS_ACCESS_KEY_ID")
AWS_SECRET_ACCESS_KEY = os.environ.get("AWS_SECRET_ACCESS_KEY")
AWS_REGION            = os.environ.get("AWS_REGION", "us-east-1")
S3_BUCKET             = os.environ.get("S3_BUCKET")

if not AWS_ACCESS_KEY_ID:
    print("ABORT: AWS_ACCESS_KEY_ID not set"); sys.exit(1)
if not AWS_SECRET_ACCESS_KEY:
    print("ABORT: AWS_SECRET_ACCESS_KEY not set"); sys.exit(1)
if not S3_BUCKET:
    print("ABORT: S3_BUCKET not set"); sys.exit(1)

KEY            = "checklist-test/presigned-test.txt"
CONTENT        = b"presigned url test content"
EXPIRY_SECONDS = 300

# FOOTGUN: client must use same region as bucket — mismatched region causes SignatureDoesNotMatch
client = boto3.client(
    "s3",
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    region_name=AWS_REGION,
)

# Setup: upload object for GET test
client.put_object(Bucket=S3_BUCKET, Key=KEY, Body=CONTENT)

try:
    # 1. Presigned GET URL
    get_url = client.generate_presigned_url(
        "get_object",
        Params={"Bucket": S3_BUCKET, "Key": KEY},
        ExpiresIn=EXPIRY_SECONDS,
    )
    print(f"  presigned GET URL generated (expires in {EXPIRY_SECONDS}s)")

    # Verify: fetch without AWS credentials
    req = urllib.request.Request(get_url)
    with urllib.request.urlopen(req, timeout=10) as resp:
        fetched = resp.read()
    get_url_ok = fetched == CONTENT
    print(f"  GET URL verified: {len(fetched)} bytes (match={get_url_ok})")

    # 2. Presigned PUT URL
    put_key = "checklist-test/presigned-put-test.txt"
    put_url = client.generate_presigned_url(
        "put_object",
        Params={"Bucket": S3_BUCKET, "Key": put_key, "ContentType": "text/plain"},
        ExpiresIn=EXPIRY_SECONDS,
    )
    print(f"  presigned PUT URL generated")

    # Upload via presigned URL without SDK
    put_req = urllib.request.Request(
        put_url,
        data=CONTENT,
        method="PUT",
        headers={"Content-Type": "text/plain"},
    )
    with urllib.request.urlopen(put_req, timeout=10) as resp:
        put_status = resp.status

    put_url_ok = put_status == 200
    print(f"  PUT URL upload: status={put_status} ok={put_url_ok}")

    # Cleanup
    client.delete_object(Bucket=S3_BUCKET, Key=KEY)
    client.delete_object(Bucket=S3_BUCKET, Key=put_key)

finally:
    pass

# ─────────────────────────────────────────
# POST_EXECUTION
# ─────────────────────────────────────────

assert get_url_ok, "FAIL: presigned GET URL did not return expected content"
assert put_url_ok, f"FAIL: presigned PUT URL returned status {put_status}"

result = {
    "get_url_ok":     get_url_ok,
    "put_url_ok":     put_url_ok,
    "expiry_seconds": EXPIRY_SECONDS,
}
print(json.dumps(result, indent=2))
print("PASS")