diff --git a/README.md b/README.md index 15f305bd..e090a42f 100644 --- a/README.md +++ b/README.md @@ -326,6 +326,7 @@ Setting `LOCK_FILE` (or `SBOM_FILE`) to `none` creates an empty SBOM and injects | `COMPONENT_VERSION` | No | Override component version in SBOM | | `COMPONENT_PURL` | No | Add or override component PURL in SBOM | | `PRODUCT_RELEASE` | No | Tag SBOM with product releases (see [Product Releases](#product-releases)) | +| `SUBMODULE_PATH` | No | Treat the component as a git submodule pinned at this path: resolve the pin to a version (exact version tag, else short commit SHA), attach the component's existing SBOM at that version if one exists, otherwise generate and upload it. Requires `LOCK_FILE` and the `sbomify` upload destination | | `UPLOAD` | No | Upload SBOM (default: true) | | `UPLOAD_DESTINATIONS` | No | Comma-separated destinations: `sbomify`, `dependency-track` (default: `sbomify`) | | `API_BASE_URL` | No | Override sbomify API URL for self-hosted instances | diff --git a/sbomify_action/cli/main.py b/sbomify_action/cli/main.py index 42e468c0..495262cf 100644 --- a/sbomify_action/cli/main.py +++ b/sbomify_action/cli/main.py @@ -226,6 +226,7 @@ class Config: component_name: Optional[str] = None component_purl: Optional[str] = None product_releases: Optional[str | list[str]] = None + submodule_path: Optional[str] = None api_base_url: str = SBOMIFY_PRODUCTION_API sbom_format: SBOMFormat = "cyclonedx" bom_type: Optional[str] = None @@ -321,6 +322,21 @@ def validate(self) -> None: if not any(inputs): raise ConfigurationError("Please provide one of: SBOM_FILE, LOCK_FILE, or DOCKER_IMAGE") + # Submodule mode: attach-or-backfill against the submodule's + # component. Needs a lockfile to backfill from, and only makes + # sense when talking to sbomify. + if self.submodule_path: + if not self.lock_file or self.is_additional_packages_only: + raise ConfigurationError( + "SUBMODULE_PATH requires LOCK_FILE (the submodule's lockfile) so the SBOM " + "can be generated when no existing one is found for the pinned version." + ) + if not self.uploads_to_sbomify: + raise ConfigurationError( + "SUBMODULE_PATH requires uploading to sbomify (UPLOAD=true with the 'sbomify' " + "destination) — submodule mode looks up and attaches SBOMs via the sbomify API." + ) + # Validate additional-packages-only mode if self.is_additional_packages_only: from ..additional_packages import has_additional_packages_configured @@ -575,6 +591,7 @@ def build_config( component_name: Optional[str] = None, component_purl: Optional[str] = None, product_releases: Optional[str] = None, + submodule_path: Optional[str] = None, api_base_url: str = SBOMIFY_PRODUCTION_API, sbom_format: str = "cyclonedx", bom_type: Optional[str] = None, @@ -622,6 +639,12 @@ def build_config( if product_releases: logger.info(f"Raw product release input: {product_releases}") + # Submodule mode: empty string (unset matrix field in the emitted + # workflow) means disabled. + normalized_submodule_path = submodule_path.strip() if submodule_path and submodule_path.strip() else None + if normalized_submodule_path: + logger.info(f"Submodule mode: {normalized_submodule_path} (attach-or-backfill)") + # Log SBOM format sbom_format_lower: SBOMFormat = cast(SBOMFormat, sbom_format.lower()) logger.info(f"SBOM format: {format_display_name(sbom_format_lower)}") @@ -660,6 +683,7 @@ def build_config( component_name=final_component_name, component_purl=component_purl, product_releases=product_releases, + submodule_path=normalized_submodule_path, api_base_url=api_base_url, sbom_format=sbom_format_lower, bom_type=normalized_bom_type, @@ -708,6 +732,7 @@ def load_config() -> Config: component_name=os.getenv("COMPONENT_NAME"), component_purl=os.getenv("COMPONENT_PURL"), product_releases=os.getenv("PRODUCT_RELEASE"), + submodule_path=os.getenv("SUBMODULE_PATH"), api_base_url=os.getenv("API_BASE_URL", SBOMIFY_PRODUCTION_API), sbom_format=os.getenv("SBOM_FORMAT", "cyclonedx"), bom_type=os.getenv("BOM_TYPE"), @@ -1296,6 +1321,154 @@ def _finalize_post_upload(results: "AggregateResult") -> None: _log_step_end(6) +def _find_existing_submodule_sbom(config: "Config", sbom_format: str) -> Optional[str]: + """ID of the submodule component's SBOM at the pin-derived version, or None. + + ``config.component_version`` must already hold the pin-derived version + (set by :func:`_prepare_submodule_mode`). Soft-fails to None on API + errors so callers fall back to generation rather than aborting. + """ + from ..sbomify_api import SbomifyApiClient + + if not (config.token and config.component_id and config.component_version): + return None + client = SbomifyApiClient(config.api_base_url, config.token) + try: + return client.find_component_sbom(config.component_id, config.component_version, sbom_format) + except APIError as e: + logger.warning(f"Could not look up an existing submodule SBOM: {e}") + return None + + +def _prepare_submodule_mode(config: "Config") -> Optional[str]: + """Resolve the submodule pin; return an existing SBOM's ID if one matches. + + Side effect: overrides ``config.component_version`` with the + pin-derived version (exact version tag at the pinned commit, else the + short SHA) so that a backfill uploads under the version the + submodule's own CI would have published. + + Returns the sbom_id to attach (skip generation/upload entirely), or + None to proceed with the normal pipeline as a backfill. + """ + from ..submodule import resolve_submodule_pin + + assert config.submodule_path is not None # guarded by the caller + pin = resolve_submodule_pin(Path.cwd(), config.submodule_path) + if pin is None: + logger.error( + f"SUBMODULE_PATH '{config.submodule_path}' is not a git submodule or vendored repo " + "in this checkout: the parent tree has no gitlink at that path and the directory " + "has no embedded .git. Check the path, and ensure the workflow checks out the " + "parent repository (the pin is read from the parent tree, not the submodule)." + ) + sys.exit(1) + + source_desc = "version tag" if pin.version_source == "tag" else "short commit SHA" + logger.info(f"Submodule '{pin.path}' pinned at {pin.sha} → version '{pin.version}' ({source_desc})") + if config.component_version and config.component_version != pin.version: + logger.info(f"Overriding COMPONENT_VERSION '{config.component_version}' with the pin-derived '{pin.version}'") + config.component_version = pin.version + + existing = _find_existing_submodule_sbom(config, config.sbom_format) + if existing: + logger.info( + f"Component {config.component_id} already has a {format_display_name(config.sbom_format)} " + f"SBOM at version '{pin.version}' (id: {existing})" + ) + return existing + logger.info( + f"No existing {format_display_name(config.sbom_format)} SBOM for component " + f"{config.component_id} at version '{pin.version}' — generating one (backfill)." + ) + return None + + +def _run_post_upload_processing(config: "Config", sbom_id: str) -> None: + """Step 6: post-upload processors (release tagging etc.) for ``sbom_id``.""" + _log_step_header(6, "Post-upload Processing") + try: + from sbomify_action._processors import ProcessorInput, ProcessorOrchestrator + + # Re-mint the OIDC token before kicking off processors: long pipelines + # (large Docker images, Yocto builds, slow enrichment) can exceed the + # 15-minute default TTL on the originally minted token. + if config.token_is_oidc_minted: + from ..exceptions import OIDCBindingMissingError, OIDCExchangeError + from ..oidc import is_github_oidc_available, obtain_sbomify_token_via_oidc + + if is_github_oidc_available(): + try: + config.token = obtain_sbomify_token_via_oidc( + component_id=config.component_id, + api_base_url=config.api_base_url, + audience=config.oidc_audience, + ) + except (OIDCBindingMissingError, OIDCExchangeError) as exc: + logger.warning( + f"Could not refresh OIDC-minted token before processors: {exc}. " + "Continuing with the original token — long-running pipelines may see 401s." + ) + + orchestrator = ProcessorOrchestrator( + api_base_url=config.api_base_url, + token=config.token, + ) + # Normalize product_releases to list[str] | None for ProcessorInput + pr_list: list[str] | None = None + if isinstance(config.product_releases, list): + pr_list = config.product_releases + elif isinstance(config.product_releases, str): + pr_list = json.loads(config.product_releases) + + processor_input = ProcessorInput( + sbom_id=sbom_id, + sbom_file=config.output_file, + product_releases=pr_list, + api_base_url=config.api_base_url, + token=config.token, + ) + + # Check if any processors are enabled + enabled_processors = orchestrator.get_enabled_processors(processor_input) + if enabled_processors: + logger.info(f"Running {len(enabled_processors)} processor(s): {enabled_processors}") + results = orchestrator.process_all(processor_input) + # Raises SystemExit(1) if any processor failed (e.g. a 403 cutting + # a release) so the failure isn't swallowed as a green run. + _finalize_post_upload(results) + else: + logger.info("No processors enabled for this run") + _log_step_end(6) + except Exception as e: + # Crash in orchestrator setup. A processor's own failure already + # comes back as a failure_result (handled above), so this only + # catches setup/import errors; keep it non-fatal as before. + logger.error(f"Step 6 (post-upload processing) failed: {e}") + _log_step_end(6, success=False) + + +def _finalize_run(config: "Config") -> None: + """Finalize + persist the audit trail and print the success summary.""" + audit_trail = get_audit_trail() + audit_trail.output_file = config.output_file + + # Write audit trail file + audit_trail_path = Path(config.output_file).parent / "audit_trail.txt" + try: + audit_trail.write_audit_file(str(audit_trail_path)) + logger.info(f"Audit trail written to: {audit_trail_path}") + except Exception as e: + logger.warning(f"Failed to write audit trail file: {e}") + + # Print summary and full audit trail for attestation + audit_trail.print_summary() + audit_trail.print_to_stdout_for_attestation() + + # Final success message + print_final_success() + + def run_pipeline(config: Config) -> None: """ Run the SBOM pipeline with the given configuration. @@ -1360,6 +1533,24 @@ def run_pipeline(config: Config) -> None: ) sys.exit(1) + # Submodule mode: resolve the pin to a version and check whether the + # submodule's component already published an SBOM at exactly that + # version. Hit → attach it to the configured release(s), skipping + # generation and upload entirely (the submodule's own CI produced the + # authoritative artifact). Miss → fall through to the normal pipeline + # as a backfill, with COMPONENT_VERSION overridden to the pin-derived + # version so the upload lands where the next run's lookup will find it. + if config.submodule_path: + existing_sbom_id = _prepare_submodule_mode(config) + if existing_sbom_id: + logger.info("Skipping SBOM generation and upload — attaching the existing SBOM instead.") + if config.product_releases: + _run_post_upload_processing(config, existing_sbom_id) + else: + logger.info("No PRODUCT_RELEASE configured; the existing SBOM already covers this pin.") + _finalize_run(config) + return + # Step 1: SBOM Generation/Validation _log_step_header(1, "SBOM Generation/Input Processing") @@ -1805,6 +1996,27 @@ def run_pipeline(config: Config) -> None: bom_type=config.bom_type, ) + if ( + not upload_result.success + and upload_result.error_code == "DUPLICATE_ARTIFACT" + and config.submodule_path + and destination == "sbomify" + ): + # Backfill race: another workflow published this + # (component, version, format) between our preflight + # lookup and this upload. The backend's uniqueness + # constraint guarantees a single winner — recover by + # re-looking it up and attaching that SBOM instead of + # failing the run. + recovered = _find_existing_submodule_sbom(config, FORMAT) + if recovered: + logger.info( + f"Duplicate upload for submodule component — another workflow published " + f"this SBOM first; reusing the existing one (id: {recovered})." + ) + sbom_id = recovered + continue + if not upload_result.success: if upload_result.error_code == "DUPLICATE_ARTIFACT": logger.error( @@ -1846,89 +2058,13 @@ def run_pipeline(config: Config) -> None: # Step 6: Post-upload Processing (releases, signing, etc.) if sbom_id: - _log_step_header(6, "Post-upload Processing") - try: - from sbomify_action._processors import ProcessorInput, ProcessorOrchestrator - - # Re-mint the OIDC token before kicking off processors: long pipelines - # (large Docker images, Yocto builds, slow enrichment) can exceed the - # 15-minute default TTL on the originally minted token. - if config.token_is_oidc_minted: - from ..exceptions import OIDCBindingMissingError, OIDCExchangeError - from ..oidc import is_github_oidc_available, obtain_sbomify_token_via_oidc - - if is_github_oidc_available(): - try: - config.token = obtain_sbomify_token_via_oidc( - component_id=config.component_id, - api_base_url=config.api_base_url, - audience=config.oidc_audience, - ) - except (OIDCBindingMissingError, OIDCExchangeError) as exc: - logger.warning( - f"Could not refresh OIDC-minted token before processors: {exc}. " - "Continuing with the original token — long-running pipelines may see 401s." - ) - - orchestrator = ProcessorOrchestrator( - api_base_url=config.api_base_url, - token=config.token, - ) - # Normalize product_releases to list[str] | None for ProcessorInput - pr_list: list[str] | None = None - if isinstance(config.product_releases, list): - pr_list = config.product_releases - elif isinstance(config.product_releases, str): - pr_list = json.loads(config.product_releases) - - processor_input = ProcessorInput( - sbom_id=sbom_id, - sbom_file=config.output_file, - product_releases=pr_list, - api_base_url=config.api_base_url, - token=config.token, - ) - - # Check if any processors are enabled - enabled_processors = orchestrator.get_enabled_processors(processor_input) - if enabled_processors: - logger.info(f"Running {len(enabled_processors)} processor(s): {enabled_processors}") - results = orchestrator.process_all(processor_input) - # Raises SystemExit(1) if any processor failed (e.g. a 403 cutting - # a release) so the failure isn't swallowed as a green run. - _finalize_post_upload(results) - else: - logger.info("No processors enabled for this run") - _log_step_end(6) - except Exception as e: - # Crash in orchestrator setup. A processor's own failure already - # comes back as a failure_result (handled above), so this only - # catches setup/import errors; keep it non-fatal as before. - logger.error(f"Step 6 (post-upload processing) failed: {e}") - _log_step_end(6, success=False) + _run_post_upload_processing(config, sbom_id) elif config.product_releases and not sbom_id: _log_step_header(6, "Post-upload Processing - SKIPPED") logger.warning("Product releases specified but no SBOM ID available (upload may have been disabled or failed)") _log_step_end(6, success=False) - # Finalize audit trail - audit_trail = get_audit_trail() - audit_trail.output_file = config.output_file - - # Write audit trail file - audit_trail_path = Path(config.output_file).parent / "audit_trail.txt" - try: - audit_trail.write_audit_file(str(audit_trail_path)) - logger.info(f"Audit trail written to: {audit_trail_path}") - except Exception as e: - logger.warning(f"Failed to write audit trail file: {e}") - - # Print summary and full audit trail for attestation - audit_trail.print_summary() - audit_trail.print_to_stdout_for_attestation() - - # Final success message - print_final_success() + _finalize_run(config) def _validate_cyclonedx_sbom(sbom_file_path: str) -> bool | None: @@ -2517,6 +2653,16 @@ def _parse_upload_destinations_callback( envvar="PRODUCT_RELEASE", help="Tag SBOM with product releases (JSON array: '[\"product_id:v1.0.0\"]').", ) +@click.option( + "--submodule-path", + envvar="SUBMODULE_PATH", + default=None, + help=( + "Treat the component as a git submodule pinned at this path: resolve the pin to a " + "version (tag or short SHA), attach the component's existing SBOM at that version if " + "one exists, otherwise generate and upload it (backfill)." + ), +) @click.option( "--api-base-url", envvar="API_BASE_URL", @@ -2600,6 +2746,7 @@ def cli( component_name: Optional[str], component_purl: Optional[str], product_releases: Optional[str], + submodule_path: Optional[str], api_base_url: str, sbom_format: str, bom_type: Optional[str], @@ -2724,6 +2871,7 @@ def cli( component_name=component_name, component_purl=component_purl, product_releases=product_releases, + submodule_path=submodule_path, api_base_url=api_base_url, sbom_format=sbom_format, bom_type=bom_type, diff --git a/sbomify_action/cli/wizard/ci_emitter.py b/sbomify_action/cli/wizard/ci_emitter.py index 07b1cda9..5ee004f4 100644 --- a/sbomify_action/cli/wizard/ci_emitter.py +++ b/sbomify_action/cli/wizard/ci_emitter.py @@ -255,11 +255,21 @@ def _matrix_block( components: list[PlannedComponent], formats: list[SbomFormat], component_ids: dict[str, str], + *, + attestation: bool = False, ) -> str: """Render the ``matrix.include:`` rows — one per (component, format). Row ``name`` is suffixed with the format only when more than one format is being emitted, so single-format workflows stay readable. + + When ``attestation`` is enabled every row carries an ``attest`` + boolean gating the attest-build-provenance step. Lockfiles inside a + submodule / vendored repo get ``attest: false``: the attestation's + Sigstore identity would be *this* repo's workflow, but the SBOM + describes code owned by another repository — signing it here would + produce provenance that fails verification against the repo the + code actually comes from. """ rows: list[str] = [] multi_format = len(formats) > 1 @@ -288,7 +298,7 @@ def _matrix_block( ext = _format_extension(fmt) row_name = f"{row_slug}-{fmt}" if multi_format else row_slug output_file = f"{row_slug}.{ext}" - rows.append( + row = ( " - name: " + row_name + "\n" " component_name: " + c.name + "\n" " component_id: " + cid + "\n" @@ -296,6 +306,25 @@ def _matrix_block( " sbom_format: " + fmt + "\n" " output_file: " + output_file + "\n" ) + if c.lockfile.nested_repo: + # Drives the action's attach-or-backfill submodule mode: + # it resolves the pinned commit to a version, attaches the + # component's existing SBOM at that version when one + # exists, and only generates + uploads otherwise. + row += " submodule_path: " + c.lockfile.nested_repo + "\n" + if attestation: + if c.lockfile.nested_repo: + kind = "submodule" if c.lockfile.nested_repo_kind == "submodule" else "vendored repo" + row += ( + f" # Deliberately NOT signed: {c.lockfile.nested_repo} is a {kind} — " + "another repository's code. An attestation from this repo's workflow\n" + " # would carry the wrong Sigstore identity and fail verification " + "against the repository the code actually comes from.\n" + " attest: false\n" + ) + else: + row += " attest: true\n" + rows.append(row) return "".join(rows) @@ -372,6 +401,7 @@ def _env_block( enrich: bool, release_strategy: ReleaseStrategy, product_id: str | None, + has_submodules: bool = False, ) -> str: """The ``env:`` block under the action step. @@ -402,6 +432,10 @@ def _env_block( " SYFT_CACHE_DIR: ${{ github.workspace }}/.sbomify-cache/syft", ] ) + if has_submodules: + # Empty for non-submodule rows (matrix field unset) — the action + # treats an empty SUBMODULE_PATH as disabled. + lines.append(" SUBMODULE_PATH: ${{ matrix.submodule_path }}") if release_strategy == "tag" and product_id: # PRODUCT_RELEASE is parsed by cli/main.py as a JSON list — see # cli/main.py's "PRODUCT_RELEASE must be a JSON list like @@ -469,7 +503,13 @@ def _attest_step() -> str: " # If you hit one of the unsupported configurations, either upgrade to\n" " # GitHub Enterprise Cloud, make the repo public, or remove this step.\n" " # Reference: https://github.com/actions/attest-build-provenance\n" + " #\n" + " # Matrix entries with attest: false are skipped: those SBOMs describe\n" + " # code owned by another repository (a git submodule or vendored\n" + " # checkout), and an attestation signed by this repo's workflow would\n" + " # fail verification against the repository the code comes from.\n" f" - uses: actions/attest-build-provenance@{PINNED_ATTEST_SHA} # {PINNED_ATTEST_VERSION}\n" + " if: ${{ matrix.attest }}\n" " with:\n" " subject-path: '${{ github.workspace }}/${{ matrix.output_file }}'\n" ) @@ -509,6 +549,7 @@ def emit_workflow( component_ids = component_ids or {} lockfile_paths = [str(c.lockfile.rel_path) for c in plan.create_components] formats = plan.sbom_formats or ["cyclonedx"] + has_submodules = any(c.lockfile.nested_repo for c in plan.create_components) permissions = _permissions_block(plan.credential_mode, plan.attestation) trigger = _trigger_block(plan.release_strategy, facts.default_branch, lockfile_paths) @@ -520,9 +561,21 @@ def emit_workflow( enrich=plan.enrich, release_strategy=plan.release_strategy, product_id=product_id or plan.use_product_id, + has_submodules=has_submodules, ) - matrix = _matrix_block(plan.create_components, formats, component_ids) + matrix = _matrix_block(plan.create_components, formats, component_ids, attestation=plan.attestation) attest_step = _attest_step() if plan.attestation else "" + if has_submodules: + # Submodule contents are only needed on the backfill path (no + # published SBOM at the pinned version yet), but the checkout has + # to cover it unconditionally. + checkout_step = ( + f" - uses: actions/checkout@{PINNED_CHECKOUT_SHA} # {PINNED_CHECKOUT_VERSION}\n" + " with:\n" + " submodules: recursive\n" + ) + else: + checkout_step = f" - uses: actions/checkout@{PINNED_CHECKOUT_SHA} # {PINNED_CHECKOUT_VERSION}\n" # Resolved by the caller (apply / review) so the GitHub lookup happens # once per run. Omitted by snapshot tests and ad-hoc callers — fall back @@ -544,7 +597,7 @@ def emit_workflow( " include:\n" f"{matrix}" " steps:\n" - f" - uses: actions/checkout@{PINNED_CHECKOUT_SHA} # {PINNED_CHECKOUT_VERSION}\n" + f"{checkout_step}" f"{_cache_step()}" f"{version_step}" f" - uses: {action_ref}\n" diff --git a/sbomify_action/cli/wizard/discovery.py b/sbomify_action/cli/wizard/discovery.py index 8195b722..56f2af7d 100644 --- a/sbomify_action/cli/wizard/discovery.py +++ b/sbomify_action/cli/wizard/discovery.py @@ -19,7 +19,7 @@ from pathlib import Path from sbomify_action._generation.utils import ALL_LOCK_FILES, get_lock_file_ecosystem -from sbomify_action.cli.wizard.state import DiscoveredLockfile +from sbomify_action.cli.wizard.state import DiscoveredLockfile, NestedRepoKind # Cap the walk so we never scan a degenerate monorepo into memory. 200 # is enormous in practice; if a repo legitimately has more lockfiles @@ -165,15 +165,21 @@ def discover(repo_root: Path, *, repo_name: str | None = None) -> list[Discovere if len(best_per_dir_eco) >= DISCOVERY_CAP: break + submodule_paths = _submodule_paths(repo_root) + nested_cache: dict[Path, tuple[str, NestedRepoKind] | None] = {} + found: list[DiscoveredLockfile] = [] for (_dir, ecosystem), (_priority, path) in best_per_dir_eco.items(): rel = path.relative_to(repo_root) + nested = _enclosing_nested_repo(path.parent, repo_root, submodule_paths, nested_cache) found.append( DiscoveredLockfile( path=path, rel_path=rel, ecosystem=ecosystem, suggested_name=_suggested_name(repo_name, path.name), + nested_repo=nested[0] if nested else None, + nested_repo_kind=nested[1] if nested else None, ) ) @@ -197,6 +203,54 @@ def discover(repo_root: Path, *, repo_name: str | None = None) -> list[Discovere return found +# ``path = `` lines inside .gitmodules. Values may be quoted when +# they contain spaces; git writes forward slashes on every platform. +_GITMODULES_PATH_RE = re.compile(r"^\s*path\s*=\s*(?P.+?)\s*$", re.MULTILINE) + + +def _submodule_paths(repo_root: Path) -> frozenset[str]: + """Repo-relative POSIX paths declared as submodules in ``.gitmodules``. + + Parsed directly from the file rather than via ``git submodule`` so it + works without a git binary and for submodules that were declared but + never initialised. + """ + try: + text = (repo_root / ".gitmodules").read_text(encoding="utf-8", errors="replace") + except OSError: + return frozenset() + return frozenset(m.group("path").strip('"').rstrip("/") for m in _GITMODULES_PATH_RE.finditer(text)) + + +def _enclosing_nested_repo( + directory: Path, + repo_root: Path, + submodule_paths: frozenset[str], + cache: dict[Path, tuple[str, NestedRepoKind] | None], +) -> tuple[str, NestedRepoKind] | None: + """Innermost ancestor of ``directory`` that is its own repository. + + A directory counts if it is declared in ``.gitmodules`` (submodule) + or has its own ``.git`` entry — a file for submodule checkouts, a + directory for vendored clones. Results are memoised per directory + since sibling lockfiles share ancestors. + """ + if directory == repo_root: + return None + if directory in cache: + return cache[directory] + rel = directory.relative_to(repo_root).as_posix() + result: tuple[str, NestedRepoKind] | None + if rel in submodule_paths: + result = (rel, "submodule") + elif (directory / ".git").exists(): + result = (rel, "vendored") + else: + result = _enclosing_nested_repo(directory.parent, repo_root, submodule_paths, cache) + cache[directory] = result + return result + + def _walk(root: Path) -> Iterator[Path]: """Yield every file under ``root``, skipping irrelevant directories.""" stack: list[Path] = [root] diff --git a/sbomify_action/cli/wizard/screens/discover.py b/sbomify_action/cli/wizard/screens/discover.py index b24723f5..cd45361c 100644 --- a/sbomify_action/cli/wizard/screens/discover.py +++ b/sbomify_action/cli/wizard/screens/discover.py @@ -38,6 +38,14 @@ def compose_body(self) -> ComposeResult: "[b]n[/] to select none, [b]Enter[/] when you're done.", classes="wizard-help", ) + if any(lf.nested_repo for lf in self.wizard.state.discovered): + yield Static( + "[#F4B57F]Lockfiles inside submodules or vendored repos are deselected " + "by default — they belong to another repository, so set up SBOMs there " + "instead.[/]", + id="nested-repo-note", + classes="wizard-help", + ) yield SelectionList[int](id="lockfile-list") yield Static("", id="discover-status", markup=True) with Horizontal(classes="button-row"): @@ -48,7 +56,12 @@ def on_mount(self) -> None: sel = self.query_one("#lockfile-list", SelectionList) for idx, lf in enumerate(self.wizard.state.discovered): label = f"{lf.rel_path} [#5E5E5E]({lf.ecosystem})[/]" - sel.add_option((label, idx, True)) # default-selected + if lf.nested_repo: + kind = "submodule" if lf.nested_repo_kind == "submodule" else "vendored repo" + label += f" [#F4B57F]({kind}: {lf.nested_repo})[/]" + # Nested-repo lockfiles default to deselected — they belong to + # another repository and are better tracked from there. + sel.add_option((label, idx, lf.nested_repo is None)) sel.focus() def action_toggle_selection(self) -> None: diff --git a/sbomify_action/cli/wizard/state.py b/sbomify_action/cli/wizard/state.py index 5665958d..5ac1fbfc 100644 --- a/sbomify_action/cli/wizard/state.py +++ b/sbomify_action/cli/wizard/state.py @@ -62,6 +62,15 @@ """ +NestedRepoKind = Literal["submodule", "vendored"] +"""Why a lockfile's directory belongs to a different repository. + +- ``submodule`` — under a path declared in ``.gitmodules``. +- ``vendored`` — under a directory with its own ``.git`` (a checked-in + clone that isn't a registered submodule). +""" + + @dataclass(frozen=True) class DiscoveredLockfile: """One lockfile the wizard found in the repo.""" @@ -70,6 +79,12 @@ class DiscoveredLockfile: rel_path: Path ecosystem: str suggested_name: str + nested_repo: str | None = None + """Repo-relative POSIX path of the enclosing submodule / vendored + repo root, when the lockfile belongs to one. Those lockfiles are + better tracked from their own repository, so the discover screen + annotates them and leaves them deselected by default.""" + nested_repo_kind: NestedRepoKind | None = None @dataclass(frozen=True) diff --git a/sbomify_action/sbomify_api.py b/sbomify_action/sbomify_api.py index 6df748de..db4551d3 100644 --- a/sbomify_action/sbomify_api.py +++ b/sbomify_action/sbomify_api.py @@ -408,6 +408,49 @@ def patch_component_visibility(self, component_id: str, visibility: str) -> None if not response.ok: logger.warning(f"Failed to set visibility for component {component_id}: [{response.status_code}]") + def list_component_sboms( + self, + component_id: str, + *, + version: str | None = None, + sbom_format: str | None = None, + ) -> list[dict[str, Any]]: + """List a component's SBOMs, newest first. + + ``version`` / ``sbom_format`` are passed to the backend's + exact-match filters (``GET /api/v1/components/{id}/sboms``). + Each item is ``{"sbom": {id, version, format, created_at, ...}, + "releases": [...], ...}``. + """ + params: dict[str, Any] = {} + if version is not None: + params["version"] = version + if sbom_format is not None: + params["format"] = sbom_format + return list( + self._paginate( + f"/api/v1/components/{component_id}/sboms", + params=params, + error_context="list component SBOMs", + ) + ) + + def find_component_sbom(self, component_id: str, version: str, sbom_format: str) -> str | None: + """ID of the newest SBOM at exactly ``(version, sbom_format)``, or None. + + The filter params are re-checked client-side: a backend that + predates the server-side filters ignores unknown query params and + would otherwise return the full unfiltered listing, silently + matching the wrong SBOM. + """ + for item in self.list_component_sboms(component_id, version=version, sbom_format=sbom_format): + sbom = item.get("sbom") + if not isinstance(sbom, dict): + continue + if sbom.get("version") == version and sbom.get("format") == sbom_format and sbom.get("id"): + return str(sbom["id"]) + return None + def get_augmentation_meta(self, component_id: str) -> dict[str, Any]: """Fetch augmentation metadata for a component. diff --git a/sbomify_action/submodule.py b/sbomify_action/submodule.py new file mode 100644 index 00000000..f3500866 --- /dev/null +++ b/sbomify_action/submodule.py @@ -0,0 +1,216 @@ +"""Resolve a git submodule's pinned commit to an SBOM version string. + +When the action runs in submodule mode (``SUBMODULE_PATH`` set), the +component's SBOM version must identify the *submodule's* pinned state, +not the parent repo's release version. The submodule's own CI publishes +SBOMs versioned either by a git tag (tag-strategy workflows) or by the +short commit SHA (trunk-strategy workflows use ``git rev-parse --short +HEAD``). To find — or backfill — the matching SBOM, we resolve the pin +the same way: + +1. a version-shaped tag (``v1.2.3``, ``1.2.3``, CalVer) pointing at the + pinned commit, resolved via ``git ls-remote --tags`` against the + submodule's remote (works without the submodule being initialised); +2. otherwise the 7-char short SHA. + +Matching is exact on purpose: a pin that is 14 commits past ``v1.2.3`` +must NOT resolve to ``v1.2.3`` — attaching that SBOM would misrepresent +what the release ships. No ``git describe``-style nearest-tag logic. +""" + +from __future__ import annotations + +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Literal + +from .logging_config import logger + +# Local git plumbing (ls-tree, rev-parse, config) is fast; ls-remote +# talks to the network and gets a generous budget. +_LOCAL_TIMEOUT = 10 +_REMOTE_TIMEOUT = 30 + +# Version-shaped tags, mirroring the tag patterns wizard-emitted +# workflows trigger on (``v*`` + ``[0-9]*``): v-prefixed or bare-numeric +# SemVer / CalVer. +_VERSION_TAG_RE = re.compile(r"^v?\d") + +_SHORT_SHA_LEN = 7 + + +@dataclass(frozen=True) +class SubmodulePin: + """The resolved state of one submodule pin.""" + + path: str + """Submodule path relative to the parent repo root.""" + + sha: str + """Full commit SHA the parent tree pins the submodule to.""" + + version: str + """SBOM version string the submodule's CI would have published + under: an exact version tag when one points at :attr:`sha`, + otherwise the short SHA.""" + + version_source: Literal["tag", "sha"] + + +def _run_git(args: list[str], *, cwd: Path, timeout: int = _LOCAL_TIMEOUT) -> str | None: + """Run ``git ``; return stripped stdout, or None on any failure. + + ``-c safe.directory=*`` keeps the commands working when the workspace + is bind-mounted into a container under a different UID (same + rationale as the wizard's repo_facts helper). All commands here are + read-only. + """ + try: + result = subprocess.run( + ["git", "-c", "safe.directory=*", *args], + cwd=cwd, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + except (FileNotFoundError, subprocess.TimeoutExpired, OSError): + return None + if result.returncode != 0: + return None + return result.stdout.strip() + + +def _gitlink_sha(repo_root: Path, submodule_path: str) -> str | None: + """The commit SHA the parent tree records for ``submodule_path``. + + Read from the parent's tree (``ls-tree`` mode 160000), so it works + even when the submodule was never initialised or checked out. + """ + out = _run_git(["ls-tree", "HEAD", "--", submodule_path], cwd=repo_root) + if not out: + return None + # " \t" — a gitlink is mode 160000 / type commit. + for line in out.splitlines(): + parts = line.split() + if len(parts) >= 3 and parts[0] == "160000" and parts[1] == "commit": + return parts[2] + return None + + +def _embedded_repo_sha(repo_root: Path, submodule_path: str) -> str | None: + """HEAD of a vendored clone (its own ``.git``, no gitlink in the tree).""" + candidate = repo_root / submodule_path + if not (candidate / ".git").exists(): + return None + return _run_git(["-C", str(candidate), "rev-parse", "HEAD"], cwd=repo_root) + + +def _submodule_url(repo_root: Path, submodule_path: str) -> str | None: + """The remote URL ``.gitmodules`` declares for ``submodule_path``. + + Uses ``git config -f .gitmodules`` so quoting/escaping is handled by + git itself rather than a hand-rolled parser. + """ + if not (repo_root / ".gitmodules").is_file(): + return None + mapping = _run_git(["config", "-f", ".gitmodules", "--get-regexp", r"^submodule\..*\.path$"], cwd=repo_root) + if not mapping: + return None + for line in mapping.splitlines(): + key, _, value = line.partition(" ") + if value.strip().rstrip("/") == submodule_path.rstrip("/"): + # key = submodule..path → submodule..url + url_key = key[: -len(".path")] + ".url" + return _run_git(["config", "-f", ".gitmodules", "--get", url_key], cwd=repo_root) + return None + + +def _tags_at_sha_remote(repo_root: Path, url: str, sha: str) -> list[str]: + """Tags on ``url`` whose target commit is ``sha``. + + Runs with the parent repo as cwd so repo-local auth config (e.g. the + ``http..extraheader`` credential actions/checkout installs) + applies to the remote call. Annotated tags are resolved through + their peeled (``^{}``) entries, which override the tag-object SHA. + """ + out = _run_git(["ls-remote", "--tags", url], cwd=repo_root, timeout=_REMOTE_TIMEOUT) + if not out: + return [] + targets: dict[str, str] = {} + for line in out.splitlines(): + line_sha, _, ref = line.partition("\t") + line_sha = line_sha.strip() + ref = ref.strip() + if not ref.startswith("refs/tags/"): + continue + tag = ref[len("refs/tags/") :] + if tag.endswith("^{}"): + # Peeled entry: the commit an annotated tag points at — this + # is the SHA that matches the gitlink, so it wins. + targets[tag[: -len("^{}")]] = line_sha + else: + targets.setdefault(tag, line_sha) + return [tag for tag, target in targets.items() if target == sha] + + +def _tags_at_sha_local(repo_root: Path, submodule_path: str, sha: str) -> list[str]: + """Tags in the checked-out submodule's own clone pointing at ``sha``. + + Fallback for when there is no usable remote URL (vendored clones) or + the remote call failed. Shallow submodule checkouts often carry no + tags at all, so an empty result here is normal. + """ + candidate = repo_root / submodule_path + if not (candidate / ".git").exists(): + return [] + out = _run_git(["-C", str(candidate), "tag", "--points-at", sha], cwd=repo_root) + if not out: + return [] + return out.splitlines() + + +def _pick_version_tag(tags: list[str]) -> str | None: + """Deterministically choose one version-shaped tag. + + v-prefixed tags win over bare-numeric ones (the dominant convention, + and the first pattern wizard-emitted workflows trigger on); + lexicographic order breaks remaining ties. + """ + candidates = sorted( + (tag.strip() for tag in tags if _VERSION_TAG_RE.match(tag.strip())), + key=lambda tag: (0 if tag.startswith("v") else 1, tag), + ) + return candidates[0] if candidates else None + + +def resolve_submodule_pin(repo_root: Path, submodule_path: str) -> SubmodulePin | None: + """Resolve the pinned commit + SBOM version for ``submodule_path``. + + Returns None when the path is neither a gitlink in the parent tree + nor a vendored clone with its own ``.git`` — the caller treats that + as a configuration error. + """ + repo_root = repo_root.resolve() + normalized = submodule_path.strip().strip("/") + + sha = _gitlink_sha(repo_root, normalized) + if not sha: + sha = _embedded_repo_sha(repo_root, normalized) + if not sha: + return None + + tags: list[str] = [] + url = _submodule_url(repo_root, normalized) + if url: + tags = _tags_at_sha_remote(repo_root, url, sha) + if not tags: + logger.debug(f"No remote tags at {sha} for submodule '{normalized}' (url: {url})") + if not tags: + tags = _tags_at_sha_local(repo_root, normalized, sha) + + if tag := _pick_version_tag(tags): + return SubmodulePin(path=normalized, sha=sha, version=tag, version_source="tag") + return SubmodulePin(path=normalized, sha=sha, version=sha[:_SHORT_SHA_LEN], version_source="sha") diff --git a/tests/test_config.py b/tests/test_config.py index 4ebc1b2d..c5835962 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1215,3 +1215,57 @@ def test_load_config_and_build_config_parity(self): if __name__ == "__main__": unittest.main() + + +class TestSubmoduleConfig(unittest.TestCase): + """Validation rules for submodule (attach-or-backfill) mode.""" + + def test_submodule_path_requires_lock_file(self): + config = Config( + token="t", + component_id="c1", + sbom_file="/path/to/sbom.json", + submodule_path="extern/lib", + ) + with self.assertRaises(ConfigurationError) as cm: + config.validate() + self.assertIn("SUBMODULE_PATH requires LOCK_FILE", str(cm.exception)) + + def test_submodule_path_requires_sbomify_upload(self): + config = Config( + token="t", + component_id="c1", + lock_file="extern/lib/Cargo.lock", + submodule_path="extern/lib", + upload=False, + ) + with self.assertRaises(ConfigurationError) as cm: + config.validate() + self.assertIn("SUBMODULE_PATH requires uploading to sbomify", str(cm.exception)) + + def test_submodule_path_valid_config(self): + config = Config( + token="t", + component_id="c1", + lock_file="extern/lib/Cargo.lock", + submodule_path="extern/lib", + ) + config.validate() # must not raise + + def test_empty_submodule_path_env_is_disabled(self): + """The emitted workflow sets SUBMODULE_PATH to an empty string on + non-submodule matrix rows — that must not enable submodule mode.""" + with tempfile.NamedTemporaryFile(suffix=".json") as f: + with patch.dict( + os.environ, + { + "SUBMODULE_PATH": "", + "TOKEN": "t", + "COMPONENT_ID": "c1", + "SBOM_FILE": f.name, + "UPLOAD": "false", + }, + clear=False, + ): + config = load_config() + self.assertIsNone(config.submodule_path) diff --git a/tests/test_sbomify_api.py b/tests/test_sbomify_api.py index c2ceb45c..e87dd3b0 100644 --- a/tests/test_sbomify_api.py +++ b/tests/test_sbomify_api.py @@ -617,3 +617,63 @@ def test_upload_sbom_explicit_sbom_sends_no_bom_type_param() -> None: with patch.object(client, "_request", return_value=MagicMock()) as req: client.upload_sbom(component_id="c1", sbom_payload=b"{}", sbom_format="cyclonedx", bom_type="sbom") assert req.call_args.kwargs["params"] is None + + +# ---------------------------------------------------------------------- +# component SBOM lookup (submodule attach-or-backfill) + + +def test_list_component_sboms_passes_exact_match_filters() -> None: + client, session = _client_with( + [_FakeResponse(200, {"items": [{"sbom": {"id": "s1", "version": "v1.2.3", "format": "cyclonedx"}}]})] + ) + items = client.list_component_sboms("c1", version="v1.2.3", sbom_format="cyclonedx") + assert [i["sbom"]["id"] for i in items] == ["s1"] + params = session.request.call_args.kwargs["params"] + assert params["version"] == "v1.2.3" + assert params["format"] == "cyclonedx" + assert "/api/v1/components/c1/sboms" in session.request.call_args.args[1] + + +def test_find_component_sbom_returns_newest_match() -> None: + client, _ = _client_with( + [ + _FakeResponse( + 200, + { + "items": [ + {"sbom": {"id": "newest", "version": "v1.2.3", "format": "cyclonedx"}}, + {"sbom": {"id": "older", "version": "v1.2.3", "format": "cyclonedx"}}, + ] + }, + ) + ] + ) + assert client.find_component_sbom("c1", "v1.2.3", "cyclonedx") == "newest" + + +def test_find_component_sbom_returns_none_on_miss() -> None: + client, _ = _client_with([_FakeResponse(200, {"items": []})]) + assert client.find_component_sbom("c1", "v9.9.9", "cyclonedx") is None + + +def test_find_component_sbom_rechecks_filters_client_side() -> None: + """A backend without the server-side filters (pre sbomify#1176) + ignores the params and returns the full unfiltered listing — the + client must not match the wrong version/format.""" + client, _ = _client_with( + [ + _FakeResponse( + 200, + { + "items": [ + {"sbom": {"id": "wrong-version", "version": "8fae865", "format": "cyclonedx"}}, + {"sbom": {"id": "wrong-format", "version": "v1.2.3", "format": "spdx"}}, + {"sbom": {"id": "match", "version": "v1.2.3", "format": "cyclonedx"}}, + {"not-sbom-shaped": True}, + ] + }, + ) + ] + ) + assert client.find_component_sbom("c1", "v1.2.3", "cyclonedx") == "match" diff --git a/tests/test_submodule.py b/tests/test_submodule.py new file mode 100644 index 00000000..97fcaf00 --- /dev/null +++ b/tests/test_submodule.py @@ -0,0 +1,228 @@ +"""Tests for submodule pin resolution and the pipeline's submodule mode.""" + +from __future__ import annotations + +import os +import subprocess +from importlib import import_module +from pathlib import Path + +import pytest + +from sbomify_action.cli.main import ( + Config, + _find_existing_submodule_sbom, + _prepare_submodule_mode, +) +from sbomify_action.submodule import SubmodulePin, _pick_version_tag, resolve_submodule_pin + +# `sbomify_action.cli.main` the *attribute* is shadowed by a function +# export, so monkeypatch string paths can't reach the module — import it +# explicitly (same workaround as tests/test_config.py). +cli_main_module = import_module("sbomify_action.cli.main") + +# ---------------------------------------------------------------------- +# git fixtures + + +def _git_env() -> dict[str, str]: + return { + **os.environ, + "GIT_AUTHOR_NAME": "T", + "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "T", + "GIT_COMMITTER_EMAIL": "t@t", + # Never pick up the developer's/CI's global config. + "GIT_CONFIG_GLOBAL": "/dev/null", + "GIT_CONFIG_SYSTEM": "/dev/null", + } + + +def _git(args: list[str], cwd: Path) -> str: + result = subprocess.run( + ["git", *args], + cwd=cwd, + env=_git_env(), + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def _make_repo(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True) + _git(["init", "-q", "-b", "main"], cwd=path) + (path / "README.md").write_text("# repo") + _git(["add", "."], cwd=path) + _git(["commit", "-q", "-m", "initial"], cwd=path) + + +@pytest.fixture +def parent_with_submodule(tmp_path: Path) -> tuple[Path, Path]: + """A parent repo with ``extern/lib`` as a real git submodule. + + The submodule's pinned commit carries tag ``v1.2.3``. + """ + sub = tmp_path / "sub-remote" + _make_repo(sub) + _git(["tag", "v1.2.3"], cwd=sub) + + parent = tmp_path / "parent" + _make_repo(parent) + # protocol.file.allow: git >= 2.38 blocks file-protocol submodules by default. + _git( + ["-c", "protocol.file.allow=always", "submodule", "add", "-q", str(sub), "extern/lib"], + cwd=parent, + ) + _git(["commit", "-q", "-m", "add submodule"], cwd=parent) + return parent, sub + + +# ---------------------------------------------------------------------- +# resolve_submodule_pin + + +def test_resolves_version_tag_at_pin(parent_with_submodule: tuple[Path, Path]) -> None: + parent, sub = parent_with_submodule + pin = resolve_submodule_pin(parent, "extern/lib") + assert pin is not None + assert pin.path == "extern/lib" + assert pin.version == "v1.2.3" + assert pin.version_source == "tag" + assert pin.sha == _git(["rev-parse", "HEAD"], cwd=sub) + + +def test_resolves_short_sha_for_untagged_pin(parent_with_submodule: tuple[Path, Path]) -> None: + parent, sub = parent_with_submodule + # Advance the submodule remote past the tag and re-pin the parent. + (sub / "new.txt").write_text("x") + _git(["add", "."], cwd=sub) + _git(["commit", "-q", "-m", "untagged"], cwd=sub) + new_sha = _git(["rev-parse", "HEAD"], cwd=sub) + checkout = parent / "extern" / "lib" + _git(["fetch", "-q", "origin"], cwd=checkout) + _git(["checkout", "-q", new_sha], cwd=checkout) + _git(["add", "extern/lib"], cwd=parent) + _git(["commit", "-q", "-m", "bump submodule"], cwd=parent) + + pin = resolve_submodule_pin(parent, "extern/lib") + assert pin is not None + assert pin.sha == new_sha + assert pin.version == new_sha[:7] + assert pin.version_source == "sha" + + +def test_pin_read_from_tree_without_initialized_submodule(parent_with_submodule: tuple[Path, Path]) -> None: + """The gitlink lives in the parent tree — a fresh clone without + ``submodule update`` must still resolve (the attach path needs no + submodule checkout at all).""" + parent, sub = parent_with_submodule + clone = parent.parent / "fresh-clone" + _git(["clone", "-q", str(parent), str(clone)], cwd=parent.parent) + + pin = resolve_submodule_pin(clone, "extern/lib") + assert pin is not None + assert pin.sha == _git(["rev-parse", "HEAD"], cwd=sub) + # .gitmodules still points at the sub remote, so the tag resolves too. + assert pin.version == "v1.2.3" + + +def test_resolves_vendored_clone_without_gitmodules(tmp_path: Path) -> None: + """A plain checked-in clone (own .git dir, no gitlink) resolves via + its embedded repo, including local tags.""" + sub = tmp_path / "sub-remote" + _make_repo(sub) + _git(["tag", "2026.7.1"], cwd=sub) + + parent = tmp_path / "parent" + _make_repo(parent) + _git(["clone", "-q", str(sub), str(parent / "third_party" / "libfoo")], cwd=parent) + + pin = resolve_submodule_pin(parent, "third_party/libfoo") + assert pin is not None + assert pin.version == "2026.7.1" + assert pin.version_source == "tag" + + +def test_returns_none_for_plain_directory(tmp_path: Path) -> None: + parent = tmp_path / "parent" + _make_repo(parent) + (parent / "just-a-dir").mkdir() + assert resolve_submodule_pin(parent, "just-a-dir") is None + assert resolve_submodule_pin(parent, "missing/path") is None + + +def test_pick_version_tag_prefers_v_prefixed() -> None: + assert _pick_version_tag(["1.2.3", "v1.2.3"]) == "v1.2.3" + assert _pick_version_tag(["2026.7.1"]) == "2026.7.1" + # Non-version tags (release names, "latest") never win. + assert _pick_version_tag(["latest", "rc-final"]) is None + assert _pick_version_tag([]) is None + + +# ---------------------------------------------------------------------- +# pipeline submodule mode + + +def _submodule_config(**overrides: object) -> Config: + defaults: dict[str, object] = { + "token": "t", + "component_id": "c1", + "lock_file": "extern/lib/Cargo.lock", + "submodule_path": "extern/lib", + "sbom_format": "cyclonedx", + } + defaults.update(overrides) + return Config(**defaults) # type: ignore[arg-type] + + +def test_prepare_submodule_mode_attaches_existing(monkeypatch: pytest.MonkeyPatch) -> None: + """Pin resolves + SBOM exists → returns its id and overrides the + component version with the pin-derived one.""" + monkeypatch.setattr( + "sbomify_action.submodule.resolve_submodule_pin", + lambda _root, path: SubmodulePin(path=path, sha="a" * 40, version="v1.2.3", version_source="tag"), + ) + monkeypatch.setattr(cli_main_module, "_find_existing_submodule_sbom", lambda _config, _fmt: "sbom-42") + config = _submodule_config(component_version="parent-version") + assert _prepare_submodule_mode(config) == "sbom-42" + assert config.component_version == "v1.2.3" + + +def test_prepare_submodule_mode_falls_back_to_backfill(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "sbomify_action.submodule.resolve_submodule_pin", + lambda _root, path: SubmodulePin(path=path, sha="b" * 40, version="bbbbbbb", version_source="sha"), + ) + monkeypatch.setattr(cli_main_module, "_find_existing_submodule_sbom", lambda _config, _fmt: None) + config = _submodule_config() + assert _prepare_submodule_mode(config) is None + assert config.component_version == "bbbbbbb" + + +def test_prepare_submodule_mode_exits_on_unresolvable_path(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("sbomify_action.submodule.resolve_submodule_pin", lambda _root, _path: None) + with pytest.raises(SystemExit): + _prepare_submodule_mode(_submodule_config()) + + +def test_find_existing_submodule_sbom_soft_fails(monkeypatch: pytest.MonkeyPatch) -> None: + """API errors during the lookup mean backfill, not a crashed run.""" + from sbomify_action.exceptions import APIError + + class _BoomClient: + def __init__(self, *args: object, **kwargs: object) -> None: + pass + + def find_component_sbom(self, *args: object) -> str: + raise APIError("down") + + monkeypatch.setattr("sbomify_action.sbomify_api.SbomifyApiClient", _BoomClient) + config = _submodule_config(component_version="v1.2.3") + assert _find_existing_submodule_sbom(config, "cyclonedx") is None + + +def test_find_existing_submodule_sbom_requires_credentials() -> None: + config = _submodule_config(token="", component_version="v1.2.3") + assert _find_existing_submodule_sbom(config, "cyclonedx") is None diff --git a/tests/test_wizard_discovery.py b/tests/test_wizard_discovery.py index f6efc780..b98970f8 100644 --- a/tests/test_wizard_discovery.py +++ b/tests/test_wizard_discovery.py @@ -124,6 +124,61 @@ def test_discover_skips_agent_worktree_dirs(tmp_path: Path) -> None: assert [str(lf.rel_path) for lf in found] == ["uv.lock"] +def test_discover_annotates_submodule_lockfiles(tmp_path: Path) -> None: + # Lockfiles under a .gitmodules-declared path belong to another repo — + # they must be flagged so the wizard can steer the user to set up + # SBOMs there instead. + (tmp_path / ".gitmodules").write_text('[submodule "lib"]\n\tpath = extern/lib\n\turl = git@example.com:lib.git\n') + sub = tmp_path / "extern" / "lib" + (sub / "nested").mkdir(parents=True) + (sub / "uv.lock").write_text("") + (sub / "nested" / "bun.lock").write_text("") # deeper inside the submodule + (tmp_path / "uv.lock").write_text("") + + found = discover(tmp_path) + by_rel = {str(lf.rel_path): lf for lf in found} + assert by_rel["uv.lock"].nested_repo is None + assert by_rel["uv.lock"].nested_repo_kind is None + for rel in (os.path.join("extern", "lib", "uv.lock"), os.path.join("extern", "lib", "nested", "bun.lock")): + assert by_rel[rel].nested_repo == "extern/lib" + assert by_rel[rel].nested_repo_kind == "submodule" + + +def test_discover_annotates_vendored_git_clone(tmp_path: Path) -> None: + # A checked-in clone (its own .git dir, not declared in .gitmodules) + # is flagged as vendored. Submodule checkouts use a .git *file*, so + # that shape must be detected too. + clone = tmp_path / "third_party" / "libfoo" + clone.mkdir(parents=True) + (clone / ".git").mkdir() + (clone / "Cargo.lock").write_text("") + + gitfile_repo = tmp_path / "extern" / "bar" + gitfile_repo.mkdir(parents=True) + (gitfile_repo / ".git").write_text("gitdir: ../../.git/modules/bar\n") + (gitfile_repo / "go.sum").write_text("") + + found = discover(tmp_path) + by_rel = {str(lf.rel_path): lf for lf in found} + assert by_rel[os.path.join("third_party", "libfoo", "Cargo.lock")].nested_repo == "third_party/libfoo" + assert by_rel[os.path.join("third_party", "libfoo", "Cargo.lock")].nested_repo_kind == "vendored" + assert by_rel[os.path.join("extern", "bar", "go.sum")].nested_repo == "extern/bar" + assert by_rel[os.path.join("extern", "bar", "go.sum")].nested_repo_kind == "vendored" + + +def test_discover_gitmodules_quoted_path(tmp_path: Path) -> None: + # git quotes submodule paths containing spaces — the parser must strip + # the quotes before comparing. + (tmp_path / ".gitmodules").write_text('[submodule "spacey"]\n\tpath = "my lib"\n\turl = u\n') + sub = tmp_path / "my lib" + sub.mkdir() + (sub / "uv.lock").write_text("") + + found = discover(tmp_path) + assert found[0].nested_repo == "my lib" + assert found[0].nested_repo_kind == "submodule" + + def test_discover_suggested_name_includes_repo_and_ecosystem(tmp_path: Path) -> None: (tmp_path / "uv.lock").write_text("") found = discover(tmp_path, repo_name="My Widget!") diff --git a/tests/test_wizard_emitter.py b/tests/test_wizard_emitter.py index 7b017909..e4df1065 100644 --- a/tests/test_wizard_emitter.py +++ b/tests/test_wizard_emitter.py @@ -260,6 +260,99 @@ def test_emit_no_attestation_by_default(tmp_path: Path) -> None: yaml = emit_workflow(plan, facts=facts, api_base_url="https://app.sbomify.com") assert "attestations: write" not in yaml assert "attest-build-provenance" not in yaml + # The per-row attest flag only exists when attestation is enabled. + assert "attest:" not in yaml + + +def test_emit_attestation_gates_rows_and_step(tmp_path: Path) -> None: + """Every matrix row carries an ``attest`` boolean and the attest step + is conditioned on it, so submodule rows can opt out per-entry.""" + facts = _facts(tmp_path) + plan = Plan( + use_product_id="prod-1", + attestation=True, + create_components=[PlannedComponent(lockfile=_python_lockfile(tmp_path), name="widget-py")], + ) + yaml = emit_workflow(plan, facts=facts, api_base_url="https://app.sbomify.com") + assert " attest: true\n" in yaml + assert " if: ${{ matrix.attest }}\n" in yaml + + +def test_emit_attestation_skips_nested_repo_lockfiles(tmp_path: Path) -> None: + """SBOMs for submodule / vendored lockfiles must NOT be attested: the + Sigstore identity would be this repo's workflow, but the code belongs + to another repository, so the attestation would fail verification + against the repo the code actually comes from.""" + facts = _facts(tmp_path) + sub = DiscoveredLockfile( + path=tmp_path / "extern" / "lib" / "Cargo.lock", + rel_path=Path("extern") / "lib" / "Cargo.lock", + ecosystem="rust", + suggested_name="widget-rust", + nested_repo="extern/lib", + nested_repo_kind="submodule", + ) + plan = Plan( + use_product_id="prod-1", + attestation=True, + create_components=[ + PlannedComponent(lockfile=_python_lockfile(tmp_path), name="widget-py"), + PlannedComponent(lockfile=sub, name="widget-rust"), + ], + ) + yaml = emit_workflow(plan, facts=facts, api_base_url="https://app.sbomify.com") + assert " attest: true\n" in yaml # own lockfile still attests + assert " attest: false\n" in yaml # submodule row opts out + assert " if: ${{ matrix.attest }}\n" in yaml + # The opt-out must be self-documenting in the YAML: an explicit + # "deliberately not signed" annotation naming the nested repo. + assert "Deliberately NOT signed: extern/lib is a submodule" in yaml + # The row-level flag order must match the component order: widget-py + # (attest: true) before widget-rust (attest: false). + assert yaml.index("attest: true") < yaml.index("attest: false") + + +def _nested_lockfile(tmp_path: Path) -> DiscoveredLockfile: + return DiscoveredLockfile( + path=tmp_path / "extern" / "lib" / "Cargo.lock", + rel_path=Path("extern") / "lib" / "Cargo.lock", + ecosystem="rust", + suggested_name="widget-rust", + nested_repo="extern/lib", + nested_repo_kind="submodule", + ) + + +def test_emit_submodule_rows_drive_attach_or_backfill(tmp_path: Path) -> None: + """Nested-repo lockfiles get a submodule_path matrix field, the env + block forwards it as SUBMODULE_PATH, and the checkout pulls + submodules so the backfill path can generate.""" + facts = _facts(tmp_path) + plan = Plan( + use_product_id="prod-1", + create_components=[ + PlannedComponent(lockfile=_python_lockfile(tmp_path), name="widget-py"), + PlannedComponent(lockfile=_nested_lockfile(tmp_path), name="widget-rust"), + ], + ) + yaml = emit_workflow(plan, facts=facts, api_base_url="https://app.sbomify.com") + assert " submodule_path: extern/lib\n" in yaml + assert " SUBMODULE_PATH: ${{ matrix.submodule_path }}\n" in yaml + assert " submodules: recursive\n" in yaml + # Exactly one row carries the field — the non-submodule row must not. + assert yaml.count("submodule_path:") == 1 + + +def test_emit_no_submodule_plumbing_without_nested_lockfiles(tmp_path: Path) -> None: + facts = _facts(tmp_path) + plan = Plan( + use_product_id="prod-1", + create_components=[PlannedComponent(lockfile=_python_lockfile(tmp_path), name="widget-py")], + ) + yaml = emit_workflow(plan, facts=facts, api_base_url="https://app.sbomify.com") + assert "submodule_path" not in yaml + assert "SUBMODULE_PATH" not in yaml + assert "submodules: recursive" not in yaml def test_emit_cache_step_always_present(tmp_path: Path) -> None: diff --git a/tests/test_wizard_textual.py b/tests/test_wizard_textual.py index 85abc35d..0ace99f7 100644 --- a/tests/test_wizard_textual.py +++ b/tests/test_wizard_textual.py @@ -209,6 +209,50 @@ async def test_welcome_to_discover_navigates(tmp_path: Path, monkeypatch: pytest assert len(app.state.discovered) == 1 +async def test_discover_deselects_nested_repo_lockfiles_by_default( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Submodule / vendored-repo lockfiles are annotated and start + deselected — they belong to another repository and should get their + own SBOM setup there.""" + from textual.widgets import SelectionList + + lockfiles = [ + DiscoveredLockfile( + path=tmp_path / "uv.lock", + rel_path=Path("uv.lock"), + ecosystem="python", + suggested_name="widget-py", + ), + DiscoveredLockfile( + path=tmp_path / "extern" / "lib" / "Cargo.lock", + rel_path=Path("extern") / "lib" / "Cargo.lock", + ecosystem="rust", + suggested_name="widget-rust", + nested_repo="extern/lib", + nested_repo_kind="submodule", + ), + ] + _stub_discovery(monkeypatch, lockfiles) + + app = WizardApp(_opts(tmp_path)) + async with app.run_test() as pilot: + await pilot.press("enter") # welcome → discover + await pilot.pause() + + from sbomify_action.cli.wizard.screens.discover import DiscoverScreen + + assert isinstance(app.screen, DiscoverScreen) + sel = app.screen.query_one("#lockfile-list", SelectionList) + # Only the top-level lockfile is pre-selected. + assert list(sel.selected) == [0] + # The submodule row carries the annotation, and the explanatory + # note is present. + labels = [str(sel.get_option_at_index(i).prompt) for i in range(sel.option_count)] + assert any("submodule: extern/lib" in label for label in labels) + app.screen.query_one("#nested-repo-note") + + async def test_enter_on_focused_radio_set_toggles_radio(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """Regression: Enter while a RadioSet has focus must commit the highlighted radio (not skip past the whole screen).