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 f00540ae..4da4cc49 100644 --- a/local-app/python-tools/cross-organization/assess_check_cloudtrail.py +++ b/local-app/python-tools/cross-organization/assess_check_cloudtrail.py @@ -1,60 +1,65 @@ #!/usr/bin/env python +#!/usr/bin/env python3 import json import argparse +import re import sys # --- VERSIONING --- -__version__ = "1.0.1" +__version__ = "1.0.3" def main(): - parser = argparse.ArgumentParser(description="AWS CloudTrail Audit Assessor") - parser.add_argument("--input", required=True, help="JSON file from check_cloudtrail") + parser = argparse.ArgumentParser(description="AWS Config Audit Assessor") + 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() - with open(args.input, 'r') as f: - data = json.load(f) + try: + with open(args.input, 'r') as f: + data = json.load(f) + except Exception as e: + print(f"Error: {e}"); sys.exit(1) - print("-" * 130) - print(f"{'Account ID':<15} | {'Global Summary':<25} | {'Security/Cost Issues'}") - print("-" * 130) + print("-" * 125) + print(f"{'Account ID':<15} | {'Global Status':<12} | {'S3 Compliance'}") + print("-" * 125) - org_stats = {"s3_objects": 0, "s3_bytes": 0, "cw_logs_size": 0, "cw_logs_count": 0} + stats = {"objects": 0, "bytes": 0, "total_accounts": len(data)} + unique_buckets = set() + non_central_buckets = set() for account in data: acc_id = account.get("account_id") checks = account.get("data", {}) summary = checks.get("account_summary", {}).get("_summary", "UNKNOWN") - issues = [] - if "DUPLICATIVE" in summary: - issues.append("DUPLICATE_LOGGING_COSTS") - - for key, val in checks.items(): - if key.startswith("trail:"): - # S3 Accumulation - org_stats["s3_objects"] += val.get("object_count", 0) - org_stats["s3_bytes"] += val.get("bucket_size_bytes", 0) + s3_issues = [] + for reg, reg_data in checks.items(): + if reg == "account_summary": continue + + bucket = reg_data.get("s3_bucket", "N/A") + if bucket != "N/A": + unique_buckets.add(bucket) + stats["objects"] += reg_data.get("object_count", 0) + stats["bytes"] += reg_data.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}") + if not re.match(args.central_bucket_regex, bucket): + s3_issues.append(bucket) + non_central_buckets.add(bucket) + + s3_status = f"NON_COMPLIANT ({len(s3_issues)} regions)" if s3_issues else "COMPLIANT" + print(f"{acc_id:<15} | {summary:<12} | {s3_status}") # 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) + size_gb = stats["bytes"] / (1024**3) + print("-" * 125) + print(f"ORGANIZATION STORAGE SUMMARY (CONFIG):") + print(f" Total S3 Storage: {size_gb:.2f} GB") + print(f" Total S3 Objects: {stats['objects']:,}") + print(f" Total S3 Buckets: {len(unique_buckets)}") + print(f" Non-Central Buckets: {len(non_central_buckets)}") + print("-" * 125) if __name__ == "__main__": main()