diff --git a/scanner_py/cli/generate_report.py b/scanner_py/cli/generate_report.py index aad29f2..d9982ba 100644 --- a/scanner_py/cli/generate_report.py +++ b/scanner_py/cli/generate_report.py @@ -316,7 +316,10 @@ def generate_markdown_report( images_with_chainguard = 0 displayed_count = 0 - for analysis in cve_analysis: + for analysis in sorted( + cve_analysis, + key=lambda a: a.get("image", "").rsplit("/", 1)[-1].lower(), + ): # Get CVE counts critical = analysis.get("critical", 0) high = analysis.get("high", 0) @@ -356,10 +359,14 @@ def generate_markdown_report( displayed_count += 1 - # Format image name - image_short = analysis["image"].split("/")[-1] - if len(image_short) > 40: - image_short = image_short[:37] + "..." + # Show only the image basename (`:`); strip registry + # host and project segments. The project prefix is identical for + # most rows in a typical chart-wide scan and adds noise without + # signal. Truncate very long refs (> 50 chars) by clipping the + # tag end. Caller is responsible for unique basenames. + image_short = analysis["image"].rsplit("/", 1)[-1] + if len(image_short) > 50: + image_short = image_short[:47] + "..." # Format cells unaddr_str = f"✅ {unaddressed}" if unaddressed == 0 else f"🔴 **{unaddressed}**" @@ -536,7 +543,10 @@ def print_cli_summary( images_with_triage = 0 images_with_chainguard = 0 - for analysis in cve_analysis: + for analysis in sorted( + cve_analysis, + key=lambda a: a.get("image", "").rsplit("/", 1)[-1].lower(), + ): critical = analysis.get("critical", 0) high = analysis.get("high", 0) medium = analysis.get("medium", 0) diff --git a/scanner_py/cli/generate_sbom_report.py b/scanner_py/cli/generate_sbom_report.py index 88ea713..c3851ca 100644 --- a/scanner_py/cli/generate_sbom_report.py +++ b/scanner_py/cli/generate_sbom_report.py @@ -50,6 +50,14 @@ def create_generate_sbom_report_parser(subparsers: Any) -> argparse.ArgumentPars default="sbom-detailed-report.md", help="Output file path (default: sbom-detailed-report.md)", ) + parser.add_argument( + "--source-label", + help=( + "Logical label for the scan source shown in the report header " + "(overrides the value read from scan-summary.json). Use this when " + "scanning a static image list that is not backed by a K8s namespace." + ), + ) parser.add_argument( "--verbose", action="store_true", @@ -161,6 +169,9 @@ def generate_report( namespace: str = "pharia-ai", ) -> bool: """Generate detailed SBOM report.""" + # Sort alphabetically so the per-image sections render in a stable order + # regardless of scan-completion order. + successful_scans = sorted(successful_scans, key=lambda s: s.lower()) total_images = len(successful_scans) # Calculate overall statistics @@ -196,7 +207,7 @@ def generate_report( # Header f.write("# Detailed SBOM Analysis Report\n\n") f.write(f"**Generated:** {datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')}\n") - f.write(f"**Source:** Successful SBOM scans from {namespace} namespace\n") + f.write(f"**Image source:** `{namespace}`\n") f.write(f"**Total Images Analyzed:** {total_images}\n\n") f.write("---\n\n") f.write("## Executive Summary\n\n") @@ -403,8 +414,11 @@ def run_generate_sbom_report(args: argparse.Namespace) -> int: logger.error("No successful scans found in scan summary") return 1 - # Get namespace - namespace = summary.get("scan_summary", {}).get("namespace", "pharia-ai") + # Resolve label: explicit --source-label wins over scan-summary.json. + namespace = ( + getattr(args, "source_label", None) + or summary.get("scan_summary", {}).get("namespace", "pharia-ai") + ) # Generate report if generate_report(scan_results_dir, output_file, successful_scans, namespace): diff --git a/scanner_py/cli/k8s_scanner.py b/scanner_py/cli/k8s_scanner.py index a9d52dc..14899d4 100644 --- a/scanner_py/cli/k8s_scanner.py +++ b/scanner_py/cli/k8s_scanner.py @@ -21,6 +21,8 @@ logger = get_logger(__name__) +TRIVY_DB_REPOSITORY = "ghcr.io/aquasecurity/trivy-db:2" + def prepare_trivy_db(verbose: bool = False) -> bool: """ @@ -31,7 +33,7 @@ def prepare_trivy_db(verbose: bool = False) -> bool: Steps: 1. trivy clean --all (clean existing db) - 2. trivy image --download-db-only (download fresh db) + 2. trivy image --download-db-only --db-repository ... (download fresh db) Returns: True if successful @@ -52,9 +54,17 @@ def prepare_trivy_db(verbose: bool = False) -> bool: # Step 2: Download fresh database if verbose: - logger.info("Downloading Trivy vulnerability database...") + logger.info( + f"Downloading Trivy vulnerability database from {TRIVY_DB_REPOSITORY}..." + ) - download_args = ["trivy", "image", "--download-db-only"] + download_args = [ + "trivy", + "image", + "--download-db-only", + "--db-repository", + TRIVY_DB_REPOSITORY, + ] if not verbose: download_args.append("--quiet") @@ -120,6 +130,14 @@ def create_k8s_scanner_parser(subparsers: Any) -> argparse.ArgumentParser: default="pharia-ai", help="Kubernetes namespace to scan (default: pharia-ai)", ) + parser.add_argument( + "--source-label", + help=( + "Logical label for the scan source shown in the report header " + "(overrides --namespace / --image-file basename). Use this when " + "scanning a static image list that is not backed by a K8s namespace." + ), + ) parser.add_argument( "--kubeconfig", help="Path to kubeconfig file", @@ -613,7 +631,15 @@ def generate_summary( failed = [r for r in results if not r.success and not r.skipped] skipped = [r for r in results if r.skipped] - namespace_label = args.image_file if args.image_file else args.namespace + # Prefer explicit --source-label, else the basename of --image-file, else --namespace. + # Using the full --image-file path as a label produced unreadable "Namespace" + # values like "/home/runner/work/.../images-harbor.txt" in the step summary. + if getattr(args, "source_label", None): + namespace_label = args.source_label + elif args.image_file: + namespace_label = Path(args.image_file).name + else: + namespace_label = args.namespace summary = ScanSummary( namespace=namespace_label, total_images_found=extraction_result.total_found, @@ -702,7 +728,7 @@ def generate_markdown_summary(summary: ScanSummary, min_cve_level: str) -> str: lines.append("") lines.append(f"| Metric | Value |") lines.append("|--------|-------|") - lines.append(f"| **Namespace** | `{summary.namespace}` |") + lines.append(f"| **Image source** | `{summary.namespace}` |") lines.append(f"| **Images Found** | {summary.total_images_found} |") lines.append(f"| **Images Processed** | {summary.images_processed} |") lines.append(f"| **Successful Scans** | ✅ {summary.successful_scans} |") @@ -742,11 +768,19 @@ def generate_markdown_summary(summary: ScanSummary, min_cve_level: str) -> str: images_with_triage = 0 images_with_chainguard = 0 - for analysis in summary.cve_analysis: - image_short = analysis["image"].split("/")[-1] - # Truncate if too long - if len(image_short) > 40: - image_short = image_short[:37] + "..." + for analysis in sorted( + summary.cve_analysis, + key=lambda a: a.get("image", "").rsplit("/", 1)[-1].lower(), + ): + # Show only the image basename (`:`); strip the registry + # host and project segments. Reports list 30-40 images at a time and + # the project prefix is the same for almost all of them, so it adds + # noise without adding signal. Truncate very long refs (> 50 chars) + # by clipping the tag end. Caller is responsible for ensuring + # basenames are unique across the input image set. + image_short = analysis["image"].rsplit("/", 1)[-1] + if len(image_short) > 50: + image_short = image_short[:47] + "..." # Get CVE counts critical = analysis.get("critical", 0) @@ -940,8 +974,11 @@ def print_summary(summary: ScanSummary, min_cve_level: str, verbose: bool = Fals images_with_triage = 0 images_with_chainguard = 0 - for analysis in summary.cve_analysis: - image_short = analysis["image"].split("/")[-1][:33] + for analysis in sorted( + summary.cve_analysis, + key=lambda a: a.get("image", "").rsplit("/", 1)[-1].lower(), + ): + image_short = analysis["image"].rsplit("/", 1)[-1][:33] # Get CVE counts critical = analysis.get("critical", 0) diff --git a/scanner_py/cli/oras_scan.py b/scanner_py/cli/oras_scan.py index 5a9ff61..9de34c0 100644 --- a/scanner_py/cli/oras_scan.py +++ b/scanner_py/cli/oras_scan.py @@ -14,10 +14,14 @@ """ import argparse +import atexit import json +import os import re import shutil import sys +import tempfile +import threading from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field from datetime import datetime @@ -30,6 +34,95 @@ logger = get_logger(__name__) +TRIVY_DB_REPOSITORY = "ghcr.io/aquasecurity/trivy-db:2" + + +# --------------------------------------------------------------------------- +# Per-worker Trivy cache pool. +# +# Trivy's default `fs` cache backend uses BoltDB and holds an exclusive lock +# on `/fanal/cache.db` for the duration of every scan. Pointing all +# parallel workers at the same cache dir therefore serialises them - the +# unlucky ones time out with: +# +# unable to initialize fs cache: cache may be in use by another process +# +# We avoid the contention by giving every worker thread its own cache dir, +# pre-populated by copying the immutable vuln-DB files (`db/trivy.db`, +# `db/metadata.json`, optional `java-db/`) that `prepare_trivy_db()` has +# already downloaded into the shared default cache. Combined with +# `--skip-db-update`, this means: no DB redownloads, no lock contention. +# +# The pool is bounded by `args.parallel` workers (typically 5-15), so peak +# disk usage is ~N x DB-size (~30-50 MB per copy) which is comfortable on a +# CI runner. +# --------------------------------------------------------------------------- + +_TRIVY_DB_SRC: Optional[Path] = None +_THREAD_CACHE = threading.local() +_CACHE_DIRS_LOCK = threading.Lock() +_CACHE_DIRS: List[Path] = [] + + +def _trivy_default_cache_dir() -> Path: + """ + Return the cache directory Trivy would use by default. + + Honors $TRIVY_CACHE_DIR, then $XDG_CACHE_HOME/trivy, then ~/.cache/trivy + (matches Trivy's own resolution order). + """ + env_dir = os.environ.get("TRIVY_CACHE_DIR") + if env_dir: + return Path(env_dir) + xdg = os.environ.get("XDG_CACHE_HOME") + if xdg: + return Path(xdg) / "trivy" + return Path.home() / ".cache" / "trivy" + + +def _get_worker_cache_dir() -> Path: + """ + Return a per-thread Trivy cache dir, populated lazily on first call. + + Each worker thread gets its own writable cache directory containing a + copy of the vuln DB so that Trivy's fanal/cache.db lock cannot collide + across workers. + """ + if _TRIVY_DB_SRC is None: + raise RuntimeError( + "prepare_trivy_db() must be called before run_trivy_scan()" + ) + + cache_dir: Optional[Path] = getattr(_THREAD_CACHE, "cache_dir", None) + if cache_dir is not None: + return cache_dir + + cache_dir = Path(tempfile.mkdtemp(prefix="trivy-cache-")) + + src_db = _TRIVY_DB_SRC / "db" + if src_db.is_dir(): + shutil.copytree(src_db, cache_dir / "db") + + src_java = _TRIVY_DB_SRC / "java-db" + if src_java.is_dir(): + shutil.copytree(src_java, cache_dir / "java-db") + + with _CACHE_DIRS_LOCK: + _CACHE_DIRS.append(cache_dir) + _THREAD_CACHE.cache_dir = cache_dir + return cache_dir + + +def _cleanup_trivy_cache_dirs() -> None: + """Remove every per-worker Trivy cache dir at process exit.""" + with _CACHE_DIRS_LOCK: + for cache_dir in _CACHE_DIRS: + shutil.rmtree(cache_dir, ignore_errors=True) + _CACHE_DIRS.clear() + + +atexit.register(_cleanup_trivy_cache_dirs) + @dataclass class ImageScanResult: @@ -287,19 +380,23 @@ def sanitize_filename(image: str) -> str: def prepare_trivy_db(verbose: bool = False) -> bool: """ - Prepare Trivy database by cleaning and downloading fresh DB. + Prepare Trivy database by cleaning and downloading a fresh DB into + Trivy's default cache directory. - This should be run once before parallel scans to avoid race conditions - and ensure all scans use the same database version. + The fresh DB is later copied into per-worker cache directories by + `_get_worker_cache_dir()`, so all parallel workers reuse the same + DB version without redownloading and without sharing a lock. Steps: 1. trivy clean --all (clean existing db) - 2. trivy image --download-db-only (download fresh db) + 2. trivy image --download-db-only --db-repository ... (download fresh db) + 3. record the cache location so workers can copy from it Returns: True if successful """ - # Step 1: Clean existing database + global _TRIVY_DB_SRC + if verbose: logger.info("Cleaning existing Trivy database...") @@ -313,11 +410,18 @@ def prepare_trivy_db(verbose: bool = False) -> bool: logger.warning(f"Failed to clean Trivy cache (may not exist): {clean_result.stderr}") # Continue anyway - might be first run - # Step 2: Download fresh database if verbose: - logger.info("Downloading Trivy vulnerability database...") + logger.info( + f"Downloading Trivy vulnerability database from {TRIVY_DB_REPOSITORY}..." + ) - download_args = ["trivy", "image", "--download-db-only"] + download_args = [ + "trivy", + "image", + "--download-db-only", + "--db-repository", + TRIVY_DB_REPOSITORY, + ] if not verbose: download_args.append("--quiet") @@ -330,8 +434,17 @@ def prepare_trivy_db(verbose: bool = False) -> bool: logger.error(f"Failed to download Trivy database: {download_result.stderr}") return False + src = _trivy_default_cache_dir() + if not (src / "db" / "trivy.db").is_file(): + logger.error( + f"Trivy DB download succeeded but trivy.db not found under {src}/db; " + "per-worker cache pool cannot be primed." + ) + return False + _TRIVY_DB_SRC = src + if verbose: - logger.info("Trivy database ready") + logger.info(f"Trivy database ready (source cache: {src})") return True @@ -347,14 +460,26 @@ def run_trivy_scan( Equivalent to: trivy image --scanners vuln --format json "$image" - Note: Uses --skip-db-update since DB is pre-downloaded by prepare_trivy_db() + Uses a per-worker `--cache-dir` (lazy-initialized via + `_get_worker_cache_dir()`) that has been pre-populated by copying the + vuln DB downloaded once by `prepare_trivy_db()`. Combined with + `--skip-db-update`, this gives us: + + * No DB redownloads (the previous per-image `--cache-dir` without any + seeding made every worker redownload the entire vuln DB - the + dominant wall-clock cost on larger image sets). + * No BoltDB lock contention (sharing the default cache across workers + made Trivy fail with "fs cache may be in use by another process"). Returns: Tuple of (success, error_message) """ + cache_dir = _get_worker_cache_dir() + args = [ "trivy", "image", - "--cache-dir", f"/tmp/trivy-cache-{image.replace('/', '_').replace(':', '_')}", + "--cache-dir", str(cache_dir), + "--skip-db-update", "--scanners", "vuln", "--format", "json", "--output", output_file, @@ -648,6 +773,10 @@ def generate_markdown_report( Equivalent to 3-gen-report.sh but with enhanced formatting. """ lines = [] + # Sort alphabetically by image basename so the rendered order matches the + # displayed `:` cells (the project prefix is stripped at render + # time). Sorting is stable across runs regardless of scan-completion order. + results = sorted(results, key=lambda r: r.image.rsplit("/", 1)[-1].lower()) successful = [r for r in results if r.success] failed = [r for r in results if not r.success] @@ -674,7 +803,7 @@ def generate_markdown_report( lines.append("") lines.append("| Metric | Value |") lines.append("|--------|------:|") - lines.append(f"| **Namespace** | `{namespace}` |") + lines.append(f"| **Image source** | `{namespace}` |") lines.append(f"| **Scan Date** | {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC |") lines.append(f"| **Minimum CVE Level** | `{min_cve_level}` |") lines.append(f"| **Total Images** | {len(successful)} |") @@ -736,8 +865,11 @@ def generate_markdown_report( displayed_count += 1 - # Format image name (truncate if too long) - image_name = result.image_ref + # Show only the image basename (`:`); strip the project + # segment from result.image_ref. Reports list 30-40 images at a time + # and the project prefix is the same for almost all of them, so it + # adds noise without signal. Truncate very long refs (> 45 chars). + image_name = result.image_ref.rsplit("/", 1)[-1] if len(image_name) > 45: image_name = image_name[:42] + "..." @@ -811,6 +943,7 @@ def print_cli_summary( min_cve_level: str = "MEDIUM", ) -> None: """Print beautiful summary to CLI matching the markdown report format.""" + results = sorted(results, key=lambda r: r.image.rsplit("/", 1)[-1].lower()) successful = [r for r in results if r.success] failed = [r for r in results if not r.success] @@ -901,8 +1034,8 @@ def print_cli_summary( displayed += 1 - # Truncate image name - image_name = result.image_ref + # Show only the basename (`:`); strip project segment. + image_name = result.image_ref.rsplit("/", 1)[-1] if len(image_name) > 43: image_name = image_name[:40] + "..." diff --git a/scanner_py/cli/retrieve_triage.py b/scanner_py/cli/retrieve_triage.py index c5c2dd5..bf53938 100644 --- a/scanner_py/cli/retrieve_triage.py +++ b/scanner_py/cli/retrieve_triage.py @@ -267,7 +267,16 @@ def dir_to_image_ref(dir_name: str) -> str: rest = "_".join(parts[:-1]) return rest.replace("_", "/") + ":" + tag - subdirs = sorted(d for d in triage_dir.iterdir() if d.is_dir()) + def _sort_key(d: Path) -> str: + # Sort by the image basename (last path segment of the derived ref) + # rather than the full sanitized directory name. Sorting by the raw + # sanitized path puts everything under the same registry prefix + # together and makes the report order effectively arbitrary from a + # reader's point of view. + ref = dir_to_image_ref(d.name) + return ref.rsplit("/", 1)[-1].lower() + + subdirs = sorted((d for d in triage_dir.iterdir() if d.is_dir()), key=_sort_key) for subdir in subdirs: triage_json = subdir / "triage.json" triage_toml = subdir / "triage.toml" diff --git a/scanner_py/core/attestation.py b/scanner_py/core/attestation.py index 7910aca..2a922d8 100644 --- a/scanner_py/core/attestation.py +++ b/scanner_py/core/attestation.py @@ -10,10 +10,11 @@ import json import base64 import re +import sys import tempfile from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field -from typing import Optional, List, Dict, Any, Set +from typing import Optional, List, Dict, Any, Set, Tuple from pathlib import Path from enum import Enum @@ -60,6 +61,24 @@ class AttestationList: """List of available attestations.""" attestations: Dict[str, int] = field(default_factory=dict) + # ``True`` when ``list_attestations`` could not enumerate referrers + # conclusively: either the OCI 1.1 Referrers API timed out, or + # bundle decode failed for enough entries that the signature + # predicate might have been suppressed under high parallel load. + # Exposed for diagnostics (``detect_attestation_type`` logs it + # alongside UNSIGNED classifications) so CI runs where a signed + # image is misclassified can be distinguished from truly unsigned + # images without toggling --verbose globally. + discovery_failed: bool = False + + # Predicate type emitted by ``cosign sign`` (the image-signature DSSE + # statement), used to detect that an image is signed without invoking + # ``cosign verify`` against the registry. This is the same payload that + # cosign verify-attestation downloads as the signature half of a + # sigstore bundle, so its mere presence in the image's OCI 1.1 referrers + # list is sufficient evidence of a cosign signature. + COSIGN_SIGNATURE_PREDICATE = "https://sigstore.dev/cosign/sign/v1" + def has_sbom(self) -> bool: """Check if SBOM attestation exists.""" sbom_types = [ @@ -72,6 +91,19 @@ def has_triage(self) -> bool: """Check if triage attestation exists.""" return PREDICATE_TYPE_MAP[AttestationTypeEnum.TRIAGE] in self.attestations + def has_signature(self) -> bool: + """Check if a cosign signature DSSE statement is present. + + The image-signature side of a cosign-signed image is a DSSE + envelope with predicateType ``https://sigstore.dev/cosign/sign/v1``, + stored as a sigstore bundle in the OCI 1.1 referrers list. We use + its presence as proof the image is signed -- avoiding a separate + ``cosign verify`` round trip which on Harbor enumerates the + Referrers API a second time (linear in referrer count and the + primary cause of long-tail timeouts on heavily-attested images). + """ + return self.COSIGN_SIGNATURE_PREDICATE in self.attestations + class AttestationExtractor: """ @@ -103,6 +135,16 @@ def __init__( self.certificate_oidc_issuer = certificate_oidc_issuer self.certificate_identity_regexp = certificate_identity_regexp self.timeout = timeout + # Per-instance cache populated by ``list_attestations`` and + # consulted by ``_find_attestation_digests``. Maps + # ``(image_name, image_digest)`` -> ``{predicate_type: [bundle_digests]}`` + # so that ``extract_sbom`` / ``extract_triage`` (which would + # otherwise serially re-decode every sigstore bundle on the + # image) can reuse the parallel decode work + # ``ImageScanner.detect_attestation_type`` has already paid for. + self._predicate_digests_cache: Dict[ + Tuple[str, str], Dict[str, List[str]] + ] = {} def resolve_digest(self, image: str) -> Optional[str]: """ @@ -139,10 +181,29 @@ def list_attestations( logger.debug(f"Listing attestations for: {image} ({digest})") - referrers = self._discover_referrers(image, digest) + # Use the tristate variant so a discovery failure (timeout, network + # blip, registry without referrers support) is distinguishable from + # a definitive "no referrers". The legacy ``_discover_referrers`` + # alias collapses both into ``[]`` and would silently misclassify a + # signed image as ``UNSIGNED`` if the registry was momentarily slow. + referrers = self.discover_referrers(image, digest) + if referrers is None: + # Discovery itself failed (timeout, network blip, registry + # without Referrers API). Flag the result. Print directly to + # stderr so the diagnostic survives the ``suppress_logging`` + # call that parallel-scan ProgressBar wraps around every + # worker (which would swallow any logger.info/error output). + print( + f"[cosign-scan] Referrers discovery failed for {image}" + f" — marking discovery_failed=True", + file=sys.stderr, + flush=True, + ) + return AttestationList(discovery_failed=True) if not referrers: - if is_verbose(): - logger.error(f"Failed to discover referrers for {image}") + # Definitive empty referrers list: image has no OCI 1.1 + # referrers attached. This is the fast-path for truly + # unsigned images. return AttestationList() bundle_type = "application/vnd.dev.sigstore.bundle.v0.3+json" @@ -154,31 +215,150 @@ def list_attestations( if not bundle_refs: return AttestationList() + # NOTE: cosign / Harbor emits the same + # ``dev.sigstore.bundle.predicateType = + # https://sigstore.dev/cosign/sign/v1`` annotation on every + # sigstore bundle manifest regardless of payload, so we still + # have to pull each bundle's manifest + blob to read the real + # in-toto predicate type from the DSSE envelope. + # + # That's 2 HTTP round-trips per bundle. Heavily-attested images + # (pharia-os-app has 400+ bundles) make a naive enumeration + # prohibitively slow even with parallelism. Two mitigations: + # + # 1. Run the decode in a 20-worker pool to absorb the per-call + # latency. + # 2. Early-terminate as soon as we have collected the predicate + # types ``ImageScanner.detect_attestation_type`` cares about + # (signature + SBOM + triage). Once all three are present, + # further bundles can only inflate counts, not change the + # classification, so we cancel pending futures and return. + # + # Empirically this brings classify latency for the heaviest known + # image from "doesn't finish in 20 minutes" down to ~5-15 s. predicate_types: Dict[str, int] = {} + # Bundle digests grouped by their decoded in-toto predicate type. + # Populated alongside ``predicate_types`` so the cache hit path in + # ``_find_attestation_digests`` (used by extract_sbom / + # extract_triage) can avoid re-decoding bundles we have already + # seen. + predicate_digests: Dict[str, List[str]] = {} image_name = self._strip_tag(image) - def fetch_predicate_type(ref: dict) -> Optional[str]: + # Predicates that change the cosign-scan classification. As soon + # as we have observed at least one bundle of each of these we can + # stop. SLSA / vuln / license bundles are not part of the + # classification path so they do not need to be fully enumerated. + decision_predicates = { + AttestationList.COSIGN_SIGNATURE_PREDICATE, + PREDICATE_TYPE_MAP[AttestationTypeEnum.CYCLONEDX], + PREDICATE_TYPE_MAP[AttestationTypeEnum.SPDX], + PREDICATE_TYPE_MAP[AttestationTypeEnum.TRIAGE], + } + + def have_decisive_set() -> bool: + seen = set(predicate_types) + has_sig = AttestationList.COSIGN_SIGNATURE_PREDICATE in seen + has_sbom = bool( + seen + & { + PREDICATE_TYPE_MAP[AttestationTypeEnum.CYCLONEDX], + PREDICATE_TYPE_MAP[AttestationTypeEnum.SPDX], + } + ) + has_triage = PREDICATE_TYPE_MAP[AttestationTypeEnum.TRIAGE] in seen + return has_sig and has_sbom and has_triage + + def fetch_predicate_type(ref: dict) -> Tuple[Optional[str], Optional[str]]: ref_digest = ref.get("digest") if not ref_digest: - return None - return self._get_predicate_type_from_bundle(image_name, ref_digest) + return None, None + return ref_digest, self._get_predicate_type_from_bundle( + image_name, ref_digest + ) - max_workers = min(len(bundle_refs), 5) + max_workers = min(len(bundle_refs), 20) + # Track decode outcomes so we can surface a diagnostic at INFO + # level when a high failure rate silently suppresses the + # signature bundle (the main reason classification randomly flips + # to UNSIGNED under 15x parallel scan load against Harbor). + decoded_ok = 0 + decoded_err = 0 + decoded_empty = 0 with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = { executor.submit(fetch_predicate_type, ref): ref for ref in bundle_refs } - for future in as_completed(futures): - try: - pred_type = future.result() - if pred_type: - predicate_types[pred_type] = predicate_types.get(pred_type, 0) + 1 - except Exception: - pass + try: + for future in as_completed(futures): + try: + ref_digest, pred_type = future.result() + except Exception: + decoded_err += 1 + continue + if not pred_type or not ref_digest: + decoded_empty += 1 + continue + decoded_ok += 1 + predicate_types[pred_type] = ( + predicate_types.get(pred_type, 0) + 1 + ) + predicate_digests.setdefault(pred_type, []).append( + ref_digest + ) + # Once we have at least one of every decision + # predicate, classifications are stable. Stop early. + if pred_type in decision_predicates and have_decisive_set(): + for pending in futures: + pending.cancel() + break + finally: + # Best-effort cancellation; ThreadPoolExecutor will join + # already-running workers on context exit. + for pending in futures: + pending.cancel() + + total_attempted = decoded_ok + decoded_err + decoded_empty + if total_attempted and decoded_err + decoded_empty > 0: + # stderr print: bypasses the logging suppression that wraps + # parallel scan workers. + print( + f"[cosign-scan] Bundle decode for {image}: ok={decoded_ok}" + f" err={decoded_err} empty={decoded_empty}" + f" (of {len(bundle_refs)} refs)", + file=sys.stderr, + flush=True, + ) - return AttestationList(attestations=predicate_types) + # Flag the result as unreliable when we enumerated sigstore + # bundles but never saw the signature predicate. This is + # specifically the CI failure mode where Harbor returns the + # referrers list but individual bundle fetches time out under + # 15x parallel scan load, so the signature bundle can be + # silently suppressed from the decoded map. Surfaced via the + # INFO log in ``detect_attestation_type`` so each instance + # becomes greppable in CI logs. + discovery_failed = ( + bool(bundle_refs) + and AttestationList.COSIGN_SIGNATURE_PREDICATE not in predicate_types + ) + + # Cache so downstream extract_sbom / extract_triage can reuse the + # bundle digests we already paid to decode. Key on + # ``(image_name, digest)`` to be tag-stable across the same + # registry digest. Mark partial enumerations explicitly so + # ``_find_attestation_digests`` can decide whether to fall back + # to a full enumeration when its requested predicate type is + # absent from the cache. + self._predicate_digests_cache[(image_name, digest)] = dict( + predicate_digests + ) + + return AttestationList( + attestations=predicate_types, discovery_failed=discovery_failed + ) def _get_predicate_type_from_bundle( self, image_name: str, ref_digest: str @@ -334,29 +514,59 @@ def _verify_attestation(self, image: str, pred_type: str) -> bool: result = run_with_timeout(args, self.timeout) return result.success - def _discover_referrers(self, image: str, image_digest: Optional[str] = None) -> List[dict]: + def discover_referrers( + self, image: str, image_digest: Optional[str] = None + ) -> Optional[List[dict]]: """ Discover OCI referrers for an image. Tries image@digest first for precision, falls back to image with tag if the registry doesn't support digest-based referrer discovery (e.g. JFrog). - Returns an empty list if the registry doesn't support the Referrers API - (timeout or error). Callers should use extract_triage_tag_based() as fallback. - - Registries that time out once are remembered (per host) so every other - image on the same registry skips the slow probe instead of re-paying the - timeout. This is the common case for JFrog, which has no Referrers API. + Returns: + * ``None`` if the discovery itself failed (timeout, network + error, JSON parse error, registry without Referrers API + support). Callers can use this to fall back to slower + alternatives such as ``cosign verify`` or + ``extract_triage_tag_based()``. + * ``[]`` if discovery succeeded and the image legitimately has + no referrers attached. Callers can rely on this as a + definitive "no attestations / no signatures" signal. + * a non-empty list of referrer manifests otherwise. + + Registries that time out once are remembered (per host) so every + other image on the same registry skips the slow probe instead of + re-paying the timeout. This is the common case for JFrog, which has + no Referrers API; such a registry returns ``None`` here (discovery + failed), not ``[]``. """ capabilities = get_registry_capability_cache() if capabilities.referrers_unsupported(image): logger.debug( f"Skipping Referrers API for {image} — registry already known to not support it" ) - return [] - - fast_timeout = min(self.timeout, 15) + return None + # The oras discover request hits the registry's OCI 1.1 Referrers + # API. On Harbor this is O(N) in referrer count -- empirically + # ~8 s for 36 refs locally, ~95 s for 403 refs locally, but + # observed 120+ s in GitHub-hosted runners against the same + # images (extra round-trip latency + TLS setup cost amplifies + # the per-referrer cost). 240 s covers the tail we have seen + # in CI while still bounding worker wall-clock; an individual + # cosign-scan image is still bounded by the 10-minute trivy + # budget after discovery completes. Capped at ``self.timeout`` + # so callers that supply a tighter budget (e.g. unit tests) + # keep control. + fast_timeout = min(self.timeout, 240) + + # Try digest-based discovery first (more precise: it avoids a + # second tag->digest round trip inside the registry) and fall + # back to tag-based on ANY failure, including timeout. A timeout + # on digest-based doesn't mean the Referrers API is unsupported; + # it usually means Harbor was slow on this particular request. + # Retrying via the tag path often succeeds because Harbor caches + # the tag->digest mapping separately. if image_digest: image_with_digest = f"{image}@{image_digest}" result = run_command( @@ -369,31 +579,64 @@ def _discover_referrers(self, image: str, image_digest: Optional[str] = None) -> refs = data.get("referrers", data.get("manifests", [])) return refs if refs is not None else [] except json.JSONDecodeError: - pass - - if result.timed_out: - capabilities.mark_referrers_unsupported(image) - logger.debug(f"Referrers API timed out for {image} — registry may not support it") - return [] - - logger.debug(f"Digest-based oras discover failed, falling back to tag: {result.stderr}") + logger.debug( + f"oras discover for {image} returned non-JSON output" + ) + return None + + # Log the failure reason (timeout / other error) at stderr + # so CI captures it; then fall through to the tag-based + # retry. The previous implementation bailed out on timeout, + # which bricked heavily-attested images whose digest-based + # request happened to be the slow one on a given run. + fail_reason = "timed out" if result.timed_out else "failed" + print( + f"[cosign-scan] Digest-based oras discover {fail_reason}" + f" for {image}; retrying via tag-based discovery", + file=sys.stderr, + flush=True, + ) result = run_command( ["oras", "discover", image, "--format", "json"], timeout=fast_timeout, ) if not result.success: + # Both digest- and tag-based discovery have now failed. A + # timeout on the final (tag-based) attempt is the strongest + # signal the registry has no usable Referrers API, so remember + # the host to spare every other image the same slow probe. if result.timed_out: capabilities.mark_referrers_unsupported(image) - logger.debug(f"Referrers API timed out for {image} — registry may not support it") - return [] + fail_reason = "timed out" if result.timed_out else "failed" + print( + f"[cosign-scan] Tag-based oras discover {fail_reason}" + f" for {image} after {fast_timeout}s", + file=sys.stderr, + flush=True, + ) + return None try: data = json.loads(result.stdout) refs = data.get("referrers", data.get("manifests", [])) return refs if refs is not None else [] except json.JSONDecodeError: - return [] + logger.debug( + f"oras discover for {image} returned non-JSON output" + ) + return None + + # Backwards-compatible alias kept as the previously private API. Existing + # call-sites that treat ``[]`` and ``None`` interchangeably can keep + # using this; new call-sites that need to distinguish between + # "discovery failed" and "no referrers" should call + # :meth:`discover_referrers` directly. + def _discover_referrers( + self, image: str, image_digest: Optional[str] = None + ) -> List[dict]: + refs = self.discover_referrers(image, image_digest) + return refs if refs is not None else [] @staticmethod def _strip_tag(image: str) -> str: @@ -407,26 +650,50 @@ def _strip_tag(image: str) -> str: def _find_attestation_digests( self, image: str, pred_type: str, image_digest: str ) -> List[str]: - """Find all attestation digests matching a predicate type.""" - referrers = self._discover_referrers(image, image_digest) - - bundle_type = "application/vnd.dev.sigstore.bundle.v0.3+json" - matching_digests = [] + """Find all attestation digests matching a predicate type. + + Reads from the per-instance predicate-digest cache populated by + ``list_attestations`` whenever possible -- the cache hit path + avoids re-enumerating + re-decoding every sigstore bundle on the + image, which on heavily-attested images (pharia-os-app: 400+ + bundles) was the dominant cosign-scan latency. Falls back to a + 20-way parallel decode if the cache has no entry for this image + OR if the requested predicate type is missing from the cached + map (e.g. ``list_attestations`` early-terminated before reaching + a SLSA / vuln bundle that an unusual caller might want). + """ image_name = self._strip_tag(image) + cached_map = self._predicate_digests_cache.get((image_name, image_digest)) + if cached_map is not None and pred_type in cached_map: + return list(cached_map[pred_type]) - for ref in referrers: - if ref.get("artifactType") != bundle_type: - continue + referrers = self._discover_referrers(image, image_digest) + bundle_type = "application/vnd.dev.sigstore.bundle.v0.3+json" + bundle_refs = [ + r for r in referrers + if r.get("artifactType") == bundle_type and r.get("digest") + ] - ref_digest = ref.get("digest") - if not ref_digest: - continue + if not bundle_refs: + return [] - actual_pred_type = self._get_predicate_type_from_bundle( + def fetch(ref: dict) -> Tuple[str, Optional[str]]: + ref_digest = ref["digest"] + return ref_digest, self._get_predicate_type_from_bundle( image_name, ref_digest ) - if actual_pred_type == pred_type: - matching_digests.append(ref_digest) + + matching_digests: List[str] = [] + max_workers = min(len(bundle_refs), 20) + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [executor.submit(fetch, ref) for ref in bundle_refs] + for future in as_completed(futures): + try: + ref_digest, actual_pred_type = future.result() + except Exception: + continue + if actual_pred_type == pred_type: + matching_digests.append(ref_digest) return matching_digests diff --git a/scanner_py/core/scanner.py b/scanner_py/core/scanner.py index 14c4501..41173ba 100644 --- a/scanner_py/core/scanner.py +++ b/scanner_py/core/scanner.py @@ -5,6 +5,7 @@ import json import re +import sys from dataclasses import dataclass from pathlib import Path from typing import Optional, List, Dict, Any @@ -265,7 +266,20 @@ def __init__( self.timeout = timeout self.verbose = verbose - # Initialize components + # Initialize components. + # + # ``timeout`` is the per-image budget for downstream trivy scans + # (default 10 min). Cosign verify itself only needs a couple of HTTP + # round-trips for lightly-attested images, but with ``--new-bundle-format`` + # cosign enumerates every OCI 1.1 referrer via the registry's Referrers + # API -- which in Harbor scales linearly with referrer count (observed + # ~8 s for 36 refs, ~95 s for 403 refs). Cap at 180 s so heavily- + # attested signed images (e.g. pharia-os-app with 400+ SBOM / triage / + # vuln / SLSA referrers) still verify successfully, while a truly + # unsigned / broken image still cannot block a worker for the full + # 10-minute trivy budget. Combined with the cheap ``oras discover`` + # pre-check in ``detect_attestation_type`` this keeps the long-tail of + # unsigned images fast. self.verifier = CosignVerifier( certificate_oidc_issuer=( certificate_oidc_issuer or CosignVerifier.DEFAULT_OIDC_ISSUER @@ -274,6 +288,7 @@ def __init__( certificate_identity_regexp or CosignVerifier.DEFAULT_IDENTITY_REGEXP ), timeout=timeout, + verify_timeout=180, ) self.extractor = AttestationExtractor( certificate_oidc_issuer=( @@ -301,7 +316,19 @@ def detect_attestation_type(self, image: str) -> AttestationType: """ Detect what attestations are available for an image. - Optimized to cache attestation info and avoid redundant network calls. + Order matters here: ``cosign verify`` is the most expensive call in + the discovery path -- on an unsigned image against a slow registry + it can sit for minutes before returning "no matching signatures". + ``oras discover`` is a single Referrers API GET capped at 15 s. + + Aleph-Alpha images attach all signature, SBOM and triage bundles via + OCI 1.1 referrers. We use the Referrers API as a fast pre-check: + when it definitively returns an empty referrers list we can mark + the image UNSIGNED without invoking the slow cosign round-trip. + On any discovery failure (timeout, network glitch, registry that + doesn't expose the Referrers API) we fall back to ``cosign verify`` + so we never misclassify a signed image as unsigned because of a + transient registry hiccup. Args: image: Image reference @@ -315,15 +342,53 @@ def detect_attestation_type(self, image: str) -> AttestationType: logger.debug(f"Failed to resolve digest for {image}") return AttestationType.UNSIGNED - # First verify the image is signed - verification = self.verifier.verify(image) - if not verification.success: - logger.debug(f"Image is not signed: {image}") - return AttestationType.UNSIGNED - - # List available attestations (already uses cached digest) + # Single source of truth: enumerate the OCI 1.1 referrers and + # decode each sigstore bundle's DSSE payload to learn its + # ``predicateType``. ``list_attestations`` already does both, in + # parallel, with cache-friendly oras calls. From its result we can + # derive every classification we need: + # + # * ``has_signature()`` -> a ``cosign/sign/v1`` bundle is present + # (proof the image was signed by *something*; downstream code + # verifies the cert chain when it actually pulls the SBOM / + # triage attestation). + # * ``has_sbom()`` -> a CycloneDX / SPDX bundle is attached. + # * ``has_triage()`` -> an Aleph-Alpha triage bundle is attached. + # + # Previous versions of this method called ``cosign verify`` here as + # a separate gating step, which on Harbor enumerated the Referrers + # API a *second* time -- adding 30-90 s per heavily-attested image + # and frequently timing out (Harbor's referrers latency is O(N) in + # referrer count, ~95 s for 400+ refs). That false-negatived every + # properly signed pharia-os image with hundreds of attestations. + # Relying on the predicate map keeps this path O(1) cosign calls + # and immune to Referrers API tail latency. attestations = self.extractor.list_attestations(image) + if not attestations.has_signature(): + # Diagnostic print to stderr (bypasses ``suppress_logging``, + # which is applied around parallel-scan workers and would + # otherwise hide this with --verbose off). Seeing the + # predicate map here distinguishes the main failure modes: + # * empty dict, discovery_failed=True -> Referrers API + # timed out (rerun + # may succeed). + # * empty dict, discovery_failed=False -> image legitimately + # has no referrers + # (truly unsigned). + # * non-empty but no sig predicate -> decode failures + # suppressed the + # signature bundle + # under high load. + print( + f"[cosign-scan] Classifying {image} as UNSIGNED;" + f" predicates={dict(attestations.attestations)}" + f" discovery_failed={attestations.discovery_failed}", + file=sys.stderr, + flush=True, + ) + return AttestationType.UNSIGNED + has_sbom = attestations.has_sbom() has_triage = attestations.has_triage() diff --git a/scanner_py/core/verification.py b/scanner_py/core/verification.py index de19810..d2795b2 100644 --- a/scanner_py/core/verification.py +++ b/scanner_py/core/verification.py @@ -45,6 +45,7 @@ def __init__( key_file: Optional[str] = None, rekor_url: str = DEFAULT_REKOR_URL, timeout: int = 600, + verify_timeout: Optional[int] = None, ): """ Initialize the verifier. @@ -55,7 +56,15 @@ def __init__( certificate_identity: Exact identity for keyless verification key_file: Path to public key file for key-based verification rekor_url: Rekor transparency log URL - timeout: Timeout for verification operations + timeout: Default timeout for verification operations. + verify_timeout: Optional hard ceiling for a single cosign verify + call. Useful when a verifier is shared with a long ``timeout`` + budgeted for downstream scans (e.g. ``ImageScanner`` runs + trivy with a 10-minute timeout, but cosign verify itself + should never need more than ~60 s and otherwise risks + blocking the worker on slow / unsigned images). Defaults to + ``timeout`` when omitted to preserve existing behavior for + standalone callers. """ self.certificate_oidc_issuer = certificate_oidc_issuer self.certificate_identity_regexp = certificate_identity_regexp @@ -63,6 +72,11 @@ def __init__( self.key_file = key_file self.rekor_url = rekor_url self.timeout = timeout + # Cap to whichever is smaller so callers can shrink (but not grow) the + # ceiling by passing a small ``timeout``. + self.verify_timeout = ( + min(timeout, verify_timeout) if verify_timeout is not None else timeout + ) self.keyless = key_file is None def resolve_image_digest(self, image: str) -> Optional[str]: @@ -109,8 +123,24 @@ def verify( logger.debug(f"Verifying image signature for: {image_ref}") - # Build cosign arguments - args = ["cosign", "verify", f"--timeout={self.timeout // 60}m"] + # Build cosign arguments. Use the (capped) verify_timeout so a hung + # call against an unsigned / slow registry does not eat minutes per + # image. We still pass it as seconds (cosign accepts ``Ns``). + # + # ``--new-bundle-format`` (cosign >= 2.5) is required so cosign + # discovers signatures stored as ``application/vnd.dev.sigstore.bundle.v0.3+json`` + # OCI 1.1 referrers. Without it cosign only probes the legacy + # ``:sha256-.sig`` tag, 404s, and returns "no + # signatures found" even when a perfectly valid sigstore bundle + # signature is attached to the image via the Referrers API. + # The sister ``_verify_attestation`` helper already uses this flag, + # so this aligns signature verification with attestation verification. + args = [ + "cosign", + "verify", + f"--timeout={self.verify_timeout}s", + "--new-bundle-format", + ] if self.keyless: logger.debug(f"Mode: Keyless verification") @@ -140,8 +170,10 @@ def verify( args.append(image_ref) - # Execute verification - result = run_with_timeout(args, self.timeout) + # Execute verification. The wall-clock ceiling matches the cosign + # ``--timeout`` so we don't keep waiting on a subprocess that has + # already given up. + result = run_with_timeout(args, self.verify_timeout) if result.success: logger.debug("Image signature verification successful!")