from __future__ import annotations
from typing import Any, Dict, List, Optional
import boto3
from botocore.exceptions import ClientError
def collect_s3(session: boto3.Session) -> Dict[str, Any]:
# S3 is global-ish; bucket location can vary.
s3 = session.client("s3", region_name=session.region_name or "us-east-1")
buckets: List[Dict[str, Any]] = []
try:
resp = s3.list_buckets()
except ClientError as e:
return {"error": str(e), "buckets": []}
for b in resp.get("Buckets", []):
name = b.get("Name")
if not name:
continue
bucket_region: Optional[str] = None
try:
loc = s3.get_bucket_location(Bucket=name)
bucket_region = loc.get("LocationConstraint") or "us-east-1"
except ClientError:
bucket_region = None
pab = None
try:
pab = s3.get_public_access_block(Bucket=name).get("PublicAccessBlockConfiguration")
except ClientError:
pab = None
enc = None
try:
rules = s3.get_bucket_encryption(Bucket=name).get("ServerSideEncryptionConfiguration", {}).get("Rules", [])
enc = rules
except ClientError:
enc = None
buckets.append(
{
"name": name,
"created_at": str(b.get("CreationDate")) if b.get("CreationDate") else None,
"region": bucket_region,
"public_access_block": pab,
"encryption_rules": enc,
}
)
return {"buckets": buckets}
