add check_cloudtrail

local-app/python-tools/cross-organization/assess_check_cloudtrail.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+
+import json
+import argparse
+import sys
+
+__version__ = "1.0.0"
+
+def main():
+    parser = argparse.ArgumentParser(description="AWS CloudTrail Audit Assessor")
+    parser.add_argument("--input", required=True, help="JSON file from check_cloudtrail")
+    args = parser.parse_args()
+
+    with open(args.input, 'r') as f:
+        data = json.load(f)
+
+    print("-" * 120)
+    print(f"{'Account ID':<15} | {'Global Summary':<25} | {'Issues/Best Practices'}")
+    print("-" * 120)
+
+    for account in data:
+        acc_id = account.get("account_id")
+        checks = account.get("data", {})
+        summary = checks.get("account_summary", {}).get("_summary", "UNKNOWN")
+
+        issues = []
+
+        # Logic for Summary Field examination
+        if "DUPLICATIVE" in summary:
+            issues.append("Redundant: Local trails exist alongside Org trail (Extra Cost)")
+        elif "NO_TRAILS" in summary:
+            issues.append("CRITICAL: No CloudTrail logging detected")
+        elif "LOCAL_ONLY" in summary:
+            issues.append("Note: Using local trails; consider migrating to Org Trail")
+
+        # Detail Checks
+        for key, val in checks.items():
+            if key.startswith("trail:"):
+                t_name = val.get("trail_name")
+                if val.get("is_logging") == "False":
+                    issues.append(f"Trail {t_name} is STOPPED")
+                if val.get("is_multi_region") == "False":
+                    issues.append(f"Trail {t_name} is Single-Region only")
+                if val.get("log_file_validation") == "False":
+                    issues.append(f"Trail {t_name} missing Log File Validation")
+                if val.get("kms_key_id") == "SSE-S3 (Default)":
+                    issues.append(f"Trail {t_name} not using CMK encryption")
+                if val.get("cw_logs_retention_days") == "Never Expire":
+                    issues.append(f"Trail {t_name} CloudWatch Logs set to Never Expire (Cost Risk)")
+
+        issue_str = " | ".join(issues) if issues else "Compliant"
+        print(f"{acc_id:<15} | {summary:<25} | {issue_str}")
+
+if __name__ == "__main__":
+    main()

local-app/python-tools/cross-organization/check_cloudtrail.py

@@ -0,0 +1,102 @@
+import boto3
+import time
+from datetime import datetime, timedelta
+
+__version__ = "1.0.0"
+
+def get_s3_metrics(session, bucket_name, region):
+    """Fetches bucket storage metrics from CloudWatch."""
+    cw = session.client('cloudwatch', region_name=region)
+    metrics = {"bucket_size_bytes": 0, "object_count": 0}
+    end = datetime.utcnow()
+    start = end - timedelta(days=2)
+    try:
+        r1 = cw.get_metric_statistics(
+            Namespace='AWS/S3', MetricName='BucketSizeBytes',
+            Dimensions=[{'Name': 'BucketName', 'Value': bucket_name}, {'Name': 'StorageType', 'Value': 'StandardStorage'}],
+            StartTime=start, EndTime=end, Period=86400, Statistics=['Average']
+        )
+        if r1['Datapoints']: metrics["bucket_size_bytes"] = int(r1['Datapoints'][-1]['Average'])
+
+        r2 = cw.get_metric_statistics(
+            Namespace='AWS/S3', MetricName='NumberOfObjects',
+            Dimensions=[{'Name': 'BucketName', 'Value': bucket_name}, {'Name': 'StorageType', 'Value': 'AllStorageTypes'}],
+            StartTime=start, EndTime=end, Period=86400, Statistics=['Average']
+        )
+        if r2['Datapoints']: metrics["object_count"] = int(r2['Datapoints'][-1]['Average'])
+    except: pass
+    return metrics
+
+def get_log_group_details(session, group_arn, region):
+    """Fetches retention and size for CloudWatch Log Groups."""
+    if not group_arn: return {}
+    logs = session.client('logs', region_name=region)
+    group_name = group_arn.split(':')[-2] # Extract name from ARN
+    try:
+        resp = logs.describe_log_groups(logGroupNamePrefix=group_name)
+        for g in resp.get('logGroups', []):
+            if g['logGroupName'] == group_name:
+                return {
+                    "cw_logs_retention_days": g.get('retentionInDays', 'Never Expire'),
+                    "cw_logs_size_bytes": g.get('storedBytes', 0)
+                }
+    except: pass
+    return {}
+
+def account_task(account_session, account_id, account_name, region):
+    results = {"alias": "N/A", "data": {}}
+    org_trail_count = 0
+    local_trail_count = 0
+
+    try:
+        results["alias"] = account_session.client('iam').list_account_aliases().get('AccountAliases', ["N/A"])[0]
+        ct = account_session.client('cloudtrail', region_name=region)
+
+        # describe_trails retrieves trails for the current region (+ shadow trails if multi-region)
+        trails = ct.describe_trails(includeShadowTrails=True).get('trailList', [])
+
+        for trail in trails:
+            trail_name = trail['Name']
+            status = ct.get_trail_status(Name=trail_name)
+
+            is_org = trail.get('IsOrganizationTrail', False)
+            if is_org: org_trail_count += 1
+            else: local_trail_count += 1
+
+            trail_data = {
+                "trail_name": trail_name,
+                "is_logging": str(status.get('IsLogging', False)),
+                "is_org_trail": str(is_org),
+                "is_multi_region": str(trail.get('IsMultiRegionTrail', False)),
+                "log_file_validation": str(trail.get('LogFileValidationEnabled', False)),
+                "s3_bucket": trail.get('S3BucketName', 'N/A'),
+                "kms_key_id": trail.get('KmsKeyId', 'SSE-S3 (Default)'),
+                "sns_topic": trail.get('SNSTopicName', 'N/A'),
+                "home_region": trail.get('HomeRegion', 'N/A')
+            }
+
+            # Fetch CloudWatch Logs Integration
+            if 'CloudWatchLogsLogGroupArn' in trail:
+                trail_data.update(get_log_group_details(account_session, trail['CloudWatchLogsLogGroupArn'], region))
+
+            # Fetch S3 Metrics (if bucket is in the same account)
+            if trail_data['s3_bucket'] != 'N/A':
+                trail_data.update(get_s3_metrics(account_session, trail_data['s3_bucket'], region))
+
+            results["data"][f"trail:{trail_name}"] = trail_data
+
+        # Add Per-Account Summary
+        summary = "OK"
+        if org_trail_count > 0 and local_trail_count > 0:
+            summary = f"DUPLICATIVE/{local_trail_count}_LOCAL_WITH_ORG"
+        elif org_trail_count == 0 and local_trail_count > 0:
+            summary = f"LOCAL_ONLY/{local_trail_count}"
+        elif org_trail_count == 0 and local_trail_count == 0:
+            summary = "NO_TRAILS"
+
+        results["data"]["account_summary"] = {"_summary": summary}
+
+    except Exception as e:
+        results["error"] = str(e)
+
+    return results