s3-bucket-policy-and-cors

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
from botocore.exceptions import ClientError

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)

client = boto3.client(
    "s3",
    aws_access_key_id=AWS_ACCESS_KEY_ID,
    aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    region_name=AWS_REGION,
)

# 1. CORS Configuration
# FOOTGUN: boto3 uses dict with CORSRules key — NOT raw XML
# FOOTGUN: AllowedMethods must be a list of strings, not a single string
cors_config = {
    "CORSRules": [
        {
            "AllowedHeaders": ["*"],
            "AllowedMethods": ["GET", "PUT", "POST", "DELETE"],
            "AllowedOrigins": ["https://checklist.day", "http://localhost:3000"],
            "ExposeHeaders":  ["ETag"],
            "MaxAgeSeconds":  3600,
        }
    ]
}

cors_set_ok   = False
cors_verified = False

try:
    client.put_bucket_cors(Bucket=S3_BUCKET, CORSConfiguration=cors_config)
    cors_set_ok = True
    print(f"  CORS config applied")

    # Verify readback
    response     = client.get_bucket_cors(Bucket=S3_BUCKET)
    rules        = response.get("CORSRules", [])
    cors_verified = (
        len(rules) > 0 and
        "GET" in rules[0].get("AllowedMethods", []) and
        "https://checklist.day" in rules[0].get("AllowedOrigins", [])
    )
    print(f"  CORS verified: {cors_verified} ({len(rules)} rule(s))")

    # Cleanup CORS
    client.delete_bucket_cors(Bucket=S3_BUCKET)

except ClientError as e:
    print(f"  CORS error: {e.response['Error']['Code']}: {e.response['Error']['Message']}")

# 2. Bucket Policy
# FOOTGUN: policy must be a JSON string, not a dict
policy_set_ok   = False
policy_verified = False

policy_dict = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid":       "ChecklistReadOnly",
            "Effect":    "Deny",
            "Principal": "*",
            "Action":    "s3:DeleteBucket",
            "Resource":  f"arn:aws:s3:::{S3_BUCKET}",
        }
    ],
}

try:
    client.put_bucket_policy(
        Bucket=S3_BUCKET,
        Policy=json.dumps(policy_dict),  # FOOTGUN: must be string
    )
    policy_set_ok = True
    print(f"  bucket policy applied")

    # Verify readback
    response        = client.get_bucket_policy(Bucket=S3_BUCKET)
    returned_policy = json.loads(response["Policy"])
    policy_verified = "Statement" in returned_policy and len(returned_policy["Statement"]) > 0
    print(f"  policy verified: {policy_verified}")

    # Cleanup policy
    client.delete_bucket_policy(Bucket=S3_BUCKET)

except ClientError as e:
    print(f"  policy error: {e.response['Error']['Code']}: {e.response['Error']['Message']}")

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

assert cors_set_ok, "FAIL: CORS config could not be applied"
assert cors_verified, "FAIL: CORS config readback did not match"

result = {
    "cors_set_ok":     cors_set_ok,
    "cors_verified":   cors_verified,
    "policy_set_ok":   policy_set_ok,
    "policy_verified": policy_verified,
}
print(json.dumps(result, indent=2))
print("PASS")