s3-connect-and-list-buckets

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, NoCredentialsError

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")

if not AWS_ACCESS_KEY_ID:
    print("ABORT: AWS_ACCESS_KEY_ID env var not set")
    sys.exit(1)
if not AWS_SECRET_ACCESS_KEY:
    print("ABORT: AWS_SECRET_ACCESS_KEY env var not set")
    sys.exit(1)

connected    = False
bucket_count = 0
buckets      = []
error        = None

try:
    # FOOTGUN: always pass credentials explicitly — don't rely on ~/.aws/credentials in agents
    client = boto3.client(
        "s3",
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
        region_name=AWS_REGION,
    )

    response     = client.list_buckets()
    buckets      = [b["Name"] for b in response.get("Buckets", [])]
    bucket_count = len(buckets)
    connected    = True
    print(f"  connected — {bucket_count} bucket(s) accessible")
    for b in buckets[:5]:
        print(f"    - {b}")
    if bucket_count > 5:
        print(f"    ... and {bucket_count - 5} more")

except NoCredentialsError as e:
    error = f"NoCredentialsError: {e}"
    print(f"  auth failed: {error}")
except ClientError as e:
    error = f"{e.response['Error']['Code']}: {e.response['Error']['Message']}"
    print(f"  client error: {error}")

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

result = {
    "connected":    connected,
    "bucket_count": bucket_count,
    "buckets":      buckets[:10],
    "error":        error,
}
print(json.dumps(result, indent=2))

if not connected:
    print(f"FAIL: {error}")
    sys.exit(1)

assert connected, "FAIL: not connected"
print("PASS")