from __future__ import annotations
from typing import Any, Dict, List
import boto3
from botocore.exceptions import ClientError
def collect_cloudtrail(session: boto3.Session, region: str) -> Dict[str, Any]:
ct = session.client("cloudtrail", region_name=region)
trails: List[Dict[str, Any]] = []
try:
resp = ct.describe_trails(includeShadowTrails=False)
for t in resp.get("trailList", []):
trails.append(
{
"name": t.get("Name"),
"trail_arn": t.get("TrailARN"),
"is_multi_region_trail": t.get("IsMultiRegionTrail"),
"s3_bucket_name": t.get("S3BucketName"),
"cloudwatch_logs_log_group_arn": t.get("CloudWatchLogsLogGroupArn"),
"kms_key_id": t.get("KmsKeyId"),
"home_region": t.get("HomeRegion"),
}
)
except ClientError as e:
return {"region": region, "error": str(e), "trails": []}
return {"region": region, "trails": trails}
def collect_cloudwatch_alarm_count(session: boto3.Session, region: str) -> Dict[str, Any]:
cw = session.client("cloudwatch", region_name=region)
try:
# MVP: just count alarms (paginates)
count = 0
p = cw.get_paginator("describe_alarms")
for page in p.paginate():
count += len(page.get("MetricAlarms", []))
count += len(page.get("CompositeAlarms", []))
return {"region": region, "alarm_count": count}
except ClientError as e:
return {"region": region, "error": str(e), "alarm_count": None}
