audit-labs/audit-tools

 1"""
 2Collect audit-logging posture: CloudTrail trails and AWS Config recorders.
 3"""
 4
 5import sys
 6
 7from botocore.exceptions import ClientError
 8
 9from .api import enabled_regions
10
11
12def cloudtrail(cfg):
13    """
14    One row per CloudTrail trail: whether it is logging, multi-region, and has
15    log-file validation. An empty result means no trails are configured.
16    """
17    region = cfg.get("region") or "us-east-1"
18    ct = cfg["session"].client("cloudtrail", region_name=region)
19    rows = []
20    for trail in ct.describe_trails(includeShadowTrails=False).get("trailList", []):
21        try:
22            status = ct.get_trail_status(Name=trail["TrailARN"])
23        except ClientError:
24            status = {}
25        rows.append(
26            {
27                "name": trail.get("Name", ""),
28                "home_region": trail.get("HomeRegion", ""),
29                "multi_region": trail.get("IsMultiRegionTrail"),
30                "log_file_validation": trail.get("LogFileValidationEnabled"),
31                "is_logging": status.get("IsLogging"),
32                "s3_bucket": trail.get("S3BucketName", ""),
33            }
34        )
35    return rows
36
37
38def config_recorders(cfg):
39    """
40    One row per region: whether AWS Config is recording. Regions with no
41    recorder are reported so gaps are visible.
42    """
43    session = cfg["session"]
44    rows = []
45    for region in enabled_regions(cfg):
46        try:
47            cc = session.client("config", region_name=region)
48            recorders = cc.describe_configuration_recorders().get(
49                "ConfigurationRecorders", []
50            )
51            statuses = {
52                s["name"]: s
53                for s in cc.describe_configuration_recorder_status().get(
54                    "ConfigurationRecordersStatus", []
55                )
56            }
57        except ClientError as e:
58            print(f"  Skipping {region}: config returned {e}", file=sys.stderr)
59            continue
60
61        if not recorders:
62            rows.append(
63                {
64                    "region": region,
65                    "recorder": "(none)",
66                    "recording": False,
67                    "last_status": "",
68                }
69            )
70            continue
71        for r in recorders:
72            st = statuses.get(r["name"], {})
73            rows.append(
74                {
75                    "region": region,
76                    "recorder": r["name"],
77                    "recording": st.get("recording"),
78                    "last_status": st.get("lastStatus", ""),
79                }
80            )
81    return rows