update to summarize sizes

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

@@ -4,7 +4,8 @@
 import argparse
 import sys
 
-__version__ = "1.0.0"
+# --- VERSIONING ---
+__version__ = "1.0.1"
 
 def main():
     parser = argparse.ArgumentParser(description="AWS CloudTrail Audit Assessor")
@@ -14,42 +15,46 @@ def main():
     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)
+    print("-" * 130)
+    print(f"{'Account ID':<15} | {'Global Summary':<25} | {'Security/Cost Issues'}")
+    print("-" * 130)
+
+    org_stats = {"s3_objects": 0, "s3_bytes": 0, "cw_logs_size": 0, "cw_logs_count": 0}
 
     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
+            issues.append("DUPLICATE_LOGGING_COSTS")
+
         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"
+                # S3 Accumulation
+                org_stats["s3_objects"] += val.get("object_count", 0)
+                org_stats["s3_bytes"] += val.get("bucket_size_bytes", 0)
+
+                # CW Logs Accumulation
+                if "cw_logs_size_bytes" in val:
+                    org_stats["cw_logs_count"] += 1
+                    org_stats["cw_logs_size"] += val["cw_logs_size_bytes"]
+
+                if val.get("is_logging") == "False": issues.append("TRAIL_STOPPED")
+                if val.get("cw_logs_retention_days") == "Never Expire": issues.append("CW_LOGS_NO_EXPIRY")
+
+        issue_str = ", ".join(issues) if issues else "COMPLIANT"
         print(f"{acc_id:<15} | {summary:<25} | {issue_str}")
 
+    # Summary Section
+    s3_gb = org_stats["s3_bytes"] / (1024**3)
+    cw_gb = org_stats["cw_logs_size"] / (1024**3)
+    print("-" * 130)
+    print(f"ORGANIZATION LOGGING FOOTPRINT (CLOUDTRAIL):")
+    print(f"  Total S3 Storage:        {s3_gb:.2f} GB ({org_stats['s3_objects']:,} objects)")
+    print(f"  Total CW Logs Storage:   {cw_gb:.2f} GB ({org_stats['cw_logs_count']} log groups)")
+    print("-" * 130)
+
 if __name__ == "__main__":
     main()

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

@@ -6,68 +6,56 @@
 import sys
 
 # --- VERSIONING ---
-__version__ = "1.0.2"
+__version__ = "1.0.3"
 
 def main():
     parser = argparse.ArgumentParser(description=f"AWS Config Audit Assessor v{__version__}")
-    parser.add_argument("--input", required=True, help="The JSON audit file from check_config")
-    parser.add_argument("--central-bucket-regex", required=True, help="Regex to validate corporate S3 standards")
+    parser.add_argument("--input", required=True, help="JSON audit file from check_config")
+    parser.add_argument("--central-bucket-regex", required=True, help="Regex for corporate S3 standards")
     args = parser.parse_args()
 
     try:
         with open(args.input, 'r') as f:
             data = json.load(f)
     except Exception as e:
-        print(f"Error reading input file: {e}")
-        sys.exit(1)
+        print(f"Error: {e}"); sys.exit(1)
 
-    print("-" * 110)
-    print(f"{'Account ID':<15} | {'Alias':<20} | {'Global Status':<12} | {'S3 Compliance'}")
-    print("-" * 110)
+    print("-" * 125)
+    print(f"{'Account ID':<15} | {'Global Status':<12} | {'S3 Compliance'}")
+    print("-" * 125)
 
-    total_accounts = len(data)
-    non_compliant_s3 = 0
-    duplicate_globals = 0
+    total_stats = {"objects": 0, "size_bytes": 0, "non_central_buckets": 0, "accounts": len(data)}
+    unique_non_central = set()
 
     for account in data:
-        acc_id = account.get("account_id", "Unknown")
-        alias = account.get("alias", "N/A")
+        acc_id = account.get("account_id")
         checks = account.get("data", {})
-
-        # 1. Evaluate Global Recording Summary
-        summary_record = checks.get("account_summary", {})
-        global_status = summary_record.get("_summary", "MISSING")
+        summary = checks.get("account_summary", {}).get("_summary", "UNKNOWN")
 
-        if "MULTIPLE" in global_status:
-            duplicate_globals += 1
-
-        # 2. Evaluate S3 Regional Compliance
         s3_issues = []
         for reg, reg_data in checks.items():
-            if reg == "account_summary":
-                continue
+            if reg == "account_summary": continue
 
             bucket = reg_data.get("s3_bucket", "N/A")
+            total_stats["objects"] += reg_data.get("object_count", 0)
+            total_stats["size_bytes"] += reg_data.get("bucket_size_bytes", 0)
+
             if bucket != "N/A":
-                is_valid = bool(re.match(args.central_bucket_regex, bucket))
-                if not is_valid:
+                if not re.match(args.central_bucket_regex, bucket):
                     s3_issues.append(f"{reg}:{bucket}")
+                    unique_non_central.add(bucket)
 
-        # Text-only compliance status (removed emojis)
-        if s3_issues:
-            non_compliant_s3 += 1
-            s3_str = "NON_COMPLIANT: " + ", ".join(s3_issues)
-        else:
-            s3_str = "COMPLIANT"
-
-        print(f"{acc_id:<15} | {alias[:20]:<20} | {global_status:<12} | {s3_str}")
-
-    print("-" * 110)
-    print(f"ASSESSMENT SUMMARY:")
-    print(f"  Total Accounts Scanned: {total_accounts}")
-    print(f"  Accounts with Duplicate Global Recording: {duplicate_globals}")
-    print(f"  Accounts with Non-Compliant S3 Buckets:   {non_compliant_s3}")
-    print("-" * 110)
+        s3_status = "NON_COMPLIANT" if s3_issues else "COMPLIANT"
+        print(f"{acc_id:<15} | {summary:<12} | {s3_status}")
+
+    # Summary Section
+    size_gb = total_stats["size_bytes"] / (1024**3)
+    print("-" * 125)
+    print(f"ORGANIZATION STORAGE SUMMARY (CONFIG):")
+    print(f"  Total S3 Objects:        {total_stats['objects']:,}")
+    print(f"  Total S3 Storage:        {size_gb:.2f} GB")
+    print(f"  Non-Central Buckets:     {len(unique_non_central)}")
+    print("-" * 125)
 
 if __name__ == "__main__":
     main()