feat: route SBOM writes to Nexus behind flag

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834

api/config.py

@@ -21,6 +21,7 @@ class Settings(BaseSettings):
     ops_run_sla_hours: float = 1.0
     sbom_nexus_url: str | None = None
     sbom_nexus_read_mode: Literal["legacy", "nexus"] = "legacy"
+    sbom_nexus_write_mode: Literal["legacy", "nexus"] = "legacy"
     sbom_nexus_timeout_seconds: float = 5.0
 
 

api/routers/sbom.py

@@ -20,7 +20,13 @@ from api.schemas.sbom import (
     SBOMSnapshotRead,
 )
 from api.services.legacy_meter import identity_from_request, record_legacy_usage
-from api.services.sbom_nexus import SBOMNexusError, get_json, reads_from_nexus
+from api.services.sbom_nexus import (
+    SBOMNexusError,
+    get_json,
+    post_json,
+    reads_from_nexus,
+    writes_to_nexus,
+)
 
 router = APIRouter(prefix="/sbom", tags=["sbom"])
 logger = logging.getLogger(__name__)
@@ -63,6 +69,29 @@ async def ingest_sbom(
 ) -> dict:
     """Create a new SBOM snapshot for a repo. Previous snapshots are retained."""
     repo = await _get_repo_by_slug(body.repo_slug, session)
