From 77ba4d3b7ce878ba3ba89e922de958df1e3b935e Mon Sep 17 00:00:00 2001 From: badra001 Date: Fri, 2 Jan 2026 09:30:45 -0500 Subject: [PATCH] update to summarize sizes --- .../assess_check_cloudtrail.py | 57 +++++++++------- .../cross-organization/assess_check_config.py | 68 ++++++++----------- 2 files changed, 59 insertions(+), 66 deletions(-) diff --git a/local-app/python-tools/cross-organization/assess_check_cloudtrail.py b/local-app/python-tools/cross-organization/assess_check_cloudtrail.py index 31e14e66..f00540ae 100644 --- a/local-app/python-tools/cross-organization/assess_check_cloudtrail.py +++ b/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,9 +15,11 @@ 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") @@ -24,32 +27,34 @@ def main(): 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() diff --git a/local-app/python-tools/cross-organization/assess_check_config.py b/local-app/python-tools/cross-organization/assess_check_config.py index cdcafa79..80601071 100755 --- a/local-app/python-tools/cross-organization/assess_check_config.py +++ b/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()