From 703611aee895dd2a2ec3e19c09e51f216816af28 Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Thu, 23 Jul 2026 12:53:17 +0000 Subject: [PATCH 1/4] fix(wizard): handle plan limits, workspace scoping, and clean errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes several wizard failures surfaced when onboarding a workspace whose token is scoped to a non-default workspace and hits a plan limit. - Bind contact-profile prefetch to the workspace the token can actually read: the auth screen now verifies the is_default_team-picked workspace and probes the rest on a 403, instead of silently treating the scope denial as "0 profiles" (which showed an empty picker and created new profiles in the wrong workspace). - Give product/component plan-limit failures a real recovery path on the Apply screen: reuse the single existing product & retry in place, or jump Back to the Pick-a-product / Components step — not a dead-end "retry". - get-or-create semantics for create_product and create_contact_profile so a retry after a partial apply (or a lost-response resubmit) reuses the existing resource instead of dying on DUPLICATE_NAME. - Detect plan limits via the backend's BILLING_LIMIT_EXCEEDED error code; tag PlanLimitError with the resource so the UI can tailor the CTA. - Strip HTTP status codes from every user-facing error string (apply banner/log, create-profile status, auth status); full codes still go to the debug log. - Rewrite British spellings to US English across the package and tests. Co-Authored-By: Claude Opus 4.8 --- sbomify_action/_enrichment/metadata.py | 8 +- sbomify_action/_enrichment/sources/pypi.py | 4 +- sbomify_action/_generation/utils.py | 2 +- sbomify_action/cli/main.py | 2 +- sbomify_action/cli/wizard/apply.py | 9 +- sbomify_action/cli/wizard/io.py | 6 +- sbomify_action/cli/wizard/screens/_base.py | 18 ++ sbomify_action/cli/wizard/screens/apply.py | 211 ++++++++++++++---- .../cli/wizard/screens/authenticate.py | 114 +++++++--- .../cli/wizard/screens/configure_sbom.py | 18 +- .../wizard/screens/configure_sbomify_json.py | 2 +- .../cli/wizard/screens/create_profile.py | 26 ++- sbomify_action/cli/wizard/screens/done.py | 4 +- sbomify_action/cli/wizard/screens/review.py | 2 +- sbomify_action/cli/wizard/screens/welcome.py | 2 +- sbomify_action/cli/wizard/state.py | 2 +- sbomify_action/cli/wizard/styles.tcss | 2 +- .../cli/wizard/widgets/pick_or_create.py | 2 +- .../cli/wizard/widgets/stateful_radio.py | 10 +- sbomify_action/enrichment.py | 42 ++-- sbomify_action/exceptions.py | 11 +- sbomify_action/sbomify_api.py | 116 ++++++++-- sbomify_action/spdx3.py | 6 +- tests/test_enrichment_module.py | 42 ++-- tests/test_sbomify_api.py | 86 ++++++- tests/test_upload_module.py | 2 +- tests/test_wizard_emitter.py | 2 +- tests/test_wizard_state.py | 87 +++++++- tests/test_wizard_textual.py | 104 ++++++++- 29 files changed, 755 insertions(+), 187 deletions(-) diff --git a/sbomify_action/_enrichment/metadata.py b/sbomify_action/_enrichment/metadata.py index 6c7c0c56..3a0a0a5e 100644 --- a/sbomify_action/_enrichment/metadata.py +++ b/sbomify_action/_enrichment/metadata.py @@ -44,15 +44,15 @@ class NormalizedMetadata: # Distribution info (BSI TR-03183-2 compliance) distribution_filename: Optional[str] = None - # Hashes of the deployable artefact, keyed by algorithm name as the + # Hashes of the deployable artifact, keyed by algorithm name as the # source provided it. Keys are lower-case strings — typically SPDX / # CycloneDX canonical names such as "sha256", "sha512", "md5", but # some sources legitimately use their own variant spelling. The PyPI # source, for instance, returns BLAKE2b-256 under the underscore # form "blake2b_256" (matching PyPI's JSON API), and we keep that # form here so the downstream algorithm map (_CYCLONEDX_HASH_ALGORITHMS / - # _SPDX_CHECKSUM_ALGORITHMS in enrichment.py) can recognise it - # without a second normalisation step. Values are lower-case hex + # _SPDX_CHECKSUM_ALGORITHMS in enrichment.py) can recognize it + # without a second normalization step. Values are lower-case hex # strings. Used to populate NTIA / BSI §5.2.2 / CISA "Component # Hash" elements. hashes: Dict[str, str] = field(default_factory=dict) @@ -102,7 +102,7 @@ def pick(field_name: str, self_val: Any, other_val: Any, is_list: bool = False) # conflicting algorithms, but every algorithm contributed by # `other` is preserved. Dropping other's keys would silently lose # useful digests when PyPI gives sha256 and another source - # contributes blake2b / md5 for the same artefact. + # contributes blake2b / md5 for the same artifact. # # Attribution for the union field: if `other` adds at least one # new algorithm key, refresh ``field_sources["hashes"]`` to diff --git a/sbomify_action/_enrichment/sources/pypi.py b/sbomify_action/_enrichment/sources/pypi.py index 0c478a69..d790a582 100644 --- a/sbomify_action/_enrichment/sources/pypi.py +++ b/sbomify_action/_enrichment/sources/pypi.py @@ -109,7 +109,7 @@ def fetch(self, purl: PackageURL, session: requests.Session) -> Optional[Normali # 3. Build the cache key from the now-validated name/version. # # Path-traversal rules: - # - Names: PEP 503 allows "." and "_" inside a normalised project + # - Names: PEP 503 allows "." and "_" inside a normalized project # name (so "a..b" is a legal-but-unusual name). Reject only # actual path separators ("/", "\\") and the whole-name # dot-segments ("." / ".."). The ":" character is refused too @@ -321,7 +321,7 @@ def _normalize_response(self, package_name: str, data: Dict[str, Any]) -> Normal # (BSI TR-03183-2 §5.2.2 filename + hash / NTIA / CISA hash element). # Prefer wheel (.whl) over sdist (.tar.gz); take the hashes of the # distribution whose filename we record so the two fields describe - # the same artefact. + # the same artifact. distribution_filename = None distribution_hashes: Dict[str, str] = {} urls = data.get("urls", []) diff --git a/sbomify_action/_generation/utils.py b/sbomify_action/_generation/utils.py index fa1e0e02..b4271240 100644 --- a/sbomify_action/_generation/utils.py +++ b/sbomify_action/_generation/utils.py @@ -414,7 +414,7 @@ def log_progress() -> None: ) except subprocess.TimeoutExpired: elapsed = int(time.time() - start_time) - # Honour log_errors here too: a cdxgen timeout on, say, a Python + # Honor log_errors here too: a cdxgen timeout on, say, a Python # lockfile is the same benign priority-chain fallback as a non-zero # exit — it shouldn't spam red ERROR when a later generator succeeds. timeout_msg = f"{command_name} command timed out after {elapsed}s (limit: {timeout}s)" diff --git a/sbomify_action/cli/main.py b/sbomify_action/cli/main.py index 42e468c0..4630907d 100644 --- a/sbomify_action/cli/main.py +++ b/sbomify_action/cli/main.py @@ -1236,7 +1236,7 @@ def _write_final_output(src_path: str, dst_path: str, bom_type: Optional[str]) - """Write the final artifact to ``dst_path``. Non-SBOM artifacts (VEX, CBOM, ...) are copied byte-for-byte so the upload matches the - authored file exactly (a text round-trip could normalise newlines). SBOMs go through the + authored file exactly (a text round-trip could normalize newlines). SBOMs go through the text path so the CycloneDX fixups in :func:`_finalize_output_content` apply. """ if bom_type and bom_type != "sbom": diff --git a/sbomify_action/cli/wizard/apply.py b/sbomify_action/cli/wizard/apply.py index 66d97cce..411cc463 100644 --- a/sbomify_action/cli/wizard/apply.py +++ b/sbomify_action/cli/wizard/apply.py @@ -405,7 +405,7 @@ def _per_component_best_effort( relative-path to error message. Catches only ``APIError`` (which is the parent of ``AuthError`` per exceptions.py and includes the ConnectionError/Timeout shims in sbomify_api._request); anything - else escapes and aborts the apply, which is the right behaviour for + else escapes and aborts the apply, which is the right behavior for programming errors. ``label`` is interpolated into per-failure warning messages ("Could not {label} for foo (cid): …"). """ @@ -549,10 +549,11 @@ def _resolve_product(state: WizardState, log: LogFn) -> dict[str, Any] | None: assert state.workspace is not None # narrowed by caller if plan.create_product: - product = api.create_product(plan.create_product) + product, was_created = api.create_product(plan.create_product) state.created_product_id = str(product.get("id") or "") - state.applied.append(f"created product {product.get('name')}") - log("success", f"Created product {product.get('name')}") + verb = "Created" if was_created else "Reused existing" + state.applied.append(f"{verb.lower()} product {product.get('name')}") + log("success" if was_created else "info", f"{verb} product {product.get('name')}") return product if plan.use_product_id: diff --git a/sbomify_action/cli/wizard/io.py b/sbomify_action/cli/wizard/io.py index bb48a5ea..b1699101 100644 --- a/sbomify_action/cli/wizard/io.py +++ b/sbomify_action/cli/wizard/io.py @@ -30,7 +30,7 @@ # emitted line appends the generating build's version (eg. # ``# sbomify-action wizard v26.7.0``; older releases wrote a literal # ``v1``), so ownership detection matches this version-agnostic -# prefix — files stamped by any wizard version stay recognised. +# prefix — files stamped by any wizard version stay recognized. WIZARD_HEADER_SENTINEL = "# sbomify-action wizard" # Top-level key on every wizard-generated sbomify.json. Its presence @@ -124,8 +124,8 @@ def write_sbomify_json(path: Path, payload: dict[str, Any]) -> None: Mirrors ``write_workflow``: refuses to overwrite a file that exists but lacks the wizard sentinel key, so a hand-crafted sbomify.json is never silently clobbered. The sentinel key is injected before - serialisation; the action's json_config provider ignores unknown - top-level keys so its presence doesn't change runtime behaviour. + serialization; the action's json_config provider ignores unknown + top-level keys so its presence doesn't change runtime behavior. When the existing wizard-stamped file's content differs from what we're about to write, the previous version is saved next to it as diff --git a/sbomify_action/cli/wizard/screens/_base.py b/sbomify_action/cli/wizard/screens/_base.py index 3f906e05..a7a51bfd 100644 --- a/sbomify_action/cli/wizard/screens/_base.py +++ b/sbomify_action/cli/wizard/screens/_base.py @@ -13,6 +13,7 @@ from __future__ import annotations +import re from typing import TYPE_CHECKING, Callable, ClassVar from textual import events @@ -28,6 +29,23 @@ TOTAL_STEPS = 8 +# ``[403] - `` / ``[404]`` markers that API error strings embed for logs. +# The wizard strips them from anything it shows the user — the status code +# is developer-facing noise next to the human-readable detail. +_STATUS_CODE_RE = re.compile(r"\s*\[\d{3}\]\s*-?\s*") + + +def strip_status_codes(message: str) -> str: + """Remove HTTP status-code markers from an API error message. + + ``"Failed to create product 'X'. [403] - You have reached…"`` becomes + ``"Failed to create product 'X'. You have reached…"``. Full error text + (codes included) still lands in the debug log via the exception itself; + this only cleans what the TUI renders. + """ + return _STATUS_CODE_RE.sub(" ", message).strip() + + # Responsive breakpoints (terminal cells). The wizard is a fit-to-viewport # TUI — nothing scrolls — so every screen has to render inside whatever the # terminal gives us. Three tiers, each guaranteed to fit at its lower bound: diff --git a/sbomify_action/cli/wizard/screens/apply.py b/sbomify_action/cli/wizard/screens/apply.py index 2a304d3b..4ec86e74 100644 --- a/sbomify_action/cli/wizard/screens/apply.py +++ b/sbomify_action/cli/wizard/screens/apply.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Any + from rich.markup import escape as rich_escape from textual.app import ComposeResult from textual.binding import Binding @@ -10,9 +12,10 @@ from textual.worker import Worker, WorkerState from sbomify_action.cli.wizard import apply as apply_mod -from sbomify_action.cli.wizard.screens._base import WizardScreen +from sbomify_action.cli.wizard.screens._base import WizardScreen, strip_status_codes +from sbomify_action.exceptions import PlanLimitError -# Log-line colours pulled from the sbomify marketing palette so they +# Log-line colors pulled from the sbomify marketing palette so they # read naturally against the wizard's dark background. See styles.tcss # for the source-of-truth token names. _COLOR_BY_KIND = { @@ -37,12 +40,21 @@ class ApplyScreen(WizardScreen): def __init__(self) -> None: super().__init__() # Toggled by on_worker_state_changed so action_back_if_done / - # the Continue button can switch behaviour once the worker has + # the Continue button can switch behavior once the worker has # finished. Don't allow Escape mid-apply — bailing while we're # part-way through API mutations leaves the workspace in a # weird state. self._worker_done = False self._worker_error = False + # Set when the apply failed on a plan limit — drives the tailored + # recovery CTA (reuse an existing product / component instead of + # "retry", which would just re-fail). + self._plan_limit: PlanLimitError | None = None + # When the plan limit was on products and the workspace has exactly + # one existing product, this holds it so the primary button can flip + # the plan to "use existing" and retry in place — one keypress + # instead of walking Back through four screens. + self._reuse_product: dict[str, Any] | None = None # Captured from the DOM on the main thread in ``on_mount`` so the # worker thread never queries widgets directly. Always assigned # before the worker is started. @@ -65,8 +77,9 @@ def compose_body(self) -> ComposeResult: # Back is disabled during apply (you can't bail mid-API- # mutation) and enabled by on_worker_state_changed when # the worker finishes. Continue is the primary path after - # success; it stays disabled on error and Back becomes - # the only viable option. + # success; on a plan-limit failure it's repurposed as the + # "use existing product & retry" action, and on any other + # error it stays disabled with Back the only viable option. yield Button("◂ Back", id="back", disabled=True) yield Button("Continue ▸", id="continue", variant="primary", disabled=True) @@ -78,14 +91,40 @@ def on_mount(self) -> None: self.run_worker(self._apply_worker, name="apply", thread=True, exclusive=True) def action_back_if_done(self) -> None: - """Escape pops back to the previous screen, but only once the - apply worker has finished — bailing mid-apply could leave the - sbomify workspace half-mutated.""" - if self._worker_done: + """Escape goes back, but only once the apply worker has finished — + bailing mid-apply could leave the sbomify workspace half-mutated.""" + self._go_back() + + def _go_back(self) -> None: + """Navigate back after the worker has finished. + + On a plan-limit failure this jumps straight to the screen where the + user can fix the plan (Pick a product / Components) instead of + stranding them on Review, where pressing Apply again would fail + identically. Any other failure (or success) pops one screen as + before. + """ + if not self._worker_done: + return + target: type[WizardScreen] | None = None + if self._plan_limit is not None: + if self._plan_limit.resource == "product": + from sbomify_action.cli.wizard.screens.product import ProductScreen + + target = ProductScreen + elif self._plan_limit.resource == "component": + from sbomify_action.cli.wizard.screens.components import ComponentsScreen + + target = ComponentsScreen + stack = self.app.screen_stack + if target is not None and any(isinstance(s, target) for s in stack): + while not isinstance(self.app.screen_stack[-1], target): + self.app.pop_screen() + else: self.app.pop_screen() - def _apply_worker(self) -> str | None: - """Run apply_plan; return None on success, error message on failure. + def _apply_worker(self) -> Exception | None: + """Run apply_plan; return None on success, the exception on failure. Runs on a Textual worker thread (``thread=True``). Textual widgets are not thread-safe, so every DOM mutation hops back to the main @@ -97,19 +136,21 @@ def _apply_worker(self) -> str | None: app = self.app def log(kind: str, message: str) -> None: - # ``RichLog`` is mounted with ``markup=True`` so the per-kind colour + # ``RichLog`` is mounted with ``markup=True`` so the per-kind color # tags work. ``message`` is untrusted (API errors, URLs containing # `[`, exception text) — escape it so a stray `[` doesn't get # parsed as markup and either misrender the line or raise mid-log. - colour = _COLOR_BY_KIND.get(kind, "white") - line = f"[{colour}]{kind:>8}[/] {rich_escape(message)}" + # HTTP status markers are stripped: the human-readable detail is + # what the user acts on; the code is developer noise. + color = _COLOR_BY_KIND.get(kind, "white") + line = f"[{color}]{kind:>8}[/] {rich_escape(strip_status_codes(message))}" app.call_from_thread(log_widget.write, line) try: apply_mod.apply_plan(self.wizard.state, self.wizard.opts, log=log) except Exception as exc: # noqa: BLE001 log("error", str(exc)) - return str(exc) + return exc return None def on_worker_state_changed(self, event: Worker.StateChanged) -> None: @@ -136,27 +177,16 @@ def on_worker_state_changed(self, event: Worker.StateChanged) -> None: continue_btn.disabled = False continue_btn.focus() else: - # _apply_worker caught an exception and returned its - # message. The user can't usefully continue to Done - # from a half-applied state — surface the error in - # the pinned banner so it doesn't scroll past, and - # leave only the Back button enabled. - self._worker_error = True - self._show_error_banner(result) - continue_btn.label = "(apply failed)" - continue_btn.disabled = True - back_btn.variant = "primary" - back_btn.focus() + self._on_apply_failed(result, back_btn, continue_btn) elif event.state == WorkerState.ERROR: self._worker_done = True self._worker_error = True back_btn = self.query_one("#back", Button) continue_btn = self.query_one("#continue", Button) - error_text = str(event.worker.error) + error_text = strip_status_codes(str(event.worker.error)) self._show_error_banner(error_text) # ``RichLog`` is markup=True; escape the worker-error message so a - # `[` in the exception text (eg "APIError [404] - ...") doesn't - # collide with the colour wrapping tags. + # `[` in the exception text doesn't collide with the color tags. self.query_one("#apply-log", RichLog).write(f"[#F87171]worker error: {rich_escape(error_text)}[/]") continue_btn.label = "(apply failed)" continue_btn.disabled = True @@ -164,28 +194,133 @@ def on_worker_state_changed(self, event: Worker.StateChanged) -> None: back_btn.disabled = False back_btn.focus() + def _on_apply_failed(self, error: Exception, back_btn: Button, continue_btn: Button) -> None: + """Render the failure state: tailored recovery for plan limits, + generic Back-and-retry for everything else.""" + self._worker_error = True + if isinstance(error, PlanLimitError): + self._plan_limit = error + self._show_plan_limit_banner(error) + else: + self._show_error_banner(strip_status_codes(str(error))) + if self._reuse_product is not None: + name = str(self._reuse_product.get("name") or "existing product") + continue_btn.label = f"Use '{name}' & retry ▸" + continue_btn.disabled = False + back_btn.disabled = False + continue_btn.focus() + else: + continue_btn.label = "(apply failed)" + continue_btn.disabled = True + back_btn.variant = "primary" + back_btn.disabled = False + back_btn.focus() + + def _show_plan_limit_banner(self, error: PlanLimitError) -> None: + """Plan-limit failures get a real recovery path, not "retry". + + Retrying the same plan re-fails identically, so the CTA depends on + what the workspace offers: reuse the (single) existing product in + place, pick one on the product step, or free up quota / upgrade. + """ + message = strip_status_codes(str(error)) + workspace = self.wizard.state.workspace + products = workspace.products if workspace else [] + + if error.resource == "product" and len(products) == 1: + self._reuse_product = products[0] + name = rich_escape(str(products[0].get("name") or "(unnamed)")) + cta = ( + f"Press [b]Use '{name}' & retry[/] to attach everything to your existing " + f"product instead — or delete an unused product / upgrade your plan in the " + "sbomify dashboard, then press [b]◂ Back[/] and Apply again." + ) + elif error.resource == "product" and products: + cta = ( + "Press [b]◂ Back[/] (or [b]Esc[/]) to return to the [b]Pick a product[/] step " + "and select one of your existing products instead — or delete an unused " + "product / upgrade your plan in the sbomify dashboard, then retry." + ) + elif error.resource == "product": + cta = ( + "Delete an unused product (or upgrade your plan) in the sbomify dashboard, " + "then press [b]◂ Back[/] and Apply again." + ) + elif error.resource == "component": + cta = ( + "Press [b]◂ Back[/] (or [b]Esc[/]) to return to the [b]Components[/] step and " + "reuse existing components where possible — or delete unused components / " + "upgrade your plan in the sbomify dashboard, then retry." + ) + else: + cta = "Free up quota (or upgrade your plan) in the sbomify dashboard, then press [b]◂ Back[/] and retry." + + self._update_banner( + f"[#F4B57F]✗ Plan limit reached.[/] [#CBCCCE]{rich_escape(message)}[/]\n[#5E5E5E]{cta}[/]" + ) + def _show_error_banner(self, message: str) -> None: """Surface the error in a pinned banner above the log so it survives the log scrolling past.""" + # ``message`` is API/exception text which can contain `[` — escape so + # a stray bracket can't mis-style the banner or raise from markup + # parsing. + self._update_banner( + f"[#F87171]✗ Apply failed.[/] [#CBCCCE]{rich_escape(message)}[/]\n" + "[#5E5E5E]Press [b]◂ Back[/] (or [b]Esc[/]) to return to Review and retry.[/]" + ) + + def _update_banner(self, markup: str) -> None: try: banner = self.query_one("#apply-error-banner", Static) except Exception: # noqa: BLE001 return - # ``banner`` is a Static with markup=True. The message is API/exception - # text which can contain `[` — escape so a stray bracket can't either - # mis-style the rest of the banner or raise from markup parsing. - banner.update( - f"[#F87171]✗ Apply failed.[/] [#CBCCCE]{rich_escape(message)}[/]\n" - "[#5E5E5E]Press [b]◂ Back[/] (or [b]Esc[/]) to return to Review and retry.[/]" - ) + banner.update(markup) banner.display = True + def _retry_with_existing_product(self) -> None: + """Flip the plan from create-product to use-existing and re-run apply. + + ``apply_plan`` resets its own output state and component creation is + get-or-create, so re-running on the same screen is safe. + """ + product = self._reuse_product + assert product is not None + plan = self.wizard.state.plan + plan.create_product = None + plan.use_product_id = str(product.get("id") or "") + + self._worker_done = False + self._worker_error = False + self._plan_limit = None + self._reuse_product = None + try: + banner = self.query_one("#apply-error-banner", Static) + banner.display = False + panel = self.query_one("#apply-panel", Vertical) + panel.border_title = "⏳ Applying" + panel.border_subtitle = "live log" + except Exception: # noqa: BLE001 + pass + back_btn = self.query_one("#back", Button) + continue_btn = self.query_one("#continue", Button) + back_btn.disabled = True + back_btn.variant = "default" + continue_btn.label = "Continue ▸" + continue_btn.disabled = True + log = self.query_one("#apply-log", RichLog) + log.write("") + log.write(f"[#CBCCCE]Retrying with existing product [b]{rich_escape(str(product.get('name') or ''))}[/] …[/]") + self.run_worker(self._apply_worker, name="apply", thread=True, exclusive=True) + def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "back": - if self._worker_done: - self.app.pop_screen() + self._go_back() return if event.button.id == "continue": + if self._reuse_product is not None: + self._retry_with_existing_product() + return if self._worker_error: # Shouldn't happen — Continue is disabled on error — # but defend in depth. diff --git a/sbomify_action/cli/wizard/screens/authenticate.py b/sbomify_action/cli/wizard/screens/authenticate.py index 15c027cd..f305f373 100644 --- a/sbomify_action/cli/wizard/screens/authenticate.py +++ b/sbomify_action/cli/wizard/screens/authenticate.py @@ -10,7 +10,7 @@ from textual.widgets import Button, Input, LoadingIndicator, Static from textual.worker import Worker, WorkerState -from sbomify_action.cli.wizard.screens._base import WizardScreen +from sbomify_action.cli.wizard.screens._base import WizardScreen, strip_status_codes from sbomify_action.cli.wizard.state import WorkspaceSnapshot from sbomify_action.exceptions import APIError, AuthError from sbomify_action.logging_config import logger @@ -62,6 +62,69 @@ def _pick_default_workspace_key(workspaces: list[dict[str, object]]) -> str | No return fallback +def _resolve_profile_workspace( + base_url: str, + token: str, + workspaces: list[dict[str, object]], + picked_key: str | None, +) -> tuple[str | None, list[dict[str, object]]]: + """Return ``(team_key, contact_profiles)`` for the workspace the token can read. + + ``GET /api/v1/workspaces/`` is NOT filtered by token scope, so + ``_pick_default_workspace_key`` can pick a workspace a scoped token has + no access to. When that happens the contact-profile listing 403s — and + silently treating that as "no profiles" is worse than it looks: the + wizard then shows an empty picker for a workspace that has profiles, + and a profile created from the wizard lands in the WRONG workspace + (the products/components endpoints resolve via the token's bound team, + not the picked key). Verified against production: a token scoped to + workspace A lists A's profiles fine and gets 403 for every other key. + + So: try the picked key first; on failure, probe the remaining + workspaces in parallel and bind to the first one whose listing + succeeds (for a scoped token exactly one can). Returns + ``(None, [])`` when no workspace is readable. + """ + candidates = [str(ws.get("key")) for ws in workspaces if isinstance(ws.get("key"), str) and ws.get("key")] + if picked_key is not None and picked_key in candidates: + candidates.remove(picked_key) + candidates.insert(0, picked_key) + if not candidates: + return None, [] + + def _probe(key: str) -> tuple[str, list[dict[str, object]] | None]: + try: + return key, SbomifyApiClient(base_url, token).list_contact_profiles(key) + except APIError as e: + logger.debug("Contact-profile probe failed for workspace %s: %s", key, e) + return key, None + + # Try the picked workspace alone first — the common case (unscoped + # token, or a scoped token whose workspace IS the picked one) needs + # exactly one request. + first_key, first_profiles = _probe(candidates[0]) + if first_profiles is not None: + return first_key, first_profiles + + rest = candidates[1:] + if not rest: + logger.warning("Could not list contact profiles for workspace %s", first_key) + return None, [] + with ThreadPoolExecutor(max_workers=min(8, len(rest))) as pool: + results = list(pool.map(_probe, rest)) + for key, profiles in results: + if profiles is not None: + logger.info( + "Token cannot read workspace %s; binding contact profiles to workspace %s instead " + "(the token appears to be scoped to it).", + first_key, + key, + ) + return key, profiles + logger.warning("Could not list contact profiles for any workspace the user belongs to") + return None, [] + + def _workspace_display_name(workspaces: list[dict[str, object]], key: str | None) -> str | None: """Return the human name (or key) of the workspace with ``key``. @@ -201,23 +264,6 @@ def _auth_worker(self, token: str) -> tuple[SbomifyApiClient | None, WorkspaceSn logger.warning("Could not list workspaces: %s", e) workspaces = [] team_key = _pick_default_workspace_key(workspaces) - if len(workspaces) > 1: - picked_name = next( - ( - str(ws.get("name") or ws.get("key")) - for ws in workspaces - if isinstance(ws.get("key"), str) and ws.get("key") == team_key - ), - "(unknown)", - ) - logger.warning( - "You belong to %d workspaces; the wizard will use %r (key=%s) — your " - "default workspace. If you intended to onboard a different " - "workspace, change the default in the sbomify UI and re-run.", - len(workspaces), - picked_name, - team_key, - ) def _list_products() -> list[dict[str, object]]: return SbomifyApiClient(base_url, token).list_products() @@ -225,31 +271,35 @@ def _list_products() -> list[dict[str, object]]: def _list_components() -> list[dict[str, object]]: return SbomifyApiClient(base_url, token).list_components() - def _list_profiles() -> list[dict[str, object]]: - if team_key is None: - return [] - try: - return SbomifyApiClient(base_url, token).list_contact_profiles(team_key) - except APIError as e: - # Non-fatal — Augmentation just appears empty in that case. - logger.warning("Could not list contact profiles: %s", e) - return [] + def _list_profiles() -> tuple[str | None, list[dict[str, object]]]: + return _resolve_profile_workspace(base_url, token, workspaces, team_key) try: with ThreadPoolExecutor(max_workers=3) as pool: # Submit all three first so they run concurrently — chaining # ``submit(...).result()`` would block on each future before - # the next is submitted, serialising the calls and defeating + # the next is submitted, serializing the calls and defeating # the entire reason for the pool. products_future = pool.submit(_list_products) components_future = pool.submit(_list_components) profiles_future = pool.submit(_list_profiles) products = products_future.result() components = components_future.result() - profiles = profiles_future.result() + team_key, profiles = profiles_future.result() except APIError as e: return None, None, f"Workspace fetch failed: {e}" + if len(workspaces) > 1: + bound_name = _workspace_display_name(workspaces, team_key) or "(unknown)" + logger.warning( + "You belong to %d workspaces; the wizard is bound to %r (key=%s). If you " + "intended to onboard a different workspace, use a token scoped to that " + "workspace (or change your default workspace in the sbomify UI) and re-run.", + len(workspaces), + bound_name, + team_key, + ) + workspace = WorkspaceSnapshot( products=products, components=components, @@ -306,11 +356,15 @@ def _on_auth_success(self, client: SbomifyApiClient, workspace: WorkspaceSnapsho self.wizard.push_screen(ProductScreen()) def _on_auth_error(self, message: str) -> None: + from rich.markup import escape as rich_escape + self.query_one("#auth-progress", Container).remove_children() self.query_one("#submit", Button).disabled = False self.query_one("#token", Input).disabled = False self.query_one("#token", Input).focus() - self._set_status(f"[#F87171]{message}[/]") + # API error text: drop the developer-facing status code and escape + # stray ``[`` so it can't be parsed as markup. + self._set_status(f"[#F87171]{rich_escape(strip_status_codes(message))}[/]") def _set_status(self, markup: str) -> None: self.query_one("#auth-status", Static).update(markup) diff --git a/sbomify_action/cli/wizard/screens/configure_sbom.py b/sbomify_action/cli/wizard/screens/configure_sbom.py index 328b2a60..7cdf6f24 100644 --- a/sbomify_action/cli/wizard/screens/configure_sbom.py +++ b/sbomify_action/cli/wizard/screens/configure_sbom.py @@ -7,7 +7,7 @@ Enrichment and augmentation are grouped together because they're both metadata-source choices — enrichment pulls package data from external -registries (PyPI, deps.dev, Repology), augmentation pulls organisational +registries (PyPI, deps.dev, Repology), augmentation pulls organizational metadata (supplier / contacts) from a sbomify contact profile. """ @@ -93,10 +93,10 @@ def compose_body(self) -> ComposeResult: ] aug = Vertical(classes="wizard-panel") aug.border_title = "◆ Augmentation" - aug.border_subtitle = "organisational metadata (supplier, contacts, authors)" + aug.border_subtitle = "organizational metadata (supplier, contacts, authors)" with aug: yield Static( - "[#5E5E5E]Lockfiles never carry organisational metadata — who the supplier is, " + "[#5E5E5E]Lockfiles never carry organizational metadata — who the supplier is, " "who authored the SBOM, how to contact security. Both [b]Supplier Name[/] and " "[b]Author of SBOM Data[/] are minimum elements under [b]NTIA[/], [b]CISA[/], " "and the [b]EU Cyber Resilience Act[/], so SBOMs without them fail compliance " @@ -250,9 +250,9 @@ def on_screen_resume(self) -> None: gained a new profile — rebuild the picker so it appears, and auto-select it via the id CreateProfileScreen stashed on the plan. If ConfigureSbomifyJsonScreen was pushed and the user - cancelled (no data on the plan), flip the augmentation back + canceled (no data on the plan), flip the augmentation back to Skip so the user isn't trapped in a re-push loop. A - cancelled CreateProfileScreen gets the same treatment: move + canceled CreateProfileScreen gets the same treatment: move the highlight off the "+ Create new" sentinel (or revert to Skip when the workspace has zero profiles to fall back on). """ @@ -272,7 +272,7 @@ def on_screen_resume(self) -> None: return # Snapshot the previous selection so we can preserve it across # the clear+add rebuild when there's no fresh auto-select - # target. Without this, a cancelled CreateProfileScreen leaves + # target. Without this, a canceled CreateProfileScreen leaves # picker.highlighted=None and the next Enter silently downgrades # augmentation to "skip". previous_highlighted_id: str | None = None @@ -304,7 +304,7 @@ def on_screen_resume(self) -> None: picker.highlighted = idx break - # Handle a cancelled CreateProfileScreen (Escape without saving). + # Handle a canceled CreateProfileScreen (Escape without saving). # With "profile" as the default strategy, a zero-profile # workspace reaches Enter→CreateProfile→Escape→Enter straight # from the screen's defaults — break the loop the same way the @@ -325,7 +325,7 @@ def on_screen_resume(self) -> None: else: self._set_radio_value(aug, target_id="aug-skip") self.notify( - "Cancelled — augmentation reverted to Skip. Pick the radio again to retry.", + "Canceled — augmentation reverted to Skip. Pick the radio again to retry.", title="Contact profile", severity="information", ) @@ -343,7 +343,7 @@ def on_screen_resume(self) -> None: self._json_form_visited = False self._set_radio_value(aug, target_id="aug-skip") self.notify( - "Cancelled — augmentation reverted to Skip. Pick the radio again to retry.", + "Canceled — augmentation reverted to Skip. Pick the radio again to retry.", title="sbomify.json", severity="information", ) diff --git a/sbomify_action/cli/wizard/screens/configure_sbomify_json.py b/sbomify_action/cli/wizard/screens/configure_sbomify_json.py index d0d77022..a3ef036a 100644 --- a/sbomify_action/cli/wizard/screens/configure_sbomify_json.py +++ b/sbomify_action/cli/wizard/screens/configure_sbomify_json.py @@ -56,7 +56,7 @@ class ConfigureSbomifyJsonScreen(PagedFormScreen): step_index = 7 step_title = "Configure (sbomify.json)" - step_subtitle = "Organisational metadata written to the repo, version-controlled alongside the code." + step_subtitle = "Organizational metadata written to the repo, version-controlled alongside the code." PAGE_TITLES = ["Supplier", "Manufacturer · author · security", "Lifecycle"] diff --git a/sbomify_action/cli/wizard/screens/create_profile.py b/sbomify_action/cli/wizard/screens/create_profile.py index 3c428141..c889873f 100644 --- a/sbomify_action/cli/wizard/screens/create_profile.py +++ b/sbomify_action/cli/wizard/screens/create_profile.py @@ -6,7 +6,7 @@ sbomify's ``ContactProfileCreateSchema`` requires: - Profile name (free-text label, internal) - - Organisation entity (supplier + manufacturer) with name + email + + - Organization entity (supplier + manufacturer) with name + email + optional phone / address / website - At least one security contact on the entity — both the backend schema and CRA compliance require it @@ -20,11 +20,13 @@ from __future__ import annotations +from rich.markup import escape as rich_escape from textual.app import ComposeResult from textual.containers import Vertical from textual.widgets import Input, Static from textual.worker import Worker, WorkerState +from sbomify_action.cli.wizard.screens._base import strip_status_codes from sbomify_action.cli.wizard.screens._paged import PagedFormScreen from sbomify_action.exceptions import APIError from sbomify_action.logging_config import logger @@ -39,7 +41,7 @@ class CreateProfileScreen(PagedFormScreen): The field set doesn't fit an 80×24 terminal, so it's split into two fit-to-viewport pages (see ``PagedFormScreen``): identity + - organisation, then security contact + author. + organization, then security contact + author. """ step_index = 7 @@ -49,7 +51,7 @@ class CreateProfileScreen(PagedFormScreen): "and the EU CRA list as minimum SBOM elements." ) - PAGE_TITLES = ["Identity & organisation", "Security & author"] + PAGE_TITLES = ["Identity & organization", "Security & author"] def compose_page(self, index: int) -> ComposeResult: if index == 0: @@ -81,17 +83,17 @@ def _compose_org_page(self) -> ComposeResult: id="profile-name", ) - # Organisation entity — fills the Supplier Name minimum element. + # Organization entity — fills the Supplier Name minimum element. org = Vertical(classes="wizard-panel") - org.border_title = "◇ Organisation" + org.border_title = "◇ Organization" org.border_subtitle = "supplier + manufacturer (CycloneDX entity)" with org: yield Static( "[#F4B57F]Name and email required[/] — feed the SBOM's [b]supplier[/] and [b]manufacturer[/] fields.", classes="wizard-help", ) - yield Input(placeholder="Organisation name (e.g. Acme Inc.)", id="org-name") - yield Input(placeholder="Organisation email (e.g. hello@acme.com)", id="org-email") + yield Input(placeholder="Organization name (e.g. Acme Inc.)", id="org-name") + yield Input(placeholder="Organization email (e.g. hello@acme.com)", id="org-email") yield Input(placeholder="Phone (optional)", id="org-phone") yield Input(placeholder="Address (optional)", id="org-address") yield Input( @@ -145,16 +147,16 @@ def save(self) -> None: sec_name = self.query_one("#sec-name", Input).value.strip() sec_email = self.query_one("#sec-email", Input).value.strip() - # Profile name + organisation live on page 1, the security contact + # Profile name + organization live on page 1, the security contact # on page 2. Jump to whichever page owns the first missing field so # the error lands next to the empty input. missing: list[str] = [] if not name: missing.append("profile name") if not org_name: - missing.append("organisation name") + missing.append("organization name") if not org_email: - missing.append("organisation email") + missing.append("organization email") if not sec_name: missing.append("security contact name") if not sec_email: @@ -234,7 +236,9 @@ def on_worker_state_changed(self, event: Worker.StateChanged) -> None: # error message, which is both confusing and re-prompts # the user to submit a duplicate. if isinstance(result, str): - self._set_status(f"[#F87171]✗ {result}[/]") + # API/exception text: drop the developer-facing status code + # and escape stray ``[`` so it can't be parsed as markup. + self._set_status(f"[#F87171]✗ {rich_escape(strip_status_codes(result))}[/]") self.next_button.disabled = False return if isinstance(result, dict) and result.get("id"): diff --git a/sbomify_action/cli/wizard/screens/done.py b/sbomify_action/cli/wizard/screens/done.py index e3f22923..ddbdaae5 100644 --- a/sbomify_action/cli/wizard/screens/done.py +++ b/sbomify_action/cli/wizard/screens/done.py @@ -56,7 +56,7 @@ def compose_body(self) -> ComposeResult: # Failures are tracked per-component now, so a partial success # falls through to the manual-fallback branch and lists ONLY # the failed components rather than blanket-listing all of - # them (the prior behaviour, which made users re-bind already- + # them (the prior behavior, which made users re-bind already- # bound components and hit 409 errors). oidc = Vertical(classes="wizard-panel-emphasis") oidc.border_title = "✓ OIDC trusted publishing is set up" @@ -147,7 +147,7 @@ def _applied_summary(self) -> str: lines.append(f" [#5E5E5E]reason: {state.attach_error}[/]") # Dry-run "would write" rows use a muted glyph + label so the user # can tell from the summary that nothing actually hit disk. The - # real-apply branch keeps the existing green ✓ + "Wrote" labelling. + # real-apply branch keeps the existing green ✓ + "Wrote" labeling. if state.is_dry_run: for path in state.written_files: lines.append(f"[#5E5E5E]◌ would write[/] {path}") diff --git a/sbomify_action/cli/wizard/screens/review.py b/sbomify_action/cli/wizard/screens/review.py index ab112dc7..69f78136 100644 --- a/sbomify_action/cli/wizard/screens/review.py +++ b/sbomify_action/cli/wizard/screens/review.py @@ -233,7 +233,7 @@ def _read_existing(path: Path) -> str: @staticmethod def _stylise(line: str) -> str: - """Colour a single unified-diff line, escaping Rich markup.""" + """Color a single unified-diff line, escaping Rich markup.""" # Escape stray '[' that might be Rich markup in the file content. escaped = line.replace("[", r"\[") if line.startswith("+++") or line.startswith("---"): diff --git a/sbomify_action/cli/wizard/screens/welcome.py b/sbomify_action/cli/wizard/screens/welcome.py index 1f25706f..b4704cde 100644 --- a/sbomify_action/cli/wizard/screens/welcome.py +++ b/sbomify_action/cli/wizard/screens/welcome.py @@ -22,7 +22,7 @@ # from any one artist — a real wizard hat is at least as tall as the # face + beard, which is what the previous draft was missing. # -# Rows are coloured to echo the sbomify gradient: peach hat tip, +# Rows are colored to echo the sbomify gradient: peach hat tip, # magenta hat base with stars, silvery beard (Gandalf cue), blue # robe deepening to brand-primary at the hem. Pure ASCII (no # box-drawing or exotic Unicode) so the figure renders uniformly diff --git a/sbomify_action/cli/wizard/state.py b/sbomify_action/cli/wizard/state.py index 5665958d..b7728983 100644 --- a/sbomify_action/cli/wizard/state.py +++ b/sbomify_action/cli/wizard/state.py @@ -34,7 +34,7 @@ fields the user fills in on the Configure (sbomify.json) screen. The action's ``json_config`` provider reads it at workflow run time. - ``skip`` — set ``AUGMENT: 'false'``; the user will manage metadata - out-of-band (or accept blank organisational fields). + out-of-band (or accept blank organizational fields). """ ReleaseStrategy = Literal["trunk", "tag", "manual"] diff --git a/sbomify_action/cli/wizard/styles.tcss b/sbomify_action/cli/wizard/styles.tcss index ae440eb7..039aab59 100644 --- a/sbomify_action/cli/wizard/styles.tcss +++ b/sbomify_action/cli/wizard/styles.tcss @@ -23,7 +23,7 @@ $sbom-grad-peach: #F4B57F; /* NOTE: we intentionally do NOT redefine Textual's built-in design * tokens ($accent, $primary, $panel, …). Those are referenced by * Textual's own internal stylesheet (e.g. ``hatch: right $panel;``) - * with shape expectations that don't accept arbitrary hex colours. + * with shape expectations that don't accept arbitrary hex colors. * Instead, every selector below uses our $sbom-* variables directly. */ /* --------------------------------------------------------------------- diff --git a/sbomify_action/cli/wizard/widgets/pick_or_create.py b/sbomify_action/cli/wizard/widgets/pick_or_create.py index 849b4c47..340433e9 100644 --- a/sbomify_action/cli/wizard/widgets/pick_or_create.py +++ b/sbomify_action/cli/wizard/widgets/pick_or_create.py @@ -30,7 +30,7 @@ class PickOrCreate(Vertical): - """OptionList + Input with shared sentinel + auto-hide behaviour.""" + """OptionList + Input with shared sentinel + auto-hide behavior.""" DEFAULT_CSS = """ PickOrCreate { diff --git a/sbomify_action/cli/wizard/widgets/stateful_radio.py b/sbomify_action/cli/wizard/widgets/stateful_radio.py index 3b0713dd..5ec84aea 100644 --- a/sbomify_action/cli/wizard/widgets/stateful_radio.py +++ b/sbomify_action/cli/wizard/widgets/stateful_radio.py @@ -3,17 +3,17 @@ Textual's stock ``RadioButton`` always renders ``▐●▌`` regardless of selection state — the on/off distinction is purely a CSS style (white+bold when selected, muted otherwise). That works in a richly -coloured TTY but breaks in two practical cases: +colored TTY but breaks in two practical cases: - - Users with limited contrast / colour vision can't tell which radio + - Users with limited contrast / color vision can't tell which radio is active. - Copy/pasting a wizard screen into chat or a bug report strips - colour formatting, leaving three identical ``▐●▌`` rows that + color formatting, leaving three identical ``▐●▌`` rows that look like a multi-select bug rather than a single-choice radio. ``StatefulRadioButton`` overrides the inner glyph to ``●`` only when selected; unselected radios render with ``○``. The visual difference -is then encoded in the character itself, not just the colour. +is then encoded in the character itself, not just the color. """ from __future__ import annotations @@ -36,7 +36,7 @@ def _button(self) -> Content: Mirrors the parent's ``_button`` shape (left + inner + right glyphs styled per ``toggle--button`` / its side variant) but swaps the inner character based on ``self.value`` so the - selection state is visually unambiguous even with colour + selection state is visually unambiguous even with color information stripped. """ button_style = self.get_visual_style("toggle--button") diff --git a/sbomify_action/enrichment.py b/sbomify_action/enrichment.py index 6521da32..7d10319e 100644 --- a/sbomify_action/enrichment.py +++ b/sbomify_action/enrichment.py @@ -382,7 +382,7 @@ def _extract_packages_from_spdx(document: Document) -> List[Tuple[Package, str]] return packages -# Filename suffixes that BSI TR-03183-2 §5.2.2 recognises as archives. +# Filename suffixes that BSI TR-03183-2 §5.2.2 recognizes as archives. # Covers common language-agnostic wheel / tarball / zip / container formats. _BSI_ARCHIVE_SUFFIXES = ( ".whl", @@ -450,7 +450,7 @@ def _apply_bsi_derived_properties(component: Component, metadata: NormalizedMeta component that reaches derivation has a distribution filename or a strongly-typed packaging semantics, both of which imply - an identifiable, parseable artefact per + an identifiable, parseable artifact per BSI §8.1.6. We don't have a reliable signal to emit "unstructured", so we leave that classification to operator @@ -469,9 +469,9 @@ def _apply_bsi_derived_properties(component: Component, metadata: NormalizedMeta Returns the list of property names that were added (for audit trail). """ added: List[str] = [] - # `Component.properties` can be None on deserialised or user-constructed + # `Component.properties` can be None on deserialized or user-constructed # components (see _enrich_os_component which already handles this). - # Initialise with a plain `set()` so the subsequent `.add()` calls work + # Initialize with a plain `set()` so the subsequent `.add()` calls work # without depending on `sortedcontainers`. if component.properties is None: component.properties = set() @@ -515,26 +515,26 @@ def _apply_bsi_derived_properties(component: Component, metadata: NormalizedMeta added.append("bsi:component:executable") # --- structured --- - # Packaged software artefacts (wheels, jars, debs, containers, firmware + # Packaged software artifacts (wheels, jars, debs, containers, firmware # images) all carry metadata files, so they qualify as "structured" per # BSI §8.1.6. Base the signal on the component's inherent shape, not on # whether this invocation happened to add archive/executable: - # 1. A recognised filename suffix (archive OR executable) → structured. + # 1. A recognized filename suffix (archive OR executable) → structured. # 2. An existing bsi:component:archive / bsi:component:executable # property already on the component (operator-supplied) → structured. # 3. A deployable component type (application / container / firmware / # operating-system) → structured. - # A bare `library` with an unrecognised filename suffix and no + # A bare `library` with an unrecognized filename suffix and no # operator-supplied archive/executable hints is ambiguous and gets # nothing — per the docstring contract, we do not guess "structured" # from weak signals. - has_recognised_filename = bool(filename) and ( + has_recognized_filename = bool(filename) and ( _filename_suffix_matches(filename, _BSI_ARCHIVE_SUFFIXES) or _filename_suffix_matches(filename, _BSI_EXECUTABLE_SUFFIXES) ) operator_has_archive_exec = "bsi:component:archive" in existing or "bsi:component:executable" in existing has_structured_signal = bool( - has_recognised_filename + has_recognized_filename or operator_has_archive_exec or archive_value is not None or exec_value is not None @@ -549,7 +549,7 @@ def _apply_bsi_derived_properties(component: Component, metadata: NormalizedMeta # Map SPDX-style algorithm names (PyPI digest keys) to CycloneDX HashAlgorithm. # The CycloneDX taxonomy uses upper-case SHA-256 etc.; the PyPI JSON API and -# SPDX checksum fields use lower-case sha256 etc. We normalise to the enum. +# SPDX checksum fields use lower-case sha256 etc. We normalize to the enum. _CYCLONEDX_HASH_ALGORITHMS: Dict[str, HashAlgorithm] = { "md5": HashAlgorithm.MD5, "sha1": HashAlgorithm.SHA_1, @@ -607,7 +607,7 @@ def _apply_bsi_derived_properties(component: Component, metadata: NormalizedMeta def _apply_spdx_checksums(package: Package, hashes: Dict[str, str]) -> List[str]: - """Add distribution-artefact checksums to an SPDX package. + """Add distribution-artifact checksums to an SPDX package. Shares the same hex-length validation as the CycloneDX path so that any non-hex or malformed payload is rejected rather than written out @@ -682,14 +682,14 @@ def _is_valid_hex_hash(alg_key: str, value: str) -> bool: def _apply_component_hashes(component: Component, hashes: Dict[str, str]) -> List[str]: - """Add distribution-artefact hashes to a CycloneDX component from a - `{algorithm: hex}` map. Only recognised algorithms with valid hex + """Add distribution-artifact hashes to a CycloneDX component from a + `{algorithm: hex}` map. Only recognized algorithms with valid hex content of the expected length are emitted; the same (alg, content) pair is not duplicated across enrichment runs. `Component.hashes` defaults to an empty collection in the CycloneDX - library, but deserialised or user-constructed components may legitimately - have `hashes is None`. Initialise with a plain `set()` (matching + library, but deserialized or user-constructed components may legitimately + have `hashes is None`. Initialize with a plain `set()` (matching `_hash_enrichment/enricher.py`) so enrichment does not crash on such input and does not depend on the transitive `sortedcontainers` package. """ @@ -709,7 +709,7 @@ def _apply_component_hashes(component: Component, hashes: Dict[str, str]) -> Lis continue # `component.hashes` is typed as Iterable[HashType] by the # cyclonedx lib (the concrete default is SortedSet), so cast it - # locally to suppress mypy while preserving runtime behaviour. + # locally to suppress mypy while preserving runtime behavior. component.hashes.add(HashType(alg=cdx_alg, content=value)) # type: ignore[union-attr] added.append(f"hash:{alg_key}") existing.add((str(cdx_alg), value)) @@ -737,8 +737,8 @@ def _apply_metadata_to_cyclonedx_component( purl_str = str(component.purl) if component.purl else component.name # `Component.properties` / `.licenses` / `.external_references` can be - # `None` on deserialised or user-constructed components. The enrichment - # path below iterates/adds to all three, so normalise them up-front + # `None` on deserialized or user-constructed components. The enrichment + # path below iterates/adds to all three, so normalize them up-front # rather than scattering `is None` guards across every branch. Plain # `set()` matches `_hash_enrichment/enricher.py` and avoids relying on # the transitive `sortedcontainers` default. @@ -757,8 +757,8 @@ def _apply_metadata_to_cyclonedx_component( added_fields.append("description") # Licenses (sanitized) — marked as "declared" because enrichment pulls - # the licence straight from the upstream registry (PyPI trove classifier, - # package metadata, etc.), which is BSI §5.2.4's "original licence". + # the license straight from the upstream registry (PyPI trove classifier, + # package metadata, etc.), which is BSI §5.2.4's "original license". has_licenses = component.licenses is not None and len(component.licenses) > 0 if not has_licenses and metadata.licenses: sanitized_licenses: list[str] = [s for lic in metadata.licenses if (s := sanitize_license(lic)) is not None] @@ -799,7 +799,7 @@ def _apply_metadata_to_cyclonedx_component( added_fields.extend(_added_bsi) # Hashes (NTIA / BSI §5.2.2 / CISA "Component Hash"). Only add algorithms - # we recognise and that aren't already on the component. + # we recognize and that aren't already on the component. if metadata.hashes: _added_hashes = _apply_component_hashes(component, metadata.hashes) added_fields.extend(_added_hashes) diff --git a/sbomify_action/exceptions.py b/sbomify_action/exceptions.py index f67d0b16..8862df33 100644 --- a/sbomify_action/exceptions.py +++ b/sbomify_action/exceptions.py @@ -88,7 +88,16 @@ class AuthError(APIError): class PlanLimitError(APIError): - """Raised when an API operation fails due to plan limits (e.g., max components).""" + """Raised when an API operation fails due to plan limits (e.g., max components). + + ``resource`` names what hit the limit (``"product"`` or ``"component"``) + so UI layers (e.g. the wizard's apply screen) can offer a targeted + recovery path — reuse an existing product vs. reuse existing components. + """ + + def __init__(self, message: str, *, resource: str | None = None) -> None: + super().__init__(message) + self.resource = resource class OIDCError(APIError): diff --git a/sbomify_action/sbomify_api.py b/sbomify_action/sbomify_api.py index 6df748de..c1f111e0 100644 --- a/sbomify_action/sbomify_api.py +++ b/sbomify_action/sbomify_api.py @@ -275,7 +275,7 @@ def iter_components(self, error_context: str = "list components") -> Iterator[di yield from self._paginate("/api/v1/components", error_context=error_context) def list_components(self) -> list[dict[str, Any]]: - """Materialise the full list of components.""" + """Materialize the full list of components.""" return list(self.iter_components()) def list_components_by_name(self) -> dict[str, str]: @@ -357,10 +357,30 @@ def create_component( return existing_id, False raise APIError(f"Component '{name}' reported as duplicate by API but could not be found via lookup") - if response.status_code == 403 and isinstance(raw_detail, str) and "maximum" in raw_detail.lower(): - raise PlanLimitError(err_msg) + if self._is_plan_limit(response.status_code, raw_detail, error_code): + # Plan-limit messages are user-actionable and surfaced verbatim in + # UIs (the wizard's apply banner) — keep them human: no status code. + plan_msg = f"Could not create component '{name}': {raw_detail}" if raw_detail else err_msg + raise PlanLimitError(plan_msg, resource="component") raise APIError(err_msg) + @staticmethod + def _is_plan_limit(status_code: int, raw_detail: Any, error_code: str) -> bool: + """True when a create was rejected because the team's plan limit is hit. + + The backend tags these with ``error_code: BILLING_LIMIT_EXCEEDED`` + (verified against ``core/apis._check_billing_limits``); the string + match on the raw detail is kept as a fallback for older deployments + that predate the error code. Only a *string* detail counts — a + pydantic list-detail whose collapsed text contains "maximum" is a + validation error, not a plan limit. + """ + if status_code != 403: + return False + if error_code == "BILLING_LIMIT_EXCEEDED": + return True + return isinstance(raw_detail, str) and "maximum" in raw_detail.lower() + def get_or_create_component( self, name: str, @@ -433,11 +453,48 @@ def get_product(self, product_id: str) -> dict[str, Any]: raise APIError(self._build_error(f"Failed to fetch product {product_id}.", response)) return self._safe_json_dict(response) or {} - def create_product(self, name: str) -> dict[str, Any]: + def get_product_by_name(self, name: str) -> dict[str, Any] | None: + """Find a product by exact name match, or None.""" + for product in self._paginate("/api/v1/products", error_context=f"look up product '{name}'"): + if product.get("name") == name: + return product + return None + + def create_product(self, name: str) -> tuple[dict[str, Any], bool]: + """Create a product with get-or-create semantics. + + Returns ``(product, was_created)``. Mirrors ``create_component``: + recovers from ``DUPLICATE_NAME`` (status 400 or 409) by looking the + existing product up by name — without this, a retry after a + partially-failed apply (product created, later step failed) dies + here instead of reusing the product it created last time. Raises + ``PlanLimitError`` (with ``resource="product"``) when the team has + hit its product-count limit. + """ response = self._request("POST", "/api/v1/products", json_body={"name": name}) - if not response.ok: - raise APIError(self._build_error(f"Failed to create product '{name}'.", response)) - return self._safe_json_dict(response) or {} + if response.ok: + return self._safe_json_dict(response) or {}, True + + body = self._safe_json_dict(response) or {} + raw_detail = body.get("detail") + error_code = body.get("error_code") or "" + + if response.status_code in (400, 409) and error_code == "DUPLICATE_NAME": + logger.info(f"Product '{name}' already exists, retrieving existing product") + existing = self.get_product_by_name(name) + if existing is not None: + return existing, False + raise APIError(f"Product '{name}' reported as duplicate by API but could not be found via lookup") + + if self._is_plan_limit(response.status_code, raw_detail, error_code): + # User-actionable, surfaced verbatim in the wizard — no status code. + plan_msg = ( + f"Could not create product '{name}': {raw_detail}" + if isinstance(raw_detail, str) and raw_detail + else f"Could not create product '{name}': your plan's product limit has been reached." + ) + raise PlanLimitError(plan_msg, resource="product") + raise APIError(self._build_error(f"Failed to create product '{name}'.", response)) def attach_components_to_product(self, product_id: str, component_ids: list[str]) -> None: """Set the full component list on a product. @@ -490,7 +547,7 @@ def list_workspaces(self) -> list[dict[str, Any]]: ``/api/v1/workspaces`` returns 301 (redirect, may drop bodies on non-GET). The nested routes (``/{key}/contact-profiles`` etc.) do NOT accept a trailing slash; they 404 when one is present. - Do not "normalise" the slashes here — each route's shape is + Do not "normalize" the slashes here — each route's shape is dictated by the backend's mount config and verified by integration probing. @@ -541,9 +598,16 @@ def list_contact_profiles(self, team_key: str) -> list[dict[str, Any]]: data = response.json() except ValueError: raise APIError("Failed to list contact profiles: invalid JSON response from API") - if not isinstance(data, list): - return [] - return [item for item in data if isinstance(item, dict)] + if isinstance(data, list): + return [item for item in data if isinstance(item, dict)] + # Accept a paginated envelope too — the shape every other list + # endpoint uses — so a future backend migration of this route + # doesn't silently return [] and hide every existing profile. + if isinstance(data, dict): + items = data.get("items") + if isinstance(items, list): + return [item for item in items if isinstance(item, dict)] + return [] def create_contact_profile(self, team_key: str, payload: dict[str, Any]) -> dict[str, Any]: """Create a contact profile in a workspace. @@ -556,16 +620,38 @@ def create_contact_profile(self, team_key: str, payload: dict[str, Any]) -> dict CISA / EU CRA list as minimum elements. Returns the created profile dict (with the new ``id``) on 201; raises ``APIError`` on validation / permission failure. + + Recovers from ``DUPLICATE_NAME`` (status 400 or 409) by looking + the existing profile up by name and returning it. This matters + for a lost-response resubmit: a first POST that created the + profile but whose result the caller never saw (canceled worker, + dropped connection) would otherwise dead-end every retry on the + duplicate error even though the profile is right there. """ response = self._request( "POST", f"/api/v1/workspaces/{team_key}/contact-profiles", json_body=payload, ) - if not response.ok: - raise APIError(self._build_error("Failed to create contact profile.", response)) - data = self._safe_json_dict(response) - return data or {} + if response.ok: + return self._safe_json_dict(response) or {} + + body = self._safe_json_dict(response) or {} + error_code = body.get("error_code") or "" + name = payload.get("name") + if response.status_code in (400, 409) and error_code == "DUPLICATE_NAME" and isinstance(name, str): + logger.info(f"Contact profile '{name}' already exists, retrieving existing profile") + for profile in self.list_contact_profiles(team_key): + if profile.get("name") == name: + return profile + # The duplicate exists but isn't in the list the token can see — + # fall through to a message that points at the dashboard instead + # of the raw constraint error. + raise APIError( + f"A contact profile named '{name}' already exists in this workspace but could not " + "be retrieved. Check Settings → Contacts in the sbomify dashboard." + ) + raise APIError(self._build_error("Failed to create contact profile.", response)) # ------------------------------------------------------------------ # releases diff --git a/sbomify_action/spdx3.py b/sbomify_action/spdx3.py index fdf7c77d..e15f0cd5 100644 --- a/sbomify_action/spdx3.py +++ b/sbomify_action/spdx3.py @@ -124,7 +124,7 @@ def __init__(self) -> None: # Map HashAlgorithm enum names (upper) to enum values _HASH_ALGORITHMS: dict[str, HashAlgorithm] = {a.name.lower(): a for a in HashAlgorithm} -# The spdx_tools writer serialises enum names via snake_case_to_camel_case +# The spdx_tools writer serializes enum names via snake_case_to_camel_case # (producing camelCase strings). We build lookup dicts that accept both the # camelCase and raw snake_case forms. _s2c is imported at module top. @@ -575,7 +575,7 @@ def parse_spdx3_data(data: dict[str, Any]) -> Spdx3Payload: continue elem_type = elem.get("type") or elem.get("@type", "") - # Normalise aliases + # Normalize aliases elem_type = _TYPE_ALIASES.get(elem_type, elem_type) try: @@ -704,7 +704,7 @@ def write_spdx3_file( ) -> None: """Write a :class:`Payload` to a JSON-LD ``.json`` file. - Uses ``spdx_tools``' converter to serialise model objects, then wraps + Uses ``spdx_tools``' converter to serialize model objects, then wraps them with the official ``@context`` URL. Args: diff --git a/tests/test_enrichment_module.py b/tests/test_enrichment_module.py index 2c03bd41..4bea13da 100644 --- a/tests/test_enrichment_module.py +++ b/tests/test_enrichment_module.py @@ -528,7 +528,7 @@ def test_fetch_populates_hashes_from_digests(self, mock_session): def test_fetch_captures_pypi_blake2b_256_digest(self, mock_session): """PyPI's JSON API emits BLAKE2b-256 under the `blake2b_256` key (underscore, not hyphen). Previously our algorithm mapping only - recognised the hyphenated canonical form, so every BLAKE2b hash + recognized the hyphenated canonical form, so every BLAKE2b hash from a PyPI wheel was silently dropped during enrichment. Pin the mapping so a regression gets caught immediately. """ @@ -556,7 +556,7 @@ def test_fetch_captures_pypi_blake2b_256_digest(self, mock_session): metadata = source.fetch(purl, mock_session) assert metadata is not None - # All three digests must land in the normalised hashes map. + # All three digests must land in the normalized hashes map. assert metadata.hashes == { "blake2b_256": "1" * 64, "md5": "2" * 32, @@ -867,7 +867,7 @@ class _Shim: fetch() method relies on. `PackageURL.from_string` actually preserves `..` in the name field (verified empirically) — so a real PURL would do, too — but the shim keeps the unit test - independent of how the third-party parser may normalise in + independent of how the third-party parser may normalize in the future.""" name = ".." @@ -893,7 +893,7 @@ def test_fetch_rejection_does_not_poison_latest_cache_sentinel(self, mock_sessio class _Shim: # `:` in the version bypasses PackageURL quoting — construct # the attacker-controlled shape directly to make sure the fix - # holds regardless of parser normalisation. + # holds regardless of parser normalization. name = "foo" version = ":latest" @@ -911,7 +911,7 @@ class _Shim: def test_fetch_encodes_special_characters_in_purl_components(self, mock_session): """Safe special characters survive but are percent-encoded in the URL, - so `requests` cannot normalise them into path traversal.""" + so `requests` cannot normalize them into path traversal.""" from sbomify_action._enrichment.sources import pypi as pypi_module pypi_module._cache.clear() @@ -3566,7 +3566,7 @@ def test_bsi_plain_library_no_filename_skipped(self): ], ) def test_bsi_filename_extensions_derive_correctly(self, filename, expected_archive, expected_exec): - """Each recognised archive / executable filename extension maps to + """Each recognized archive / executable filename extension maps to the expected BSI boolean-style property values.""" from sbomify_action._enrichment.metadata import NormalizedMetadata from sbomify_action.enrichment import _apply_metadata_to_cyclonedx_component @@ -3691,9 +3691,9 @@ def test_component_blake2b_wrong_length_rejected(self, alg, bad_value): f"got hashes={list(component.hashes)!r}" ) - def test_component_hashes_initialised_when_hashes_is_none(self): - """Components deserialised from some inputs can legitimately have - `hashes is None`. The hash-enrichment path must initialise the + def test_component_hashes_initialized_when_hashes_is_none(self): + """Components deserialized from some inputs can legitimately have + `hashes is None`. The hash-enrichment path must initialize the collection instead of iterating None (which would raise).""" from sbomify_action._enrichment.metadata import NormalizedMetadata from sbomify_action.enrichment import _apply_metadata_to_cyclonedx_component @@ -3900,7 +3900,7 @@ def test_cyclonedx_sha512_hash_emitted(self): assert str(hashes[0].content) == "f" * 128 def test_cyclonedx_hash_whitespace_trimmed_and_lowercased(self): - """Algorithm keys and hex content are normalised.""" + """Algorithm keys and hex content are normalized.""" from sbomify_action._enrichment.metadata import NormalizedMetadata from sbomify_action.enrichment import _apply_metadata_to_cyclonedx_component @@ -3922,7 +3922,7 @@ def test_cyclonedx_empty_hash_value_rejected(self): _apply_metadata_to_cyclonedx_component(component, metadata) assert len(list(component.hashes)) == 0 - # --- P2 #6: enriched licences marked as BSI "original/declared" -------------- + # --- P2 #6: enriched licenses marked as BSI "original/declared" -------------- def test_enriched_license_marked_declared(self): from sbomify_action._enrichment.metadata import NormalizedMetadata @@ -3931,10 +3931,10 @@ def test_enriched_license_marked_declared(self): component = Component(name="django", version="5.1", type=ComponentType.LIBRARY) metadata = NormalizedMetadata(licenses=["BSD-3-Clause"]) _apply_metadata_to_cyclonedx_component(component, metadata) - licences = list(component.licenses) - assert len(licences) == 1 + licenses = list(component.licenses) + assert len(licenses) == 1 # CycloneDX LicenseExpression exposes the enum via .acknowledgement - ack = getattr(licences[0], "acknowledgement", None) + ack = getattr(licenses[0], "acknowledgement", None) assert ack is not None assert str(ack.value) == "declared" @@ -3943,7 +3943,7 @@ def test_enriched_license_marked_declared(self): [ # CDX 1.3 / 1.4 / 1.5: license.acknowledgement did not exist in # the schema. cyclonedx-python-lib's version-specific outputter - # drops the field on serialisation. If we emit it on a <1.6 + # drops the field on serialization. If we emit it on a <1.6 # BOM we must NOT see it in the output JSON. ("1.3", False), ("1.4", False), @@ -3954,7 +3954,7 @@ def test_enriched_license_marked_declared(self): ], ) def test_acknowledgement_serialization_is_version_gated(self, spec_version, expect_acknowledgement): - """Locks the serialisation-time contract: license.acknowledgement is + """Locks the serialization-time contract: license.acknowledgement is dropped on CDX <1.6 and present on >=1.6. The enrichment helper unconditionally attaches `acknowledgement=declared`; the cyclonedx- python-lib outputter is responsible for the version filter. If that @@ -3984,16 +3984,16 @@ def test_acknowledgement_serialization_is_version_gated(self, spec_version, expe payload = _json.loads(out) emitted = payload.get("components", [{}])[0] - licence_entries = emitted.get("licenses", []) - assert licence_entries, f"no licence emitted for CDX {spec_version}" - first = licence_entries[0] - # LicenseExpression may serialise the field at the top level or nest + license_entries = emitted.get("licenses", []) + assert license_entries, f"no license emitted for CDX {spec_version}" + first = license_entries[0] + # LicenseExpression may serialize the field at the top level or nest # it under "license" depending on the library version; probe both. body = first if "acknowledgement" in first else first.get("license", {}) present = "acknowledgement" in body assert present is expect_acknowledgement, ( f"CDX {spec_version}: expected acknowledgement " - f"{'present' if expect_acknowledgement else 'absent'}, got {licence_entries!r}" + f"{'present' if expect_acknowledgement else 'absent'}, got {license_entries!r}" ) if expect_acknowledgement: assert body.get("acknowledgement") == "declared" diff --git a/tests/test_sbomify_api.py b/tests/test_sbomify_api.py index c2ceb45c..0936141c 100644 --- a/tests/test_sbomify_api.py +++ b/tests/test_sbomify_api.py @@ -381,8 +381,48 @@ def test_list_products_paginates() -> None: def test_create_product() -> None: client, _ = _client_with([_FakeResponse(201, {"id": "p1", "name": "X"})]) - product = client.create_product("X") + product, was_created = client.create_product("X") assert product["id"] == "p1" + assert was_created is True + + +def test_create_product_recovers_from_duplicate_name() -> None: + """A DUPLICATE_NAME rejection resolves to the existing product — the + retry-after-partial-apply path (product created, later step failed) + must reuse the product instead of dead-ending on the duplicate.""" + client, _ = _client_with( + [ + _FakeResponse(400, {"error_code": "DUPLICATE_NAME", "detail": "exists"}), + _FakeResponse( + 200, + {"items": [{"id": "p-existing", "name": "X"}], "pagination": {"has_next": False}}, + ), + ] + ) + product, was_created = client.create_product("X") + assert product["id"] == "p-existing" + assert was_created is False + + +def test_create_product_plan_limit_is_clean_and_typed() -> None: + """The plan-limit 403 raises PlanLimitError tagged with the resource, + and the message carries the human detail without the status code.""" + detail = "You have reached the maximum 1 products allowed by your plan. You currently have 1 products." + client, _ = _client_with([_FakeResponse(403, {"detail": detail, "error_code": "BILLING_LIMIT_EXCEEDED"})]) + with pytest.raises(PlanLimitError) as exc: + client.create_product("Notipus") + assert exc.value.resource == "product" + message = str(exc.value) + assert detail in message + assert "[403]" not in message + + +def test_create_component_plan_limit_is_clean_and_typed() -> None: + client, _ = _client_with([_FakeResponse(403, {"detail": "maximum components reached"})]) + with pytest.raises(PlanLimitError) as exc: + client.create_component("foo", component_type="bom") + assert exc.value.resource == "component" + assert "[403]" not in str(exc.value) def test_attach_components_unions_existing() -> None: @@ -427,6 +467,50 @@ def test_list_contact_profiles_success() -> None: assert profiles[0]["id"] == "cp1" +def test_list_contact_profiles_accepts_paginated_envelope() -> None: + """A future backend migration to the `{items: [...]}` envelope must not + silently hide every existing profile.""" + client, _ = _client_with([_FakeResponse(200, {"items": [{"id": "cp1", "name": "Team"}]})]) + profiles = client.list_contact_profiles("acme-team") + assert profiles[0]["id"] == "cp1" + + +def test_create_contact_profile_recovers_from_duplicate_name() -> None: + """A DUPLICATE_NAME rejection resolves to the existing profile — a + resubmit whose first POST created the profile but lost the response + must succeed instead of dead-ending on the constraint error.""" + client, _ = _client_with( + [ + _FakeResponse( + 400, + { + "detail": "Could not save contact profile due to a database constraint (possibly a duplicate name)", + "error_code": "DUPLICATE_NAME", + }, + ), + _FakeResponse(200, [{"id": "cp-existing", "name": "Default"}]), + ] + ) + profile = client.create_contact_profile("acme-team", {"name": "Default", "entities": []}) + assert profile["id"] == "cp-existing" + + +def test_create_contact_profile_duplicate_not_found_points_at_dashboard() -> None: + """When the duplicate exists but the token can't see it in the list, + the error points the user at the dashboard instead of the raw + constraint text.""" + client, _ = _client_with( + [ + _FakeResponse(400, {"detail": "constraint", "error_code": "DUPLICATE_NAME"}), + _FakeResponse(200, []), + ] + ) + with pytest.raises(APIError) as exc: + client.create_contact_profile("acme-team", {"name": "Default", "entities": []}) + assert "already exists" in str(exc.value) + assert "dashboard" in str(exc.value) + + def test_list_workspaces_returns_workspace_keys() -> None: client, _ = _client_with([_FakeResponse(200, [{"key": "acme", "name": "Acme Inc"}])]) workspaces = client.list_workspaces() diff --git a/tests/test_upload_module.py b/tests/test_upload_module.py index 3ce341b9..9a798622 100644 --- a/tests/test_upload_module.py +++ b/tests/test_upload_module.py @@ -1558,7 +1558,7 @@ def test_upload_timeout_error(self, mock_client_cls): Path(sbom_file).unlink() -def test_upload_input_normalises_bom_type_case(tmp_path): +def test_upload_input_normalizes_bom_type_case(tmp_path): """Programmatic callers passing 'VEX' get the lowercase canonical value, not a ValueError.""" from sbomify_action._upload.protocol import UploadInput diff --git a/tests/test_wizard_emitter.py b/tests/test_wizard_emitter.py index 7b017909..90302795 100644 --- a/tests/test_wizard_emitter.py +++ b/tests/test_wizard_emitter.py @@ -634,7 +634,7 @@ def test_apply_plan_create_new_product(tmp_path: Path) -> None: state = _state(tmp_path) api = state.api assert api is not None - api.create_product.return_value = {"id": "prod-new", "name": "Widget"} + api.create_product.return_value = ({"id": "prod-new", "name": "Widget"}, True) api.get_or_create_component.return_value = ("comp-1", True) state.plan = Plan( diff --git a/tests/test_wizard_state.py b/tests/test_wizard_state.py index 324b8c9a..126cc22f 100644 --- a/tests/test_wizard_state.py +++ b/tests/test_wizard_state.py @@ -55,7 +55,7 @@ def test_require_api_raises_when_unset(tmp_path: Path) -> None: state.require_api() -def test_wizard_state_repr_summarises(tmp_path: Path) -> None: +def test_wizard_state_repr_summarizes(tmp_path: Path) -> None: state = WizardState(facts=_facts(tmp_path)) state.workspace = WorkspaceSnapshot(products=[{"id": "p1"}]) state.selected = [ @@ -275,7 +275,7 @@ def test_write_sbomify_json_sentinel_wins_over_payload_collision(tmp_path: Path) # payload's override attempt. assert data[WIZARD_JSON_SENTINEL_KEY]["managed"] is True assert data[WIZARD_JSON_SENTINEL_KEY]["version"] == 1 - # Sentinel-helpers still recognises the file as wizard-owned. + # Sentinel-helpers still recognizes the file as wizard-owned. assert sbomify_json_has_wizard_sentinel(path) is True # The non-sentinel payload still made it through. assert data["supplier"] == {"name": "Acme"} @@ -325,6 +325,87 @@ def test_pick_default_workspace_key_handles_empty_and_missing_members() -> None: assert _pick_default_workspace_key([{"key": 42}, {"key": "real"}]) == "real" +def test_resolve_profile_workspace_uses_picked_key_when_readable(monkeypatch) -> None: + """The common case — unscoped token, or scoped token whose workspace is + the picked one — binds to the picked key with a single request.""" + from sbomify_action.cli.wizard.screens import authenticate as auth_mod + + calls: list[str] = [] + + class _FakeClient: + def __init__(self, base_url: str, token: str) -> None: + pass + + def list_contact_profiles(self, key: str) -> list[dict[str, object]]: + calls.append(key) + return [{"id": "cp1", "name": "Default"}] + + monkeypatch.setattr(auth_mod, "SbomifyApiClient", _FakeClient) + workspaces = [{"key": "alpha"}, {"key": "beta"}] + team_key, profiles = auth_mod._resolve_profile_workspace("https://x", "t", workspaces, "beta") + assert team_key == "beta" + assert profiles[0]["id"] == "cp1" + assert calls == ["beta"] + + +def test_resolve_profile_workspace_probes_when_picked_key_forbidden(monkeypatch) -> None: + """A scoped token can't read the is_default_team-picked workspace — + ``/api/v1/workspaces/`` isn't filtered by token scope, so the picker + can land on a workspace the token has no access to. The resolver must + probe the rest and bind to the workspace the token CAN read; silently + returning [] here made the wizard show an empty profile picker and + create new profiles in the wrong workspace.""" + from sbomify_action.cli.wizard.screens import authenticate as auth_mod + from sbomify_action.exceptions import APIError + + class _FakeClient: + def __init__(self, base_url: str, token: str) -> None: + pass + + def list_contact_profiles(self, key: str) -> list[dict[str, object]]: + if key != "scoped": + raise APIError("Failed to list contact profiles. [403] - Forbidden") + return [{"id": "cp-scoped", "name": "Default"}] + + monkeypatch.setattr(auth_mod, "SbomifyApiClient", _FakeClient) + workspaces = [{"key": "default"}, {"key": "other"}, {"key": "scoped"}] + team_key, profiles = auth_mod._resolve_profile_workspace("https://x", "t", workspaces, "default") + assert team_key == "scoped" + assert profiles[0]["id"] == "cp-scoped" + + +def test_resolve_profile_workspace_none_readable(monkeypatch) -> None: + """When no workspace listing succeeds the resolver returns (None, []) + rather than binding profile creation to a workspace the token can't + verify — writing into an unverified workspace is the failure mode + this exists to prevent.""" + from sbomify_action.cli.wizard.screens import authenticate as auth_mod + from sbomify_action.exceptions import APIError + + class _FakeClient: + def __init__(self, base_url: str, token: str) -> None: + pass + + def list_contact_profiles(self, key: str) -> list[dict[str, object]]: + raise APIError("Forbidden") + + monkeypatch.setattr(auth_mod, "SbomifyApiClient", _FakeClient) + team_key, profiles = auth_mod._resolve_profile_workspace("https://x", "t", [{"key": "a"}, {"key": "b"}], "a") + assert team_key is None + assert profiles == [] + + +def test_strip_status_codes() -> None: + from sbomify_action.cli.wizard.screens._base import strip_status_codes + + assert ( + strip_status_codes("Failed to create product 'Notipus'. [403] - You have reached the maximum 1 products.") + == "Failed to create product 'Notipus'. You have reached the maximum 1 products." + ) + assert strip_status_codes("Authentication failed [401]") == "Authentication failed" + assert strip_status_codes("no codes here") == "no codes here" + + def test_sbomify_json_has_wizard_sentinel_helpers(tmp_path: Path) -> None: """sbomify_json_has_wizard_sentinel covers absent / malformed / list / dict.""" assert sbomify_json_has_wizard_sentinel(tmp_path / "absent.json") is False @@ -780,7 +861,7 @@ def test_apply_workflow_emission_uses_created_product_id(tmp_path: Path) -> None state, api = _oidc_apply_state(tmp_path, credential_mode="token") state.plan.create_product = "Acme Widget" state.plan.release_strategy = "tag" - api.create_product.return_value = {"id": "prod-NEW", "name": "Acme Widget"} + api.create_product.return_value = ({"id": "prod-NEW", "name": "Acme Widget"}, True) apply_plan(state, _real_opts(tmp_path)) diff --git a/tests/test_wizard_textual.py b/tests/test_wizard_textual.py index 85abc35d..d12d7324 100644 --- a/tests/test_wizard_textual.py +++ b/tests/test_wizard_textual.py @@ -249,7 +249,7 @@ async def test_enter_on_focused_radio_set_toggles_radio(tmp_path: Path, monkeypa app = WizardApp(_opts(tmp_path)) # Larger viewport so the augmentation panel + profile picker + Next - # button all render — Textual focus/visibility behaviour can shift + # button all render — Textual focus/visibility behavior can shift # when widgets are clipped on tiny pilot terminals. async with app.run_test(size=(120, 60)) as pilot: # Walk to ConfigureSbom (where Augmentation now lives — moved @@ -342,7 +342,7 @@ async def test_escape_from_components_goes_back_in_any_focus_state( The Components screen mounts one PickOrCreate per lockfile; depending on auto-match the user may be focused on the OptionList (existing picked) or the "Create new" Input (no auto-match). Both paths must - honour the screen's Escape binding so Back navigation isn't trapped + honor the screen's Escape binding so Back navigation isn't trapped by whichever widget happened to take focus. """ from textual.widgets import Input, OptionList @@ -576,7 +576,7 @@ async def test_augmentation_default_with_no_profiles_bootstraps_create_and_cance ) -> None: """With zero workspace profiles, the default recommended 'profile' strategy highlights the '+ Create new' sentinel, so Enter routes to - CreateProfileScreen (the bootstrap path). Cancelling that form must + CreateProfileScreen (the bootstrap path). Canceling that form must revert augmentation to Skip — otherwise the screen's own defaults put the user in an Enter→Escape→Enter loop. """ @@ -646,7 +646,7 @@ async def test_augmentation_default_with_no_profiles_bootstraps_create_and_cance assert app.screen is configure_screen pressed = configure_screen.query_one("#augmentation", RadioSet).pressed_button assert pressed is not None and pressed.id == "aug-skip", ( - f"Cancelled CreateProfile must revert augmentation to Skip, got {pressed.id if pressed else None}" + f"Canceled CreateProfile must revert augmentation to Skip, got {pressed.id if pressed else None}" ) assert picker.display is False @@ -700,3 +700,99 @@ async def test_enter_on_focused_back_button_goes_back(tmp_path: Path, monkeypatc assert isinstance(app.screen, WelcomeScreen), ( "Enter on focused Back button must pop the screen, not advance forward" ) + + +async def test_apply_plan_limit_offers_reuse_and_retries(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """A product plan-limit failure with exactly one existing product must + repurpose the primary button as "use existing & retry" — flipping the + plan to the existing product and re-running apply in place. "Back and + retry" alone is a dead end: retrying the same create-product plan + fails identically. + """ + from textual.widgets import Button + + from sbomify_action.cli.wizard import apply as apply_mod + from sbomify_action.cli.wizard.screens.apply import ApplyScreen + from sbomify_action.cli.wizard.state import WorkspaceSnapshot + from sbomify_action.exceptions import PlanLimitError + + _stub_discovery(monkeypatch, []) + calls: list[tuple[str | None, str | None]] = [] + + def fake_apply(state, opts, *, log=None): # noqa: ANN001, ANN202 + calls.append((state.plan.create_product, state.plan.use_product_id)) + if len(calls) == 1: + raise PlanLimitError( + "Could not create product 'Notipus': you have reached the maximum 1 products allowed by your plan.", + resource="product", + ) + + monkeypatch.setattr(apply_mod, "apply_plan", fake_apply) + + app = WizardApp(_opts(tmp_path, dry_run=False)) + async with app.run_test() as pilot: + app.state.api = MagicMock() + app.state.workspace = WorkspaceSnapshot(products=[{"id": "p1", "name": "Existing"}], team_key="acme") + app.state.plan.create_product = "Notipus" + app.push_screen(ApplyScreen()) + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + + screen = app.screen + assert isinstance(screen, ApplyScreen) + continue_btn = screen.query_one("#continue", Button) + assert not continue_btn.disabled, "reuse-existing must be actionable after a product plan-limit failure" + assert "retry" in str(continue_btn.label).lower() + # The pinned banner must NOT leak the HTTP status code. + banner = screen.query_one("#apply-error-banner") + assert "[403]" not in str(banner.render()) + + screen.on_button_pressed(Button.Pressed(continue_btn)) + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + + # Second apply ran with the plan flipped to the existing product. + assert calls == [("Notipus", None), (None, "p1")] + assert not screen.query_one("#continue", Button).disabled + + +async def test_apply_plan_limit_back_jumps_to_product_screen(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """With several existing products the wizard can't pick one for the + user — Back must jump straight to the Pick-a-product step (not strand + the user on Review, where Apply would fail identically).""" + from sbomify_action.cli.wizard import apply as apply_mod + from sbomify_action.cli.wizard.screens.apply import ApplyScreen + from sbomify_action.cli.wizard.screens.product import ProductScreen + from sbomify_action.cli.wizard.state import WorkspaceSnapshot + from sbomify_action.exceptions import PlanLimitError + + _stub_discovery(monkeypatch, []) + + def fake_apply(state, opts, *, log=None): # noqa: ANN001, ANN202 + raise PlanLimitError("plan limit reached", resource="product") + + monkeypatch.setattr(apply_mod, "apply_plan", fake_apply) + + app = WizardApp(_opts(tmp_path, dry_run=False)) + async with app.run_test() as pilot: + app.state.api = MagicMock() + app.state.workspace = WorkspaceSnapshot( + products=[{"id": "p1", "name": "One"}, {"id": "p2", "name": "Two"}], + team_key="acme", + ) + app.state.plan.create_product = "Another" + product_screen = ProductScreen() + app.push_screen(product_screen) + await pilot.pause() + app.push_screen(ApplyScreen()) + await pilot.pause() + await app.workers.wait_for_complete() + await pilot.pause() + + screen = app.screen + assert isinstance(screen, ApplyScreen) + screen._go_back() + await pilot.pause() + assert app.screen is product_screen, "Back after a product plan-limit must land on the product step" From d4f510779d728fc3ad2ea192a39dd8ebd3a42a84 Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Thu, 23 Jul 2026 13:06:26 +0000 Subject: [PATCH 2/4] Address Copilot's round-1 review on the wizard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Keep SbomifyApiClient.create_product(name) -> dict backward-compatible (thin POST wrapper: returns the product, raises PlanLimitError on the plan limit, APIError otherwise). Move the DUPLICATE_NAME get-or-create recovery into a new get_or_create_product(name) -> (product, was_created) that the wizard uses — matching the module's "orchestration lives on top, not inside the endpoint wrapper" design and avoiding a breaking signature change for external callers. - Reword the workspace-scoping log line so it no longer trips the Opengrep python-logger-credential-disclosure heuristic; it only ever logged workspace keys, never a credential. - Update tests to the split API (create_product returns dict; get_or_create_product covers create / duplicate-recovery / not-found re-raise / plan-limit propagation). Co-Authored-By: Claude Opus 4.8 --- sbomify_action/cli/wizard/apply.py | 2 +- .../cli/wizard/screens/authenticate.py | 4 +- sbomify_action/sbomify_api.py | 50 ++++++++++++------- tests/test_sbomify_api.py | 37 ++++++++++++-- tests/test_wizard_emitter.py | 5 +- tests/test_wizard_state.py | 2 +- 6 files changed, 72 insertions(+), 28 deletions(-) diff --git a/sbomify_action/cli/wizard/apply.py b/sbomify_action/cli/wizard/apply.py index 411cc463..5b8884b2 100644 --- a/sbomify_action/cli/wizard/apply.py +++ b/sbomify_action/cli/wizard/apply.py @@ -549,7 +549,7 @@ def _resolve_product(state: WizardState, log: LogFn) -> dict[str, Any] | None: assert state.workspace is not None # narrowed by caller if plan.create_product: - product, was_created = api.create_product(plan.create_product) + product, was_created = api.get_or_create_product(plan.create_product) state.created_product_id = str(product.get("id") or "") verb = "Created" if was_created else "Reused existing" state.applied.append(f"{verb.lower()} product {product.get('name')}") diff --git a/sbomify_action/cli/wizard/screens/authenticate.py b/sbomify_action/cli/wizard/screens/authenticate.py index f305f373..68409812 100644 --- a/sbomify_action/cli/wizard/screens/authenticate.py +++ b/sbomify_action/cli/wizard/screens/authenticate.py @@ -115,8 +115,8 @@ def _probe(key: str) -> tuple[str, list[dict[str, object]] | None]: for key, profiles in results: if profiles is not None: logger.info( - "Token cannot read workspace %s; binding contact profiles to workspace %s instead " - "(the token appears to be scoped to it).", + "Workspace %s is not readable with the current API scope; binding contact " + "profiles to workspace %s instead (it appears to be the scoped workspace).", first_key, key, ) diff --git a/sbomify_action/sbomify_api.py b/sbomify_action/sbomify_api.py index c1f111e0..6e396158 100644 --- a/sbomify_action/sbomify_api.py +++ b/sbomify_action/sbomify_api.py @@ -460,32 +460,22 @@ def get_product_by_name(self, name: str) -> dict[str, Any] | None: return product return None - def create_product(self, name: str) -> tuple[dict[str, Any], bool]: - """Create a product with get-or-create semantics. - - Returns ``(product, was_created)``. Mirrors ``create_component``: - recovers from ``DUPLICATE_NAME`` (status 400 or 409) by looking the - existing product up by name — without this, a retry after a - partially-failed apply (product created, later step failed) dies - here instead of reusing the product it created last time. Raises - ``PlanLimitError`` (with ``resource="product"``) when the team has - hit its product-count limit. + def create_product(self, name: str) -> dict[str, Any]: + """Create a product. Thin wrapper over ``POST /products``. + + Returns the created product dict. Raises ``PlanLimitError`` (tagged + ``resource="product"``) when the team has hit its product-count + limit, and ``APIError`` for any other non-2xx (including a + ``DUPLICATE_NAME`` collision — callers that want get-or-create + semantics use :meth:`get_or_create_product`). """ response = self._request("POST", "/api/v1/products", json_body={"name": name}) if response.ok: - return self._safe_json_dict(response) or {}, True + return self._safe_json_dict(response) or {} body = self._safe_json_dict(response) or {} raw_detail = body.get("detail") error_code = body.get("error_code") or "" - - if response.status_code in (400, 409) and error_code == "DUPLICATE_NAME": - logger.info(f"Product '{name}' already exists, retrieving existing product") - existing = self.get_product_by_name(name) - if existing is not None: - return existing, False - raise APIError(f"Product '{name}' reported as duplicate by API but could not be found via lookup") - if self._is_plan_limit(response.status_code, raw_detail, error_code): # User-actionable, surfaced verbatim in the wizard — no status code. plan_msg = ( @@ -496,6 +486,28 @@ def create_product(self, name: str) -> tuple[dict[str, Any], bool]: raise PlanLimitError(plan_msg, resource="product") raise APIError(self._build_error(f"Failed to create product '{name}'.", response)) + def get_or_create_product(self, name: str) -> tuple[dict[str, Any], bool]: + """Create a product, recovering from a name collision. + + Returns ``(product, was_created)``. On a create failure the product + is looked up by name and reused when found — this is what lets a + retry after a partially-failed apply (product created, a later step + failed) reuse the product instead of dead-ending on ``DUPLICATE_NAME``. + ``PlanLimitError`` propagates unchanged (it is not a name collision, + so a lookup would be wrong); any other error re-raises when no + existing product matches the name. + """ + try: + return self.create_product(name), True + except PlanLimitError: + raise + except APIError: + existing = self.get_product_by_name(name) + if existing is not None: + logger.info(f"Product '{name}' already exists, reusing it") + return existing, False + raise + def attach_components_to_product(self, product_id: str, component_ids: list[str]) -> None: """Set the full component list on a product. diff --git a/tests/test_sbomify_api.py b/tests/test_sbomify_api.py index 0936141c..555bd9c7 100644 --- a/tests/test_sbomify_api.py +++ b/tests/test_sbomify_api.py @@ -380,13 +380,20 @@ def test_list_products_paginates() -> None: def test_create_product() -> None: + """The thin wrapper stays backward-compatible: returns the product dict.""" client, _ = _client_with([_FakeResponse(201, {"id": "p1", "name": "X"})]) - product, was_created = client.create_product("X") + product = client.create_product("X") + assert product["id"] == "p1" + + +def test_get_or_create_product_creates() -> None: + client, _ = _client_with([_FakeResponse(201, {"id": "p1", "name": "X"})]) + product, was_created = client.get_or_create_product("X") assert product["id"] == "p1" assert was_created is True -def test_create_product_recovers_from_duplicate_name() -> None: +def test_get_or_create_product_recovers_from_duplicate_name() -> None: """A DUPLICATE_NAME rejection resolves to the existing product — the retry-after-partial-apply path (product created, later step failed) must reuse the product instead of dead-ending on the duplicate.""" @@ -399,11 +406,35 @@ def test_create_product_recovers_from_duplicate_name() -> None: ), ] ) - product, was_created = client.create_product("X") + product, was_created = client.get_or_create_product("X") assert product["id"] == "p-existing" assert was_created is False +def test_get_or_create_product_reraises_when_not_found() -> None: + """A non-duplicate create failure with no matching existing product + must re-raise, not silently swallow the error.""" + client, _ = _client_with( + [ + _FakeResponse(500, {"detail": "boom"}), + _FakeResponse(200, {"items": [], "pagination": {"has_next": False}}), + ] + ) + with pytest.raises(APIError): + client.get_or_create_product("X") + + +def test_get_or_create_product_propagates_plan_limit() -> None: + """A plan-limit failure is not a name collision — it must propagate as + PlanLimitError, not get masked by a name lookup.""" + client, _ = _client_with( + [_FakeResponse(403, {"detail": "maximum products reached", "error_code": "BILLING_LIMIT_EXCEEDED"})] + ) + with pytest.raises(PlanLimitError) as exc: + client.get_or_create_product("X") + assert exc.value.resource == "product" + + def test_create_product_plan_limit_is_clean_and_typed() -> None: """The plan-limit 403 raises PlanLimitError tagged with the resource, and the message carries the human detail without the status code.""" diff --git a/tests/test_wizard_emitter.py b/tests/test_wizard_emitter.py index 90302795..a815bb44 100644 --- a/tests/test_wizard_emitter.py +++ b/tests/test_wizard_emitter.py @@ -634,7 +634,7 @@ def test_apply_plan_create_new_product(tmp_path: Path) -> None: state = _state(tmp_path) api = state.api assert api is not None - api.create_product.return_value = ({"id": "prod-new", "name": "Widget"}, True) + api.get_or_create_product.return_value = ({"id": "prod-new", "name": "Widget"}, True) api.get_or_create_component.return_value = ("comp-1", True) state.plan = Plan( @@ -651,7 +651,7 @@ def test_apply_plan_create_new_product(tmp_path: Path) -> None: ) apply_mod.apply_plan(state, opts) - api.create_product.assert_called_once_with("Widget") + api.get_or_create_product.assert_called_once_with("Widget") assert state.created_product_id == "prod-new" @@ -688,6 +688,7 @@ def test_apply_plan_dry_run_skips_api_mutations_and_writes(tmp_path: Path) -> No # attach, no patch, no OIDC binding. api.get_or_create_component.assert_not_called() api.create_product.assert_not_called() + api.get_or_create_product.assert_not_called() api.attach_components_to_product.assert_not_called() api.patch_component.assert_not_called() api.create_oidc_binding.assert_not_called() diff --git a/tests/test_wizard_state.py b/tests/test_wizard_state.py index 126cc22f..870dda3d 100644 --- a/tests/test_wizard_state.py +++ b/tests/test_wizard_state.py @@ -861,7 +861,7 @@ def test_apply_workflow_emission_uses_created_product_id(tmp_path: Path) -> None state, api = _oidc_apply_state(tmp_path, credential_mode="token") state.plan.create_product = "Acme Widget" state.plan.release_strategy = "tag" - api.create_product.return_value = ({"id": "prod-NEW", "name": "Acme Widget"}, True) + api.get_or_create_product.return_value = ({"id": "prod-NEW", "name": "Acme Widget"}, True) apply_plan(state, _real_opts(tmp_path)) From 8fcb022ebf0157d5aaa7646305680d74898af80b Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Thu, 23 Jul 2026 13:17:38 +0000 Subject: [PATCH 3/4] Address Copilot's round-2 review on the wizard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - strip_status_codes: anchor the regex to the exact shape _build_error emits — a real HTTP status (1xx–5xx) in brackets, followed by " - " or end-of-string. A bracketed number embedded in a (quoted) name like 'Widget [123]' is now left untouched instead of being mangled. - _resolve_profile_workspace: only switch workspaces on a genuine scope 403, never on a transient error. Added a typed ForbiddenError (403, subclass of APIError) raised by list_contact_profiles; the resolver probes other workspaces only on ForbiddenError and stays on the picked workspace (empty profiles, best-effort) for a 500/timeout — guessing a different workspace off a blip was the wrong-workspace binding this code exists to prevent. - Tests: ForbiddenError on 403, transient-error-stays-put, and the quoted-name / out-of-range-code strip cases. Co-Authored-By: Claude Opus 4.8 --- sbomify_action/cli/wizard/screens/_base.py | 15 ++++-- .../cli/wizard/screens/authenticate.py | 35 +++++++++---- sbomify_action/exceptions.py | 8 +++ sbomify_action/sbomify_api.py | 8 ++- tests/test_sbomify_api.py | 11 ++++ tests/test_wizard_state.py | 52 ++++++++++++++++--- 6 files changed, 105 insertions(+), 24 deletions(-) diff --git a/sbomify_action/cli/wizard/screens/_base.py b/sbomify_action/cli/wizard/screens/_base.py index a7a51bfd..a73e1bdb 100644 --- a/sbomify_action/cli/wizard/screens/_base.py +++ b/sbomify_action/cli/wizard/screens/_base.py @@ -29,10 +29,13 @@ TOTAL_STEPS = 8 -# ``[403] - `` / ``[404]`` markers that API error strings embed for logs. -# The wizard strips them from anything it shows the user — the status code -# is developer-facing noise next to the human-readable detail. -_STATUS_CODE_RE = re.compile(r"\s*\[\d{3}\]\s*-?\s*") +# HTTP status markers as ``_build_error`` emits them: ``prefix [NNN]`` at the +# end of a clause, or ``prefix [NNN] - detail`` when a detail follows. Anchored +# to that exact shape — the code must be a real HTTP status (1xx–5xx) AND be +# followed by `` - `` or the end of the string — so an embedded, quoted name +# like ``'Widget [123]'`` (the ``[123]`` is followed by ``'``, not `` - ``/end) +# is left untouched. +_STATUS_CODE_RE = re.compile(r"\s*\[[1-5]\d{2}\]\s*(?:-\s*|$)") def strip_status_codes(message: str) -> str: @@ -41,7 +44,9 @@ def strip_status_codes(message: str) -> str: ``"Failed to create product 'X'. [403] - You have reached…"`` becomes ``"Failed to create product 'X'. You have reached…"``. Full error text (codes included) still lands in the debug log via the exception itself; - this only cleans what the TUI renders. + this only cleans what the TUI renders. Only markers in the shape + ``_build_error`` produces are stripped, so a bracketed number inside a + product/component name isn't mangled. """ return _STATUS_CODE_RE.sub(" ", message).strip() diff --git a/sbomify_action/cli/wizard/screens/authenticate.py b/sbomify_action/cli/wizard/screens/authenticate.py index 68409812..0cb3b02f 100644 --- a/sbomify_action/cli/wizard/screens/authenticate.py +++ b/sbomify_action/cli/wizard/screens/authenticate.py @@ -12,7 +12,7 @@ from sbomify_action.cli.wizard.screens._base import WizardScreen, strip_status_codes from sbomify_action.cli.wizard.state import WorkspaceSnapshot -from sbomify_action.exceptions import APIError, AuthError +from sbomify_action.exceptions import APIError, AuthError, ForbiddenError from sbomify_action.logging_config import logger from sbomify_action.sbomify_api import SbomifyApiClient @@ -80,10 +80,16 @@ def _resolve_profile_workspace( not the picked key). Verified against production: a token scoped to workspace A lists A's profiles fine and gets 403 for every other key. - So: try the picked key first; on failure, probe the remaining - workspaces in parallel and bind to the first one whose listing - succeeds (for a scoped token exactly one can). Returns - ``(None, [])`` when no workspace is readable. + So: try the picked key first; only when it comes back ``403 Forbidden`` + (a definitive "this token can't touch this workspace") do we probe the + remaining workspaces in parallel and bind to the first that succeeds — + for a scoped token exactly one will. A transient failure on the picked + workspace (500/timeout) must NOT trigger a rebind: guessing a different + workspace off a blip is exactly the wrong-workspace binding this exists + to prevent, so we stay on the picked key with empty profiles (best- + effort; the picker just appears empty and a re-run recovers). Returns + ``(None, [])`` only when the picked workspace is forbidden and no other + workspace is readable either. """ candidates = [str(ws.get("key")) for ws in workspaces if isinstance(ws.get("key"), str) and ws.get("key")] if picked_key is not None and picked_key in candidates: @@ -102,13 +108,22 @@ def _probe(key: str) -> tuple[str, list[dict[str, object]] | None]: # Try the picked workspace alone first — the common case (unscoped # token, or a scoped token whose workspace IS the picked one) needs # exactly one request. - first_key, first_profiles = _probe(candidates[0]) - if first_profiles is not None: - return first_key, first_profiles + picked = candidates[0] + try: + return picked, SbomifyApiClient(base_url, token).list_contact_profiles(picked) + except ForbiddenError: + # Scope denial — the token can't read the picked workspace. Fall + # through to probing the others. + pass + except APIError as e: + # Transient/unknown failure — don't rebind to a different workspace + # off a blip. Stay on the picked key with empty profiles. + logger.warning("Could not list contact profiles for workspace %s: %s", picked, e) + return picked, [] rest = candidates[1:] if not rest: - logger.warning("Could not list contact profiles for workspace %s", first_key) + logger.warning("Workspace %s is not readable with the current API scope, and no other exists", picked) return None, [] with ThreadPoolExecutor(max_workers=min(8, len(rest))) as pool: results = list(pool.map(_probe, rest)) @@ -117,7 +132,7 @@ def _probe(key: str) -> tuple[str, list[dict[str, object]] | None]: logger.info( "Workspace %s is not readable with the current API scope; binding contact " "profiles to workspace %s instead (it appears to be the scoped workspace).", - first_key, + picked, key, ) return key, profiles diff --git a/sbomify_action/exceptions.py b/sbomify_action/exceptions.py index 8862df33..dceae09a 100644 --- a/sbomify_action/exceptions.py +++ b/sbomify_action/exceptions.py @@ -87,6 +87,14 @@ class AuthError(APIError): """Raised when the sbomify API rejects credentials (401).""" +class ForbiddenError(APIError): + """Raised when the sbomify API returns 403 — authenticated but not + permitted (e.g. a workspace-scoped token reaching a workspace outside + its scope). Distinct from ``AuthError`` (401, bad credentials) so callers + can tell "this token can't touch this resource" apart from a transient + failure and react accordingly.""" + + class PlanLimitError(APIError): """Raised when an API operation fails due to plan limits (e.g., max components). diff --git a/sbomify_action/sbomify_api.py b/sbomify_action/sbomify_api.py index 6e396158..e5d4f31f 100644 --- a/sbomify_action/sbomify_api.py +++ b/sbomify_action/sbomify_api.py @@ -17,7 +17,7 @@ import requests -from sbomify_action.exceptions import APIError, AuthError, PlanLimitError +from sbomify_action.exceptions import APIError, AuthError, ForbiddenError, PlanLimitError from sbomify_action.http_client import get_default_headers from sbomify_action.logging_config import logger @@ -604,6 +604,12 @@ def list_contact_profiles(self, team_key: str) -> list[dict[str, Any]]: if response.status_code == 404: logger.debug("Contact profiles endpoint not available for workspace %s", team_key) return [] + if response.status_code == 403: + # Scope denial — this token can't read this workspace. Raise a + # typed error so callers (the wizard's workspace resolver) can + # tell it apart from a transient failure and switch workspaces + # only on a genuine 403, never on a 500/timeout. + raise ForbiddenError(self._build_error("Failed to list contact profiles.", response)) if not response.ok: raise APIError(self._build_error("Failed to list contact profiles.", response)) try: diff --git a/tests/test_sbomify_api.py b/tests/test_sbomify_api.py index 555bd9c7..7afa72e3 100644 --- a/tests/test_sbomify_api.py +++ b/tests/test_sbomify_api.py @@ -490,6 +490,17 @@ def test_list_contact_profiles_404_returns_empty() -> None: assert client.list_contact_profiles("acme-team") == [] +def test_list_contact_profiles_403_raises_forbidden() -> None: + """A 403 raises the typed ForbiddenError (a subclass of APIError) so the + wizard's workspace resolver can tell scope denial apart from a transient + failure and only switch workspaces on the former.""" + from sbomify_action.exceptions import ForbiddenError + + client, _ = _client_with([_FakeResponse(403, {"detail": "Forbidden"})]) + with pytest.raises(ForbiddenError): + client.list_contact_profiles("acme-team") + + def test_list_contact_profiles_success() -> None: # Real endpoint returns a bare list — `[{...}, {...}]` — not a # paginated envelope. Filtered out non-dict entries defensively. diff --git a/tests/test_wizard_state.py b/tests/test_wizard_state.py index 870dda3d..c7932ca9 100644 --- a/tests/test_wizard_state.py +++ b/tests/test_wizard_state.py @@ -356,7 +356,7 @@ def test_resolve_profile_workspace_probes_when_picked_key_forbidden(monkeypatch) returning [] here made the wizard show an empty profile picker and create new profiles in the wrong workspace.""" from sbomify_action.cli.wizard.screens import authenticate as auth_mod - from sbomify_action.exceptions import APIError + from sbomify_action.exceptions import ForbiddenError class _FakeClient: def __init__(self, base_url: str, token: str) -> None: @@ -364,7 +364,7 @@ def __init__(self, base_url: str, token: str) -> None: def list_contact_profiles(self, key: str) -> list[dict[str, object]]: if key != "scoped": - raise APIError("Failed to list contact profiles. [403] - Forbidden") + raise ForbiddenError("Failed to list contact profiles. [403] - Forbidden") return [{"id": "cp-scoped", "name": "Default"}] monkeypatch.setattr(auth_mod, "SbomifyApiClient", _FakeClient) @@ -374,20 +374,47 @@ def list_contact_profiles(self, key: str) -> list[dict[str, object]]: assert profiles[0]["id"] == "cp-scoped" -def test_resolve_profile_workspace_none_readable(monkeypatch) -> None: - """When no workspace listing succeeds the resolver returns (None, []) - rather than binding profile creation to a workspace the token can't - verify — writing into an unverified workspace is the failure mode - this exists to prevent.""" +def test_resolve_profile_workspace_stays_put_on_transient_error(monkeypatch) -> None: + """A transient (non-403) failure on the picked workspace must NOT trigger + a rebind — guessing a different workspace off a 500/timeout is the exact + wrong-workspace binding this function exists to prevent. Stay on the + picked key with empty profiles and never probe the others.""" from sbomify_action.cli.wizard.screens import authenticate as auth_mod from sbomify_action.exceptions import APIError + probed: list[str] = [] + + class _FakeClient: + def __init__(self, base_url: str, token: str) -> None: + pass + + def list_contact_profiles(self, key: str) -> list[dict[str, object]]: + probed.append(key) + if key == "picked": + raise APIError("Failed to list contact profiles. [500] - boom") + return [{"id": "leaked", "name": "Other"}] + + monkeypatch.setattr(auth_mod, "SbomifyApiClient", _FakeClient) + workspaces = [{"key": "picked"}, {"key": "other"}] + team_key, profiles = auth_mod._resolve_profile_workspace("https://x", "t", workspaces, "picked") + assert team_key == "picked" + assert profiles == [] + assert probed == ["picked"], "must not probe other workspaces on a transient error" + + +def test_resolve_profile_workspace_none_readable(monkeypatch) -> None: + """When the picked workspace is forbidden and no other is readable, the + resolver returns (None, []) rather than binding profile creation to a + workspace the token can't verify.""" + from sbomify_action.cli.wizard.screens import authenticate as auth_mod + from sbomify_action.exceptions import ForbiddenError + class _FakeClient: def __init__(self, base_url: str, token: str) -> None: pass def list_contact_profiles(self, key: str) -> list[dict[str, object]]: - raise APIError("Forbidden") + raise ForbiddenError("Forbidden") monkeypatch.setattr(auth_mod, "SbomifyApiClient", _FakeClient) team_key, profiles = auth_mod._resolve_profile_workspace("https://x", "t", [{"key": "a"}, {"key": "b"}], "a") @@ -404,6 +431,15 @@ def test_strip_status_codes() -> None: ) assert strip_status_codes("Authentication failed [401]") == "Authentication failed" assert strip_status_codes("no codes here") == "no codes here" + # A bracketed number inside a (quoted) name is not an HTTP marker — the + # regex is anchored to `` - ``/end-of-string, and here `[123]` is followed + # by a quote, so only the real `[404]` marker is stripped. + assert ( + strip_status_codes("Failed to create component 'Widget [123]'. [404] - not found") + == "Failed to create component 'Widget [123]'. not found" + ) + # A non-HTTP code (outside 1xx–5xx) is left alone. + assert strip_status_codes("weird [999] token") == "weird [999] token" def test_sbomify_json_has_wizard_sentinel_helpers(tmp_path: Path) -> None: From 09e17b33fd58ee2ac632080f18381c4f2edc1338 Mon Sep 17 00:00:00 2001 From: Viktor Petersson Date: Thu, 23 Jul 2026 13:23:47 +0000 Subject: [PATCH 4/4] Address Copilot's round-3 review on the wizard - Factor plan-limit message building into _plan_limit_message(resource, name, raw_detail), shared by create_product and create_component. It uses the backend detail only when it is actually a non-empty string; a list/dict detail falls back to a generic human sentence instead of being interpolated as a Python repr, and it never carries the HTTP status code. Fixes the component path, which previously interpolated a non-string detail and fell back to err_msg (which included [status]). - Test: BILLING_LIMIT_EXCEEDED 403 with a structured (list) detail stays clean and status-code-free. Co-Authored-By: Claude Opus 4.8 --- sbomify_action/sbomify_api.py | 28 +++++++++++++++++----------- tests/test_sbomify_api.py | 21 +++++++++++++++++++++ 2 files changed, 38 insertions(+), 11 deletions(-) diff --git a/sbomify_action/sbomify_api.py b/sbomify_action/sbomify_api.py index e5d4f31f..3a84110f 100644 --- a/sbomify_action/sbomify_api.py +++ b/sbomify_action/sbomify_api.py @@ -358,12 +358,24 @@ def create_component( raise APIError(f"Component '{name}' reported as duplicate by API but could not be found via lookup") if self._is_plan_limit(response.status_code, raw_detail, error_code): - # Plan-limit messages are user-actionable and surfaced verbatim in - # UIs (the wizard's apply banner) — keep them human: no status code. - plan_msg = f"Could not create component '{name}': {raw_detail}" if raw_detail else err_msg - raise PlanLimitError(plan_msg, resource="component") + raise PlanLimitError(self._plan_limit_message("component", name, raw_detail), resource="component") raise APIError(err_msg) + @staticmethod + def _plan_limit_message(resource: str, name: str, raw_detail: Any) -> str: + """Build a human, status-code-free plan-limit message. + + Plan-limit errors are surfaced verbatim in the UI (the wizard's apply + banner), so the message must never carry an HTTP status marker or a + raw structured ``detail``. Uses the backend's string detail only when + it is actually a non-empty string — a list/dict detail (pydantic-shaped + or otherwise) falls back to a generic sentence rather than being + interpolated as a Python repr. + """ + if isinstance(raw_detail, str) and raw_detail: + return f"Could not create {resource} '{name}': {raw_detail}" + return f"Could not create {resource} '{name}': your plan's {resource} limit has been reached." + @staticmethod def _is_plan_limit(status_code: int, raw_detail: Any, error_code: str) -> bool: """True when a create was rejected because the team's plan limit is hit. @@ -477,13 +489,7 @@ def create_product(self, name: str) -> dict[str, Any]: raw_detail = body.get("detail") error_code = body.get("error_code") or "" if self._is_plan_limit(response.status_code, raw_detail, error_code): - # User-actionable, surfaced verbatim in the wizard — no status code. - plan_msg = ( - f"Could not create product '{name}': {raw_detail}" - if isinstance(raw_detail, str) and raw_detail - else f"Could not create product '{name}': your plan's product limit has been reached." - ) - raise PlanLimitError(plan_msg, resource="product") + raise PlanLimitError(self._plan_limit_message("product", name, raw_detail), resource="product") raise APIError(self._build_error(f"Failed to create product '{name}'.", response)) def get_or_create_product(self, name: str) -> tuple[dict[str, Any], bool]: diff --git a/tests/test_sbomify_api.py b/tests/test_sbomify_api.py index 7afa72e3..a3744072 100644 --- a/tests/test_sbomify_api.py +++ b/tests/test_sbomify_api.py @@ -456,6 +456,27 @@ def test_create_component_plan_limit_is_clean_and_typed() -> None: assert "[403]" not in str(exc.value) +def test_create_component_plan_limit_non_string_detail_stays_clean() -> None: + """A BILLING_LIMIT_EXCEEDED 403 whose detail is a structured list (not a + plain string) must not leak the raw repr or the status code into the + user-facing message — it falls back to a generic human sentence.""" + client, _ = _client_with( + [ + _FakeResponse( + 403, + {"detail": [{"msg": "limit"}], "error_code": "BILLING_LIMIT_EXCEEDED"}, + ) + ] + ) + with pytest.raises(PlanLimitError) as exc: + client.create_component("foo", component_type="bom") + message = str(exc.value) + assert exc.value.resource == "component" + assert "[403]" not in message + assert "{'msg'" not in message and "[{" not in message + assert "your plan's component limit has been reached" in message + + def test_attach_components_unions_existing() -> None: client, session = _client_with( [