+    if writes_to_nexus():
+        payload = await _nexus_post("/sbom/ingest/", body=body.model_dump(mode="json"))
+        try:
+            snapshot_at = datetime.fromisoformat(
+                payload["snapshot_at"].replace("Z", "+00:00")
+            )
+            result = {
+                "repo_slug": payload["repo_slug"],
+                "snapshot_id": payload["snapshot_id"],
+                "ingested": payload["ingested"],
+                "snapshot_at": payload["snapshot_at"],
+            }
+        except (AttributeError, KeyError, TypeError, ValueError) as exc:
+            raise HTTPException(
+                status_code=502,
+                detail="SBOM Nexus returned an invalid ingest response",
+            ) from exc
+        repo.last_sbom_at = snapshot_at
+        repo.sbom_source = "sbom-nexus"
+        await session.commit()
+        await _meter_compat(session, request, "POST", "/sbom/ingest/")
+        return result
+
     now = datetime.now(tz=timezone.utc)
 
     snap = SBOMSnapshot(
@@ -315,6 +344,13 @@ async def _nexus_get(path: str, *, params: dict | None = None):
         raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
 
 
+async def _nexus_post(path: str, *, body: dict):
+    try:
+        return await post_json(path, body=body)
+    except SBOMNexusError as exc:
+        raise HTTPException(status_code=exc.status_code, detail=exc.detail) from exc
+
+
 async def _local_repo_ids(items: list[dict], session: AsyncSession) -> dict[str, uuid.UUID]:
     slugs = {item.get("repo_slug") for item in items}
     if None in slugs:

api/services/sbom_nexus.py

@@ -21,12 +21,18 @@ def reads_from_nexus() -> bool:
     return settings.sbom_nexus_read_mode == "nexus"
 
 
-async def get_json(
+def writes_to_nexus() -> bool:
+    return settings.sbom_nexus_write_mode == "nexus"
+
+
+async def request_json(
+    method: str,
     path: str,
     *,
     params: dict[str, Any] | None = None,
+    body: dict[str, Any] | None = None,
 ) -> Any:
-    """Fetch one Nexus resource; never fall back silently to legacy storage."""
+    """Call Nexus; never fall back silently to legacy storage."""
     if not settings.sbom_nexus_url:
         raise SBOMNexusError(503, "SBOM Nexus read mode is enabled without SBOM_NEXUS_URL")
 
@@ -35,7 +41,7 @@ async def get_json(
             base_url=settings.sbom_nexus_url.rstrip("/"),
             timeout=settings.sbom_nexus_timeout_seconds,
         ) as client:
-            response = await client.get(path, params=params)
+            response = await client.request(method, path, params=params, json=body)
     except httpx.RequestError as exc:
         raise SBOMNexusError(502, f"SBOM Nexus is unavailable: {exc.__class__.__name__}") from exc
 
@@ -54,3 +60,11 @@ async def get_json(
         return response.json()
     except ValueError as exc:
         raise SBOMNexusError(502, "SBOM Nexus returned invalid JSON") from exc
+
+
+async def get_json(path: str, *, params: dict[str, Any] | None = None) -> Any:
+    return await request_json("GET", path, params=params)
+
+
+async def post_json(path: str, *, body: dict[str, Any]) -> Any:
+    return await request_json("POST", path, body=body)

deploy/railiance/apps/charts/state-hub/templates/configmap.yaml

@@ -8,4 +8,5 @@ data:
   CORS_ORIGINS: {{ .Values.config.corsOrigins | quote }}
   SBOM_NEXUS_URL: {{ .Values.config.sbomNexusUrl | quote }}
   SBOM_NEXUS_READ_MODE: {{ .Values.config.sbomNexusReadMode | quote }}
+  SBOM_NEXUS_WRITE_MODE: {{ .Values.config.sbomNexusWriteMode | quote }}
 {{- end }}

deploy/railiance/apps/charts/state-hub/values.yaml

@@ -24,6 +24,7 @@ config:
   corsOrigins: "http://localhost:3000,http://127.0.0.1:3000,http://localhost:3001,http://127.0.0.1:3001"
   sbomNexusUrl: ""
   sbomNexusReadMode: legacy
+  sbomNexusWriteMode: legacy
 
 secret:
   name: state-hub-env

deploy/railiance/apps/helm/state-hub-values.yaml

@@ -15,6 +15,8 @@ config:
   sbomNexusUrl: "http://sbom-nexus.sbom-nexus.svc.cluster.local:8010"
   # Reversible T04 read cutover; set back to `legacy` to roll back.
   sbomNexusReadMode: nexus
+  # Write authority moves only after the independently flagged image is live.
+  sbomNexusWriteMode: legacy
 
 resources:
   # The single 4-core node currently has less than 250m unallocated. Keep enough

tests/test_sbom_nexus_compat.py

@@ -156,6 +156,58 @@ async def test_nexus_read_mode_does_not_move_ingest_write_authority(client, monk
     assert response.json()["ingested"] == 1
 
 
+async def test_nexus_write_mode_moves_authority_and_updates_projection(client, monkeypatch):
+    repo = await _create_repo(client)
+    snapshot_id = str(uuid.uuid4())
+    observed = {}
+
+    async def fake_post(path: str, *, body):
+        observed["path"] = path
+        observed["body"] = body
+        return {
+            "repo_slug": "testrepo",
+            "snapshot_id": snapshot_id,
+            "ingested": 1,
+            "snapshot_at": "2026-08-22T13:00:00Z",
+            "status": "ingested",
+        }
+
+    monkeypatch.setattr(settings, "sbom_nexus_read_mode", "legacy")
+    monkeypatch.setattr(settings, "sbom_nexus_write_mode", "nexus")
+    monkeypatch.setattr("api.routers.sbom.post_json", fake_post)
+    response = await client.post(
+        "/sbom/ingest/",
+        json={
+            "repo_slug": "testrepo",
+            "entries": [
+                {
+                    "package_name": "fastapi",
+                    "package_version": "0.115.0",
+                    "ecosystem": "python",
+                    "license_spdx": "MIT",
+                }
+            ],
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "repo_slug": "testrepo",
+        "snapshot_id": snapshot_id,
+        "ingested": 1,
+        "snapshot_at": "2026-08-22T13:00:00Z",
+    }
+    assert observed["path"] == "/sbom/ingest/"
+    assert observed["body"]["entries"][0]["ecosystem"] == "python"
+
+    legacy_snapshots = await client.get("/sbom/snapshots/?repo_slug=testrepo")
+    projected_repo = await client.get("/repos/testrepo")
+    assert legacy_snapshots.json() == []
+    assert projected_repo.json()["id"] == repo["id"]
+    assert projected_repo.json()["last_sbom_at"] == "2026-08-22T13:00:00Z"
+    assert projected_repo.json()["sbom_source"] == "sbom-nexus"
+
+
 async def test_nexus_failure_is_visible_and_does_not_fall_back(client, monkeypatch):
     await _create_repo(client)