state-hub/api/routers/sbom.py

import uuid
import logging
from datetime import datetime, timezone

from fastapi import APIRouter, Depends, HTTPException, Query, Request
from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from api.database import get_session
from api.models.managed_repo import ManagedRepo
from api.models.sbom_entry import Ecosystem, SBOMEntry
from api.models.sbom_snapshot import SBOMSnapshot
from api.schemas.sbom import (
    LicenceGroup,
    LicenceReport,
    SBOMEntryRead,
    SBOMIngest,
    SBOMRepoView,
    SBOMSnapshotDetail,
    SBOMSnapshotRead,
)
from api.services.legacy_meter import identity_from_request, record_legacy_usage
from api.services.sbom_nexus import (
    SBOMNexusError,
    get_json,
    post_json,
    reads_from_nexus,
    writes_to_nexus,
)
from api.services.repository_aliases import resolve_repository_slug

router = APIRouter(prefix="/sbom", tags=["sbom"])
logger = logging.getLogger(__name__)

_COPYLEFT_PATTERNS = {"GPL", "AGPL", "LGPL", "EUPL", "CDDL", "MPL"}


def _is_copyleft(spdx: str | None) -> bool:
    if not spdx:
        return False
    upper = spdx.upper()
    return any(pat in upper for pat in _COPYLEFT_PATTERNS)


def _latest_snapshot_ids_subquery():
    """Subquery returning the latest SBOMSnapshot.id per repo."""
    max_at_sq = (
        select(SBOMSnapshot.repo_id, func.max(SBOMSnapshot.snapshot_at).label("max_at"))
        .group_by(SBOMSnapshot.repo_id)
        .subquery("max_snap_at")
    )
    return (
        select(SBOMSnapshot.id)
        .join(
            max_at_sq,
            and_(
                SBOMSnapshot.repo_id == max_at_sq.c.repo_id,
                SBOMSnapshot.snapshot_at == max_at_sq.c.max_at,
            ),
        )
        .subquery("latest_snap_ids")
    )


