Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
124 changes: 124 additions & 0 deletions apps/backend/api/v1/_snapshot_anchor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 TRUSCA contributors
"""
The shared ``?scan_id=`` / ``?release=`` snapshot anchor for detail reads.

Fourteen endpoints let a caller read a project surface as of one succeeded
scan. Until now the only way to name that scan was its UUID, which a caller
cannot know in advance: answering "what shipped in 4.0" meant listing releases,
matching the label client-side, then re-requesting with the id it found. Two
round-trips, and no URL you could write down or paste into a ticket.

``release`` closes that: ``/notice?release=4.0`` is a permanent address for a
version, because a label identifies exactly one live snapshot (see
``tasks.scan_retention.supersede_prior_release_scans``).

Resolving it HERE rather than in each service is deliberate. The alternative —
a second parameter threaded through fourteen endpoints and the ten services
behind them — multiplies the number of places the precedence rule could drift,
for a translation that is the same everywhere: turn a label into the scan id
the endpoint already knows how to handle. Services keep their existing
``snapshot_scan_id`` contract and never learn that labels exist.

Precedence: ``scan_id`` wins when both are given. It names one immutable
snapshot, whereas a label names whichever snapshot currently holds it — so the
more specific of the two should not be overridden by the looser one.

Authorization: none here, matching ``services.scan_resolution``. The lookup is
scoped to ``project_id`` and both "no such label" and "no such project you can
see" surface as the same 404, so resolving before the endpoint's team check
tells an outside caller nothing it could not already infer.
"""

from __future__ import annotations

import uuid

from fastapi import Depends, Query
from sqlalchemy import String, cast, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from core.db import get_db
from models import Scan
from services.scan_resolution import SnapshotScanNotFound

_SCAN_ID_DESCRIPTION = (
"Optional release-snapshot anchor. Read this surface as of ONE specific "
"succeeded scan instead of the project's current state. Must belong to this "
"project and be succeeded, else 404. Takes precedence over ``release``."
)

_RELEASE_DESCRIPTION = (
"Optional version anchor — read this surface as of the release carrying "
"this label (e.g. '4.0'). Equivalent to looking the label up on "
"``/releases?release=`` and pinning the ``scan_id`` it returns, but as one "
"permanent URL. Matched exactly, whitespace trimmed. Unknown label → 404. "
"Ignored when ``scan_id`` is also given."
)


async def resolve_release_label(
session: AsyncSession,
project_id: uuid.UUID,
label: str,
) -> uuid.UUID | None:
"""Return the live snapshot carrying *label*, or ``None``.

Mirrors the releases-list filter exactly — succeeded, not superseded,
trimmed comparison — so a label that appears there resolves here and one
that does not, does not. Superseded rows are excluded because a rescan
moves the label: ``4.0`` must mean the snapshot that currently holds it,
not every scan that ever claimed it.

Served by ``ix_scans_project_release_label``.
"""
stripped = label.strip()
if not stripped:
return None
stmt = (
select(Scan.id)
.where(Scan.project_id == project_id)
.where(cast(Scan.status, String) == "succeeded")
.where(Scan.superseded_at.is_(None))
.where(func.jsonb_typeof(Scan.scan_metadata["release"]) == "string")
.where(func.btrim(Scan.scan_metadata["release"].astext) == stripped)
.order_by(Scan.created_at.desc(), Scan.id.desc())
.limit(1)
)
result = await session.execute(stmt)
return result.scalar_one_or_none()


async def snapshot_anchor(
project_id: uuid.UUID,
scan_id: uuid.UUID | None = Query(default=None, description=_SCAN_ID_DESCRIPTION),
release: str | None = Query(
default=None, max_length=100, description=_RELEASE_DESCRIPTION
),
session: AsyncSession = Depends(get_db),
) -> uuid.UUID | None:
"""Resolve the effective ``scan_id`` for a detail read.

Returns what the endpoint's service already expects: ``None`` for "current
state", or a scan id to pin. An unresolvable label raises
:class:`SnapshotScanNotFound`, which ``core.errors`` renders as the same
existence-hiding 404 an unresolvable ``scan_id`` produces — a caller must
not be able to tell "that version does not exist" from "that scan id is not
yours".

A pinned ``scan_id`` is returned unvalidated; the service still passes it
through ``resolve_snapshot_scan_id``, which owns the ownership + succeeded
checks. Validating twice here would cost a round-trip on every request to
move a check that is already in the right place.
"""
if scan_id is not None or release is None:
return scan_id
resolved = await resolve_release_label(session, project_id, release)
if resolved is None:
raise SnapshotScanNotFound(
f"no release labelled {release!r} in project {project_id}"
)
return resolved


