feat(sbom): add authoritative Nexus client

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d

tests/test_sbom.py

@@ -4,11 +4,15 @@ import json
 import subprocess
 from pathlib import Path
 
+import httpx
 import pytest
 
 from repo_manager.cli import main
 from repo_manager.sbom_client import (
     SBOMContractError,
+    SBOMNexusClient,
+    SBOMNexusConfig,
+    SBOMServiceError,
     licence_report_from_snapshot,
     scan_repository_via_nexus,
     validate_snapshot_contract,
@@ -174,3 +178,157 @@ def test_cli_scan_preserves_output_file_behavior(monkeypatch, tmp_path: Path, ca
     assert exit_code == 0
     assert json.loads(output.read_text())["schema"] == "sbom-nexus.snapshot.v1"
     assert json.loads(capsys.readouterr().out)["product_owner"] == "sbom-nexus"
+
+
+def test_service_config_is_explicit_bounded_and_redacts_token(monkeypatch) -> None:
+    monkeypatch.setenv("SBOM_NEXUS_URL", "https://nexus.example.test/")
+    monkeypatch.setenv("SBOM_NEXUS_TIMEOUT_SECONDS", "12.5")
+    monkeypatch.setenv("SBOM_NEXUS_TOKEN", "do-not-print-this")
+
+    config = SBOMNexusConfig.from_environment()
+
+    assert config.base_url == "https://nexus.example.test"
+    assert config.timeout_seconds == 12.5
+    assert "do-not-print-this" not in repr(config)
+    with pytest.raises(ValueError, match="must not contain credentials"):
+        SBOMNexusConfig("https://user:secret@nexus.example.test")
+    with pytest.raises(ValueError, match="between 0.1 and 300"):
+        SBOMNexusConfig("https://nexus.example.test", timeout_seconds=301)
+
+
+def test_authoritative_client_calls_pinned_routes_and_marks_service_context() -> None:
+    calls = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(request)
+        if request.url.path.startswith("/repositories/"):
+            return httpx.Response(200, json={"slug": "demo", "active": True})
+        if request.method == "POST":
+            return httpx.Response(
+                200,
+                json={
+                    "repo_slug": "demo",
+                    "snapshot_id": "snapshot-1",
+                    "status": "ingested",
+                    "entry_count": 2,
+                    "snapshot_at": "2026-08-22T20:00:00Z",
+                    "source_revision": "abc123",
+                },
+            )
+        if request.url.path == "/sbom/report/licences/":
+            return httpx.Response(
+                200,
+                json={
+                    "groups": [],
+                    "copyleft_direct_prod": [],
+                    "copyleft_direct_count": 0,
+                },
+            )
+        return httpx.Response(
+            200,
+            json={
+                "repo_slug": "demo",
+                "last_attempt_at": "2026-08-22T20:00:00Z",
+                "last_success_at": "2026-08-22T20:00:00Z",
+                "last_status": "ingested",
+                "entry_count": 2,
+                "snapshot_id": "snapshot-1",
+                "entries": [],
+            },
+        )
+
+    client = SBOMNexusClient(
+        SBOMNexusConfig("https://nexus.example.test", bearer_token="runtime-secret"),
+        transport=httpx.MockTransport(handler),
+    )
+
+    projected = client.upsert_repository(
+        "demo",
+        nexus_checkout_path="/srv/controlled/demo/abc123",
+    )
+    ingested = client.ingest_repository(
+        "demo",
+        expected_source_revision="abc123",
+        operation_id="operation-1",
+    )
+    latest = client.latest_snapshot("demo")
+    report = client.licence_report()
+
+    assert [request.url.path for request in calls] == [
+        "/repositories/demo",
+        "/sbom/demo/ingest",
+        "/sbom/demo",
+        "/sbom/report/licences/",
+    ]
+    assert all(request.headers["Authorization"] == "Bearer runtime-secret" for request in calls)
+    assert calls[1].headers["Idempotency-Key"] == "operation-1"
+    assert projected["repo_manager_context"]["writes_state"] is True
+    assert ingested["repo_manager_context"]["mode"] == "authoritative-service"
+    assert latest["repo_manager_context"]["authoritative"] is True
+    assert report["repo_manager_context"]["operation"] == "licence-report"
+
+
+def test_authoritative_client_fails_closed_on_source_revision_mismatch() -> None:
+    transport = httpx.MockTransport(
+        lambda request: httpx.Response(
+            200,
+            json={
+                "repo_slug": "demo",
+                "snapshot_id": "snapshot-1",
+                "status": "ingested",
+                "entry_count": 1,
+                "snapshot_at": "2026-08-22T20:00:00Z",
+                "source_revision": "different",
+            },
+        )
+    )
+    client = SBOMNexusClient(SBOMNexusConfig("https://nexus.example.test"), transport=transport)
+
+    with pytest.raises(SBOMServiceError) as raised:
+        client.ingest_repository("demo", expected_source_revision="expected")
+
+    assert raised.value.code == "source_revision_mismatch"
+    assert raised.value.mutation_may_have_committed is True
+
+
+def test_authoritative_client_has_sanitized_timeout_and_http_failures() -> None:
+    secret = "never-echo-this-token"
+
+    def timeout_handler(request: httpx.Request) -> httpx.Response:
+        raise httpx.ReadTimeout("upstream timed out", request=request)
+
+    timeout_client = SBOMNexusClient(
+        SBOMNexusConfig("https://nexus.example.test", bearer_token=secret),
+        transport=httpx.MockTransport(timeout_handler),
+    )
+    with pytest.raises(SBOMServiceError) as timeout:
+        timeout_client.latest_snapshot("demo")
+    assert timeout.value.to_dict()["code"] == "timeout"
+    assert secret not in str(timeout.value)
+    assert secret not in json.dumps(timeout.value.to_dict())
+
+    http_client = SBOMNexusClient(
+        SBOMNexusConfig("https://nexus.example.test", bearer_token=secret),
+        transport=httpx.MockTransport(lambda request: httpx.Response(503, json={"detail": secret})),
+    )
+    with pytest.raises(SBOMServiceError) as failed:
+        http_client.ingest_repository("demo")
+    assert failed.value.status_code == 503
+    assert failed.value.mutation_may_have_committed is True
+    assert secret not in str(failed.value)
+    assert secret not in json.dumps(failed.value.to_dict())
+
+
+def test_authoritative_client_rejects_malformed_success_without_exposing_body() -> None:
+    client = SBOMNexusClient(
+        SBOMNexusConfig("https://nexus.example.test"),
+        transport=httpx.MockTransport(
+            lambda request: httpx.Response(200, json={"unexpected": "sensitive body"})
+        ),
+    )
+
+    with pytest.raises(SBOMServiceError) as failed:
+        client.latest_snapshot("demo")
+
+    assert failed.value.code == "contract_error"
+    assert "sensitive body" not in str(failed.value)