@router.post("/ingest/")
async def ingest_sbom(
    body: SBOMIngest,
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> dict:
    """Create a new SBOM snapshot for a repo. Previous snapshots are retained."""
    resolution = await resolve_repository_slug(session, body.repo_slug)
    repo = resolution.repo
    if writes_to_nexus():
        nexus_body = body.model_dump(mode="json")
        nexus_body["repo_slug"] = resolution.canonical_slug
        payload = await _nexus_post("/sbom/ingest/", body=nexus_body)
        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(
        repo_id=repo.id,
        snapshot_at=now,
        source="manual",
        entry_count=len(body.entries),
        created_at=now,
    )
    session.add(snap)
    await session.flush()  # materialise snap.id before creating entries

    for entry in body.entries:
        sbom = SBOMEntry(
            repo_id=repo.id,
            snapshot_id=snap.id,
            package_name=entry.package_name,
            package_version=entry.package_version,
            ecosystem=entry.ecosystem,
            license_spdx=entry.license_spdx,
            is_direct=entry.is_direct,
            is_dev=entry.is_dev,
            snapshot_at=now,
            created_at=now,
        )
        session.add(sbom)

    repo.last_sbom_at = now
    if not repo.sbom_source:
        repo.sbom_source = "manual"

    await session.commit()
    await _meter_compat(session, request, "POST", "/sbom/ingest/")
    return {
        "repo_slug": resolution.canonical_slug,
        "snapshot_id": str(snap.id),
        "ingested": len(body.entries),
        "snapshot_at": now.isoformat(),
    }


@router.get("/snapshots/", response_model=list[SBOMSnapshotRead])
async def list_snapshots(
    request: Request,
    repo_slug: str | None = Query(None),
    session: AsyncSession = Depends(get_session),
) -> list[SBOMSnapshotRead]:
    """List SBOM snapshots, newest first. Optionally filter by repo."""
    await _meter_compat(session, request, "GET", "/sbom/snapshots/")
    if reads_from_nexus():
        if repo_slug:
            repo_slug = (
                await resolve_repository_slug(session, repo_slug)
            ).canonical_slug
        payload = await _nexus_get(
            "/sbom/snapshots/",
            params={"repo_slug": repo_slug} if repo_slug else None,
        )
        return await _translate_snapshots(payload, session)

    q = select(SBOMSnapshot).order_by(SBOMSnapshot.snapshot_at.desc())
    if repo_slug:
        repo = await _get_repo_by_slug(repo_slug, session)
        q = q.where(SBOMSnapshot.repo_id == repo.id)
    result = await session.execute(q)
    return [SBOMSnapshotRead.model_validate(s) for s in result.scalars().all()]


@router.get("/snapshots/{snapshot_id}", response_model=SBOMSnapshotDetail)
async def get_snapshot(
    snapshot_id: uuid.UUID,
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> SBOMSnapshotDetail:
    """Get a snapshot with its full entry list."""
    await _meter_compat(session, request, "GET", "/sbom/snapshots/{snapshot_id}")
    if reads_from_nexus():
        payload = await _nexus_get(f"/sbom/snapshots/{snapshot_id}")
        repo_ids = await _local_repo_ids([payload], session)
        repo_id = repo_ids[payload["repo_slug"]]
        translated = _translate_snapshot(payload, repo_id)
        translated["entries"] = [
            _translate_entry(entry, repo_id) for entry in payload.get("entries", [])
        ]
        return SBOMSnapshotDetail.model_validate(translated)

    snap = await session.get(SBOMSnapshot, snapshot_id)
    if snap is None:
        raise HTTPException(status_code=404, detail=f"Snapshot '{snapshot_id}' not found")
    result = await session.execute(
        select(SBOMEntry)
        .where(SBOMEntry.snapshot_id == snapshot_id)
        .order_by(SBOMEntry.package_name)
    )
    entries = list(result.scalars().all())
    return SBOMSnapshotDetail(
        id=snap.id,
        repo_id=snap.repo_id,
        snapshot_at=snap.snapshot_at,
        source=snap.source,
        entry_count=snap.entry_count,
        created_at=snap.created_at,
        entries=[SBOMEntryRead.model_validate(e) for e in entries],
    )


@router.get("/")
async def list_sbom_entries(
    request: Request,
    repo_slug: str | None = Query(None),
    ecosystem: Ecosystem | None = Query(None),
    license_spdx: str | None = Query(None),
    is_direct: bool | None = Query(None),
    is_dev: bool | None = Query(None),
    session: AsyncSession = Depends(get_session),
) -> list[SBOMEntryRead]:
    """Return entries from the latest snapshot per repo (default) or filter by repo."""
    await _meter_compat(session, request, "GET", "/sbom/")
    if reads_from_nexus():
        if repo_slug:
            repo_slug = (
                await resolve_repository_slug(session, repo_slug)
            ).canonical_slug
        params = {
            key: value
            for key, value in {
                "repo_slug": repo_slug,
                "ecosystem": ecosystem.value if ecosystem is not None else None,
                "license_spdx": license_spdx,
                "is_direct": is_direct,
                "is_dev": is_dev,
            }.items()
            if value is not None
        }
        payload = await _nexus_get("/sbom/", params=params)
        repo_ids = await _local_repo_ids(payload, session)
        return [
            SBOMEntryRead.model_validate(_translate_entry(entry, repo_ids[entry["repo_slug"]]))
            for entry in payload
        ]

    if repo_slug:
        repo = await _get_repo_by_slug(repo_slug, session)
        latest_snap_id_sq = (
            select(SBOMSnapshot.id)
            .where(SBOMSnapshot.repo_id == repo.id)
            .order_by(SBOMSnapshot.snapshot_at.desc())
            .limit(1)
            .scalar_subquery()
        )
        q = select(SBOMEntry).where(SBOMEntry.snapshot_id == latest_snap_id_sq)
    else:
        latest_ids_sq = _latest_snapshot_ids_subquery()
        q = select(SBOMEntry).where(SBOMEntry.snapshot_id.in_(select(latest_ids_sq.c.id)))

    if ecosystem is not None:
        q = q.where(SBOMEntry.ecosystem == ecosystem)
    if license_spdx:
        q = q.where(SBOMEntry.license_spdx == license_spdx)
    if is_direct is not None:
        q = q.where(SBOMEntry.is_direct == is_direct)
    if is_dev is not None:
        q = q.where(SBOMEntry.is_dev == is_dev)
    q = q.order_by(SBOMEntry.package_name)
    result = await session.execute(q)
    return [SBOMEntryRead.model_validate(e) for e in result.scalars().all()]


@router.get("/report/licences/", response_model=LicenceReport)
async def licence_report(
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> LicenceReport:
    """Group latest-snapshot SBOM entries by SPDX licence identifier, flag copyleft."""
    await _meter_compat(session, request, "GET", "/sbom/report/licences/")
    if reads_from_nexus():
        payload = await _nexus_get("/sbom/report/licences/")
        return LicenceReport.model_validate(payload)

    latest_ids_sq = _latest_snapshot_ids_subquery()
    rows = await session.execute(
        select(SBOMEntry, ManagedRepo.slug)
        .join(ManagedRepo, ManagedRepo.id == SBOMEntry.repo_id)
        .where(SBOMEntry.snapshot_id.in_(select(latest_ids_sq.c.id)))
    )
    groups: dict[str | None, dict] = {}
    copyleft_direct_count = 0
    for entry, repo_slug in rows.all():
        key = entry.license_spdx
        if key not in groups:
            groups[key] = {"count": 0, "repos": set()}
        groups[key]["count"] += 1
        groups[key]["repos"].add(repo_slug)
        if _is_copyleft(key) and entry.is_direct and not entry.is_dev:
            copyleft_direct_count += 1

    licence_groups = [
        LicenceGroup(
            license_spdx=lic,
            count=info["count"],
            repos=sorted(info["repos"]),
            is_copyleft=_is_copyleft(lic),
        )
        for lic, info in sorted(groups.items(), key=lambda x: -x[1]["count"])
    ]
    return LicenceReport(groups=licence_groups, copyleft_direct_count=copyleft_direct_count)


@router.get("/{repo_slug}", response_model=SBOMRepoView)
async def get_repo_sbom(
    repo_slug: str,
    request: Request,
    session: AsyncSession = Depends(get_session),
) -> SBOMRepoView:
    """Return the latest snapshot entries for a specific repo."""
    resolution = await resolve_repository_slug(session, repo_slug)
    repo = resolution.repo
    await _meter_compat(session, request, "GET", "/sbom/{repo_slug}")
    if reads_from_nexus():
        payload = await _nexus_get(f"/sbom/{resolution.canonical_slug}")
        payload["entries"] = [
            _translate_entry(entry, repo.id) for entry in payload.get("entries", [])
        ]
        return SBOMRepoView.model_validate(payload)

    latest_snap_id_sq = (
        select(SBOMSnapshot.id)
        .where(SBOMSnapshot.repo_id == repo.id)
        .order_by(SBOMSnapshot.snapshot_at.desc())
        .limit(1)
        .scalar_subquery()
    )
    rows = await session.execute(
        select(SBOMEntry)
        .where(SBOMEntry.snapshot_id == latest_snap_id_sq)
        .order_by(SBOMEntry.package_name)
    )
    entries = list(rows.scalars().all())
    return SBOMRepoView(
        repo_slug=resolution.canonical_slug,
        last_sbom_at=repo.last_sbom_at,
        entry_count=len(entries),
        entries=[SBOMEntryRead.model_validate(e) for e in entries],
    )


async def _get_repo_by_slug(slug: str, session: AsyncSession) -> ManagedRepo:
    resolution = await resolve_repository_slug(session, slug)
    return resolution.repo


async def _nexus_get(path: str, *, params: dict | None = None):
    try:
        return await get_json(path, params=params)
    except SBOMNexusError as exc:
        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:
        raise HTTPException(status_code=502, detail="SBOM Nexus response omitted repo_slug")
    repo_ids: dict[str, uuid.UUID] = {}
    for slug in slugs:
        resolution = await resolve_repository_slug(session, slug, required=False)
        if resolution is not None:
            repo_ids[slug] = resolution.repo.id
    missing = sorted(slugs - repo_ids.keys())
    if missing:
        raise HTTPException(
            status_code=502,
            detail=f"SBOM Nexus returned repositories absent from State Hub: {', '.join(missing)}",
        )
    return repo_ids


async def _translate_snapshots(
    items: list[dict], session: AsyncSession
) -> list[SBOMSnapshotRead]:
    repo_ids = await _local_repo_ids(items, session)
    return [
        SBOMSnapshotRead.model_validate(_translate_snapshot(item, repo_ids[item["repo_slug"]]))
        for item in items
    ]


def _translate_snapshot(item: dict, repo_id: uuid.UUID) -> dict:
    translated = {**item, "repo_id": repo_id}
    source = translated.get("source")
    if (
        translated.get("legacy_id")
        and isinstance(source, str)
        and source.startswith("state-hub:")
    ):
        translated["source"] = source.removeprefix("state-hub:")
    return translated


def _translate_entry(item: dict, repo_id: uuid.UUID) -> dict:
    return {**item, "repo_id": repo_id}


async def _meter_compat(
    session: AsyncSession,
    request: Request,
    method: str,
    route: str,
) -> None:
    """Meter the compatibility surface without making route success depend on it."""
    interface_key = f"rest_api:{method} {route}"
    try:
        await record_legacy_usage(
            session,
            interface_key=interface_key,
            interface_kind="rest_api",
            replacement_ref="sbom-nexus:/sbom/",
            owner_component="state-hub.sbom-compat",
            replacement_verified=reads_from_nexus(),
            identity=identity_from_request(request),
        )
    except Exception:
        await session.rollback()
        logger.warning("legacy-meter failed to record %s", interface_key, exc_info=True)