__all__ = ["resolve_release_label", "snapshot_anchor"]
11 changes: 2 additions & 9 deletions apps/backend/api/v1/compliance.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
from fastapi import APIRouter, Depends, Query, Request, Response, status
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.db import get_db
from core.errors import problem_response
from core.security import CurrentUser, require_role
Expand Down Expand Up @@ -114,15 +115,7 @@ async def list_project_compliance_endpoint(
pattern=r"^(category|license_name|spdx_id|affected_count)$",
),
order: str = Query(default="desc", pattern=r"^(asc|desc)$"),
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, the "
"grid reflects this specific succeeded scan instead of the "
"project's latest succeeded scan. Must belong to this project "
"and be succeeded, else 404."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> Response:
Expand Down
11 changes: 2 additions & 9 deletions apps/backend/api/v1/licenses.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
from fastapi import APIRouter, Depends, Query, Request, Response, status
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.db import get_db
from core.errors import problem_response
from core.security import CurrentUser, require_role
Expand Down Expand Up @@ -120,15 +121,7 @@ async def list_project_licenses_endpoint(
"non_commercial = CC-BY-NC…. Omit to list all licenses."
),
),
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, list "
"license rows of this SPECIFIC succeeded scan instead of the project's "
"latest succeeded scan. Must belong to this project and be succeeded, "
"else 404. Omit for the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> Response:
Expand Down
23 changes: 3 additions & 20 deletions apps/backend/api/v1/obligations.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
from fastapi import APIRouter, Depends, Query, Request, Response, status
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.db import get_db
from core.errors import problem_response
from core.ratelimit import limiter
Expand Down Expand Up @@ -130,15 +131,7 @@ async def list_project_obligations_endpoint(
pattern=r"^(category|license_name|kind|affected_count)$",
),
order: str = Query(default="desc", pattern=r"^(asc|desc)$"),
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, list "
"obligation rows of this SPECIFIC succeeded scan instead of the "
"project's latest succeeded scan. Must belong to this project and be "
"succeeded, else 404. Omit for the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> Response:
Expand Down Expand Up @@ -312,17 +305,7 @@ def _format_content_disposition(project_name: str, ext: str) -> str:
async def get_project_notice_endpoint(
request: Request,
project_id: uuid.UUID,
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor. When given, compose the NOTICE "
"from this SPECIFIC succeeded scan instead of the project's latest "
"succeeded scan — this is how the attribution document for an "
"already-shipped release stays retrievable after a newer scan "
"succeeds. Must belong to this project and be succeeded, else 404. "
"Omit for the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
fmt: str = Query(
default="text",
alias="format",
Expand Down
12 changes: 2 additions & 10 deletions apps/backend/api/v1/policy_gate.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
from sqlalchemy import String, cast, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.api_key_auth import get_api_key_principal
from core.audit import bind_audit_team, get_audit_context, mask_sensitive_columns
from core.authz import assert_team_access
Expand Down Expand Up @@ -239,16 +240,7 @@ def _build_response_body(result: GateResult) -> GateResultResponse:
async def get_gate_result_endpoint(
request: Request,
project_id: uuid.UUID,
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, evaluate "
"the build gate against this SPECIFIC succeeded scan instead of the "
"project's latest succeeded scan (so the Overview gate card can reflect "
"a pinned release). Must belong to this project and be succeeded, else "
"404. Omit for the default latest-succeeded behaviour (the CI contract)."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
ref: str | None = Query(
default=None,
max_length=255,
Expand Down
31 changes: 4 additions & 27 deletions apps/backend/api/v1/projects.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
from fastapi import APIRouter, Depends, File, Query, Request, Response, UploadFile, status
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.api_key_auth import require_role_or_api_key
from core.audit import bind_audit_team, get_audit_context, mask_sensitive_columns
from core.config import scan_trigger_rate_limit
Expand Down Expand Up @@ -429,15 +430,7 @@ async def delete_project_endpoint(
async def get_project_overview_endpoint(
request: Request,
project_id: uuid.UUID,
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, aggregate "
"this SPECIFIC succeeded scan instead of the project's latest succeeded "
"scan. Must belong to this project and be succeeded, else 404. Omit for "
"the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> Response:
Expand Down Expand Up @@ -593,15 +586,7 @@ async def list_project_components_endpoint(
),
sort: str = Query(default="name", pattern=r"^(name|severity|license)$"),
order: str = Query(default="asc", pattern=r"^(asc|desc)$"),
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, list "
"components of this SPECIFIC succeeded scan instead of the project's "
"latest succeeded scan. Must belong to this project and be succeeded, "
"else 404. Omit for the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> Response:
Expand Down Expand Up @@ -833,15 +818,7 @@ async def diff_project_releases_endpoint(
async def get_dependency_graph_endpoint(
request: Request,
project_id: uuid.UUID,
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor. When given, serialize this SPECIFIC "
"succeeded scan's graph instead of the project's latest succeeded scan. "
"Must belong to this project and be succeeded, else 404 (existence-hide). "
"Omit for the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> Response:
Expand Down
10 changes: 2 additions & 8 deletions apps/backend/api/v1/reports.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
from fastapi.concurrency import run_in_threadpool
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.authz import assert_team_access
from core.db import get_db
from core.errors import problem_response
Expand Down Expand Up @@ -463,14 +464,7 @@ async def list_project_report_history_endpoint(
"Omit for all four types."
),
),
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional filter — return only rows where ``scan_id`` matches. "
"Pair with ``type=sbom`` etc. to find all artefacts produced for "
"one scan."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
page: int = Query(default=1, ge=1, description="1-based page number."),
page_size: int = Query(
default=PAGE_SIZE_DEFAULT,
Expand Down
11 changes: 2 additions & 9 deletions apps/backend/api/v1/sbom.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.api_key_auth import require_role_or_api_key
from core.authz import assert_team_access
from core.config import (
Expand Down Expand Up @@ -245,15 +246,7 @@ async def export_project_sbom_endpoint(
alias="format",
description="SBOM output format.",
),
scan_id: uuid.UUID | None = Query(
default=None,
description=(
"Optional release-snapshot anchor (feature #28). When given, export "
"this SPECIFIC succeeded scan instead of the project's latest succeeded "
"scan. Must belong to this project and be succeeded, else 404. Omit for "
"the default latest-succeeded behaviour."
),
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
profile: SBOMProfile | None = Query(
default=None,
description=(
Expand Down
11 changes: 3 additions & 8 deletions apps/backend/api/v1/source_tree.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
from fastapi.responses import JSONResponse, StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession

from api.v1._snapshot_anchor import snapshot_anchor
from core.db import get_db
from core.errors import problem_response
from core.security import CurrentUser, require_role
Expand Down Expand Up @@ -116,10 +117,7 @@ async def get_source_tree(
),
page: int = Query(default=1, ge=1, description="1-based page index."),
size: int = Query(default=100, ge=1, le=500, description="Page size (max 500)."),
scan_id: uuid.UUID | None = Query(
default=None,
description="Scan to read; defaults to the project's latest scan.",
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
session: AsyncSession = Depends(get_db),
actor: CurrentUser = Depends(require_role("developer")),
) -> JSONResponse | SourceTreePage:
Expand Down Expand Up @@ -185,10 +183,7 @@ async def get_source_file(
request: Request,
project_id: uuid.UUID,
path: str = Query(description="File to read, relative to the source root."),
scan_id: uuid.UUID | None = Query(
default=None,
description="Scan to read; defaults to the project's latest scan.",
),
scan_id: uuid.UUID | None = Depends(snapshot_anchor),
raw: bool = Query(
default=False,
description=(
Expand Down
Loading
Loading