diff --git a/local-app/python-tools/cross-organization/assess_check_ecr.py b/local-app/python-tools/cross-organization/assess_check_ecr.py index 82b16897..595ec9f5 100755 --- a/local-app/python-tools/cross-organization/assess_check_ecr.py +++ b/local-app/python-tools/cross-organization/assess_check_ecr.py @@ -5,10 +5,11 @@ from collections import Counter, defaultdict # --- VERSIONING --- -__version__ = "1.3.2" +__version__ = "1.3.3" def find_latest_file(pattern): - files = glob.glob(pattern) + """Locates the most recent ECR audit JSON file.""" + files = glob.glob("audit_results.check_ecr.*.json") return max(files, key=os.path.getctime) if files else None def get_days_ago(iso_str): @@ -29,7 +30,7 @@ def bucket_age(days, counters): else: counters['<30'] += 1 def main(): - parser = argparse.ArgumentParser(description="ECR Full Spectrum Assessor - v1.3.2") + parser = argparse.ArgumentParser(description="ECR Full Spectrum Assessor - v1.3.3") parser.add_argument("--input", help="JSON audit file") args = parser.parse_args() @@ -54,7 +55,6 @@ def main(): "org_vulns": Counter(), "scanned_imgs": 0 } - # Tracking for duplicates using a Counter for digests digest_tracker = Counter() for idx, account in enumerate(data, 1): @@ -68,31 +68,34 @@ def main(): stats["mut"][repo_mut] += 1 if val.get("has_lifecycle") == "False": stats["no_lc"] += 1 - repo_size = val.get("repo_size_bytes", 0) - stats["total_bytes"] += repo_size - stats["region_bytes"][region] += repo_size + # Tracking per-repo size for the table display + repo_size_accum = 0 repo_vulns = Counter() images = val.get("images", []) stats["total_images"] += len(images) for img in images: - # Security - if isinstance(img, list) and len(img) > 0: - img = img[0] + if isinstance(img, list) and len(img) > 0: img = img[0] - # ADD: Record digest for duplicate checking + # RESTORED: Storage Aggregation logic + img_size = img.get("size_bytes", 0) + stats["total_bytes"] += img_size + stats["region_bytes"][region] += img_size + repo_size_accum += img_size + + # Digest for duplicate checking digest = img.get("image_digest") - if digest: - digest_tracker[digest] += 1 + if digest: digest_tracker[digest] += 1 + # Security & Aging counts = img.get("severity_counts", {}) if counts: stats["scanned_imgs"] += 1 for sev, count in counts.items(): repo_vulns[sev] += count stats["org_vulns"][sev] += count - # Aging + p_days = get_days_ago(img.get("pushed_at")) if p_days is not None: bucket_age(p_days, stats["push_ages"]) @@ -100,9 +103,10 @@ def main(): stats["push_count"] += 1 bucket_age(get_days_ago(img.get("last_pulled_at")), stats["pull_ages"]) - print(f"{idx:<4} | {account['account_id']:<15} | {region:<12} | {val['repo_name']:<40} | {repo_size/(1024**3):<10.2f} | {repo_mut:<11} | {'YES' if val.get('has_lifecycle')=='True' else 'NO':<7} | {repo_vulns['CRITICAL']:<8} | {repo_vulns['HIGH']:<8}") + print(f"{idx:<4} | {account['account_id']:<15} | {region:<12} | {val['repo_name']:<40} | {repo_size_accum/(1024**3):<10.2f} | {repo_mut:<11} | {'YES' if val.get('has_lifecycle')=='True' else 'NO':<7} | {repo_vulns['CRITICAL']:<8} | {repo_vulns['HIGH']:<8}") - # Footers + # Aggregated Summary Calculations + total_gb = stats["total_bytes"] / (1024**3) avg_img_mb = (stats["total_bytes"] / stats["total_images"]) / (1024**2) if stats["total_images"] > 0 else 0 avg_push_age = stats["total_push_days"] / stats["push_count"] if stats["push_count"] > 0 else 0 @@ -110,22 +114,17 @@ def main(): print(f"ORGANIZATION ECR SUMMARY\n") print(f" --- Config & Storage ---") print(f" Repos: {stats['total_repos']} ({stats['mut']['IMMUTABLE']} Immutable / {stats['mut']['MUTABLE']} Mutable) | Missing Lifecycle: {stats['no_lc']}") - print(f" Total Data: {stats['total_bytes']/(1024**3):.2f} GB | Average Image Size: {avg_img_mb:.2f} MB\n") + print(f" Total Data: {total_gb:.2f} GB | Average Image Size: {avg_img_mb:.2f} MB\n") print(f" --- Security & Vulnerabilities ---") print(f" Scan Coverage: {stats['scanned_imgs']} of {stats['total_images']} images scanned") print(f" Org Totals: " + " | ".join([f"{s}: {stats['org_vulns'][s]:,}" for s in ["CRITICAL", "HIGH", "MEDIUM", "LOW"]])) - # ADD: Duplicate Image Audit Summary + # Duplicate Image Audit duplicates = {d: c for d, c in digest_tracker.items() if c > 1} print(f"\n --- Duplicate Image Audit ---") print(f" Total Unique Digests: {len(digest_tracker)}") print(f" Total Duplicated Image Instances: {sum(duplicates.values())} (across {len(duplicates)} unique digests)") - if duplicates: - print(f" {'Digest':<75} | {'Instance Count'}") - print(f" {'-'*75} | {'-'*14}") - for d, c in sorted(duplicates.items(), key=lambda x: x[1], reverse=True): - print(f" {d:<75} | {c}") print(f"\n --- Age Distribution (Days since Push vs Pull) ---") print(f" Average Image Age: {avg_push_age:.1f} days")