coulomb/state-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: project Nexus SBOM state into summaries
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details
Build and Publish Multi-Context Image / build-and-push (push) Successful in 25s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
This commit is contained in:
tegwick 2026-08-22 20:12:16 +02:00
parent e1e259cf87
commit 76e6eda086
4 changed files with 172 additions and 42 deletions
Download patch file Download diff file Expand all files Collapse all files

71
api/routers/state.py
View file

@ -54,6 +54,8 @@ from api.services.summary_cache import (
register_summary_cache_invalidation,
)
from api.services.ops_run_projection import get_ops_run_projection
from api.services.sbom_nexus import get_json as get_sbom_nexus_json
from api.services.sbom_nexus import reads_from_nexus
def _dual_workplan_refs(
@ -364,23 +366,7 @@ async def build_state_summary(session: AsyncSession) -> StateSummary:
)}
contribution_counts = {**contrib_type_counts, **contrib_status_counts}
# Licence risk: copyleft packages in direct prod deps
_COPYLEFT_PATS = ("GPL", "AGPL", "LGPL", "EUPL", "CDDL", "MPL")
copyleft_risk_rows = await session.execute(
select(func.count()).select_from(SBOMEntry)
.where(SBOMEntry.is_direct.is_(True))
.where(SBOMEntry.is_dev.is_(False))
)
# Filter in Python since ILIKE across multiple patterns is verbose in SQLAlchemy
all_direct_prod_rows = await session.execute(
select(SBOMEntry.license_spdx)
.where(SBOMEntry.is_direct.is_(True))
.where(SBOMEntry.is_dev.is_(False))
)
licence_risk_count = sum(
1 for (lic,) in all_direct_prod_rows.all()
if lic and any(pat in lic.upper() for pat in _COPYLEFT_PATS)
)
licence_risk_count, _, _ = await _sbom_metrics(session)
# Open capability requests (non-terminal statuses)
open_cap_req_count = (await session.execute(
@ -646,23 +632,7 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
)}
contribution_counts = {**contrib_type_counts, **contrib_status_counts}
_COPYLEFT_PATS = ("GPL", "AGPL", "LGPL", "EUPL", "CDDL", "MPL")
all_direct_prod_rows = await session.execute(
select(SBOMEntry.license_spdx)
.where(SBOMEntry.is_direct.is_(True))
.where(SBOMEntry.is_dev.is_(False))
)
licence_risk_count = sum(
1 for (lic,) in all_direct_prod_rows.all()
if lic and any(pat in lic.upper() for pat in _COPYLEFT_PATS)
)
snapshot_count, package_total = (await session.execute(
select(
func.count(SBOMSnapshot.id),
func.coalesce(func.sum(SBOMSnapshot.entry_count), 0),
)
)).one()
licence_risk_count, snapshot_count, package_total = await _sbom_metrics(session)
open_cap_req_count = (await session.execute(
select(func.count()).select_from(CapabilityRequest).where(
@ -742,6 +712,39 @@ async def _build_dashboard_overview(session: AsyncSession) -> DashboardOverview:
)
async def _sbom_metrics(session: AsyncSession) -> tuple[int, int, int]:
"""Return compatibility summary metrics from the selected SBOM authority."""
if reads_from_nexus():
snapshots = await get_sbom_nexus_json("/sbom/snapshots/")
report = await get_sbom_nexus_json("/sbom/report/licences/")
return (
int(report.get("copyleft_direct_count") or 0),
len(snapshots),
sum(int(snapshot.get("entry_count") or 0) for snapshot in snapshots),
)
copyleft_patterns = ("GPL", "AGPL", "LGPL", "EUPL", "CDDL", "MPL")
rows = await session.execute(
select(SBOMEntry.license_spdx)
.where(SBOMEntry.is_direct.is_(True))
.where(SBOMEntry.is_dev.is_(False))
)
licence_risk_count = sum(
1
for (licence,) in rows.all()
if licence and any(pattern in licence.upper() for pattern in copyleft_patterns)
)
snapshot_count, package_total = (
await session.execute(
select(
func.count(SBOMSnapshot.id),
func.coalesce(func.sum(SBOMSnapshot.entry_count), 0),
)
)
).one()
return licence_risk_count, int(snapshot_count or 0), int(package_total or 0)
async def _build_domain_summaries(session: AsyncSession) -> list[DomainSummary]:
"""Compute per-domain stats for the state summary."""
domains_rows = await session.execute(