From d0555149fb851e4351369305e1bc28000ee1bd74 Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Mon, 20 Jul 2026 15:16:32 +0000 Subject: [PATCH 1/4] feat(wizard): annotate submodule/vendored lockfiles in discovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lockfiles that live inside a git submodule (declared in .gitmodules) or a vendored clone (a directory with its own .git) belong to another repository and are better tracked from there. The discover screen now annotates those rows with the enclosing repo path and leaves them deselected by default, with a note explaining why. Detection parses .gitmodules directly (works without a git binary and for declared-but-uninitialized submodules) and checks ancestors for a .git entry — a file for submodule checkouts, a directory for vendored clones. Co-Authored-By: Claude Fable 5 --- sbomify_action/cli/wizard/discovery.py | 56 ++++++++++++++++++- sbomify_action/cli/wizard/screens/discover.py | 15 ++++- sbomify_action/cli/wizard/state.py | 15 +++++ tests/test_wizard_discovery.py | 55 ++++++++++++++++++ tests/test_wizard_textual.py | 44 +++++++++++++++ 5 files changed, 183 insertions(+), 2 deletions(-) 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/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_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). From 45e4563872958cb4365a0b8b394248ab65c1975f Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Tue, 21 Jul 2026 11:00:37 +0000 Subject: [PATCH 2/4] feat(wizard): skip attestation for submodule/vendored lockfiles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An attest-build-provenance signature carries the Sigstore identity of the workflow that ran it. For lockfiles inside a submodule or vendored checkout, that identity is the parent repo — but the SBOM describes code owned by another repository, so the attestation would fail verification against the repo the code actually comes from. When attestation is enabled, every matrix row now carries an 'attest' boolean and the attest step is gated on it; nested-repo rows get attest: false. This is also the per-entry hook the upcoming attach-or-backfill submodule flow will need. Co-Authored-By: Claude Fable 5 --- sbomify_action/cli/wizard/ci_emitter.py | 24 +++++++++++-- tests/test_wizard_emitter.py | 47 +++++++++++++++++++++++++ 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/sbomify_action/cli/wizard/ci_emitter.py b/sbomify_action/cli/wizard/ci_emitter.py index 07b1cda9..4f6cc077 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,10 @@ def _matrix_block( " sbom_format: " + fmt + "\n" " output_file: " + output_file + "\n" ) + if attestation: + attest = "false" if c.lockfile.nested_repo else "true" + row += " attest: " + attest + "\n" + rows.append(row) return "".join(rows) @@ -469,7 +483,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" ) @@ -521,7 +541,7 @@ def emit_workflow( release_strategy=plan.release_strategy, product_id=product_id or plan.use_product_id, ) - 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 "" # Resolved by the caller (apply / review) so the GitHub lookup happens diff --git a/tests/test_wizard_emitter.py b/tests/test_wizard_emitter.py index 7b017909..25ea58c6 100644 --- a/tests/test_wizard_emitter.py +++ b/tests/test_wizard_emitter.py @@ -260,6 +260,53 @@ 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 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 test_emit_cache_step_always_present(tmp_path: Path) -> None: From 9e4425cd59e7ff5d3200ab1e047800bc778b424c Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Tue, 21 Jul 2026 11:55:54 +0000 Subject: [PATCH 3/4] feat(wizard): annotate the attestation opt-out inline on matrix rows Each attest: false row now carries an explicit 'Deliberately NOT signed' comment naming the nested repo and why, so the generated YAML documents the decision where it takes effect, not only at the attest step. Co-Authored-By: Claude Fable 5 --- sbomify_action/cli/wizard/ci_emitter.py | 13 +++++++++++-- tests/test_wizard_emitter.py | 3 +++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/sbomify_action/cli/wizard/ci_emitter.py b/sbomify_action/cli/wizard/ci_emitter.py index 4f6cc077..f8717ab5 100644 --- a/sbomify_action/cli/wizard/ci_emitter.py +++ b/sbomify_action/cli/wizard/ci_emitter.py @@ -307,8 +307,17 @@ def _matrix_block( " output_file: " + output_file + "\n" ) if attestation: - attest = "false" if c.lockfile.nested_repo else "true" - row += " attest: " + attest + "\n" + 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) diff --git a/tests/test_wizard_emitter.py b/tests/test_wizard_emitter.py index 25ea58c6..cde9e195 100644 --- a/tests/test_wizard_emitter.py +++ b/tests/test_wizard_emitter.py @@ -304,6 +304,9 @@ def test_emit_attestation_skips_nested_repo_lockfiles(tmp_path: Path) -> None: 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") From 38b1e4fafebda54517ce6f201ed466fb419f52b4 Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Tue, 21 Jul 2026 12:08:32 +0000 Subject: [PATCH 4/4] feat: attach-or-backfill submodule SBOMs at release time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New SUBMODULE_PATH mode for the action: the component is a git submodule pinned in the parent tree. At run time the action resolves the pin to a version — an exact version tag on the pinned commit (resolved via git ls-remote, no submodule checkout needed), else the 7-char short SHA, matching what trunk-strategy workflows publish. Exact match only; no nearest-tag guessing. If the submodule's component already has an SBOM at that (version, format) — looked up via the new GET filters from sbomify#1176 — the action skips generation and upload entirely and just tags the existing SBOM into the configured product release(s). Otherwise it generates from the submodule's lockfile and uploads under the pin-derived version (backfill), so the next release with the same pin takes the attach path. A DUPLICATE_ARTIFACT on upload (two parents racing to backfill the same pin) recovers by re-looking up the winner and attaching it. - sbomify_api: list_component_sboms / find_component_sbom, with client-side re-checking of the filters for backends that predate the server-side ones - submodule.py: pin + version resolution (gitlink from the parent tree, .gitmodules URL via git config, ls-remote tag matching with local-tag fallback, vendored-clone support) - pipeline: preflight before Step 1, Step 6 + finalization extracted into reusable helpers, duplicate-upload recovery in submodule mode - wizard emitter: submodule rows carry submodule_path, the env block forwards SUBMODULE_PATH, and checkout pulls submodules for the backfill path - README: document SUBMODULE_PATH Co-Authored-By: Claude Fable 5 --- README.md | 1 + sbomify_action/cli/main.py | 304 ++++++++++++++++++------ sbomify_action/cli/wizard/ci_emitter.py | 26 +- sbomify_action/sbomify_api.py | 43 ++++ sbomify_action/submodule.py | 216 +++++++++++++++++ tests/test_config.py | 54 +++++ tests/test_sbomify_api.py | 60 +++++ tests/test_submodule.py | 228 ++++++++++++++++++ tests/test_wizard_emitter.py | 43 ++++ 9 files changed, 896 insertions(+), 79 deletions(-) create mode 100644 sbomify_action/submodule.py create mode 100644 tests/test_submodule.py 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 f8717ab5..5ee004f4 100644 --- a/sbomify_action/cli/wizard/ci_emitter.py +++ b/sbomify_action/cli/wizard/ci_emitter.py @@ -306,6 +306,12 @@ 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" @@ -395,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. @@ -425,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 @@ -538,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) @@ -549,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, 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 @@ -573,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/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_emitter.py b/tests/test_wizard_emitter.py index cde9e195..e4df1065 100644 --- a/tests/test_wizard_emitter.py +++ b/tests/test_wizard_emitter.py @@ -312,6 +312,49 @@ def test_emit_attestation_skips_nested_repo_lockfiles(tmp_path: Path) -> None: 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: facts = _facts(tmp_path) plan = Plan(