feat(graph): BFS attack path traversal and API endpoints

alembic/versions/f2a3b4c5d6e7_attack_paths.py

@@ -0,0 +1,65 @@
+"""Add attack_paths table for pre-computed BFS traversal results.
+
+Revision ID: f2a3b4c5d6e7
+Revises: e1f2a3b4c5d6
+Create Date: 2026-09-24 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "f2a3b4c5d6e7"
+down_revision: Union[str, Sequence[str], None] = "e1f2a3b4c5d6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "attack_paths",
+        sa.Column("path_id", postgresql.UUID(), nullable=False),
+        sa.Column("tenant_id", sa.Text(), nullable=False),
+        sa.Column("scan_id", sa.Text(), nullable=False),
+        sa.Column("source_node_id", postgresql.UUID(), nullable=False),
+        sa.Column("target_node_id", postgresql.UUID(), nullable=False),
+        sa.Column("path_node_ids", postgresql.ARRAY(postgresql.UUID()), nullable=False),
+        sa.Column("path_length", sa.Integer(), nullable=False),
+        sa.Column("min_confidence", sa.Float(), nullable=False, server_default=sa.text("1.0")),
+        sa.Column(
+            "relationship_types", postgresql.ARRAY(sa.Text()), nullable=False, server_default=sa.text("ARRAY[]::text[]")
+        ),
+        sa.Column("computed_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.ForeignKeyConstraint(
+            ["source_node_id"],
+            ["graph_nodes.node_id"],
+            name="attack_paths_source_fkey",
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["target_node_id"],
+            ["graph_nodes.node_id"],
+            name="attack_paths_target_fkey",
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("path_id", name="attack_paths_pkey"),
+    )
+    op.create_index("idx_attack_paths_tenant_scan", "attack_paths", ["tenant_id", "scan_id"])
+    op.create_index("idx_attack_paths_source", "attack_paths", ["source_node_id"])
+    op.create_index("idx_attack_paths_target", "attack_paths", ["target_node_id"])
+    op.create_index(
+        "uq_attack_paths_source_target_scan",
+        "attack_paths",
+        ["source_node_id", "target_node_id", "scan_id"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("uq_attack_paths_source_target_scan", table_name="attack_paths")
+    op.drop_index("idx_attack_paths_target", table_name="attack_paths")
+    op.drop_index("idx_attack_paths_source", table_name="attack_paths")
+    op.drop_index("idx_attack_paths_tenant_scan", table_name="attack_paths")
+    op.drop_table("attack_paths")

api/app.py

@@ -256,6 +256,7 @@ def verify_jwt() -> None:
     # ------------------------------------------------------------------ #
     from api.routes.ai import ai_bp
     from api.routes.assurance import assurance_bp
+    from api.routes.attack_graph import attack_graph_bp
     from api.routes.cbom import cbom_bp
     from api.routes.compliance import compliance_bp
     from api.routes.drift import drift_bp
@@ -268,6 +269,7 @@ def verify_jwt() -> None:
 
     app.register_blueprint(ai_bp)
     app.register_blueprint(assurance_bp)
+    app.register_blueprint(attack_graph_bp)
     app.register_blueprint(cbom_bp)
     app.register_blueprint(compliance_bp)
     app.register_blueprint(drift_bp)

api/routes/attack_graph.py

@@ -0,0 +1,232 @@
+"""Attack graph API: resource nodes, edges, and pre-computed attack paths."""
+
+import logging
+import os
+
+import psycopg2.extras
+from flask import Blueprint, g, jsonify, request
+
+from api.models.finding import DatabaseManager
+from api.validation import ValidationError, positive_integer, uuid_string
+
+attack_graph_bp = Blueprint("attack_graph", __name__)
+logger = logging.getLogger(__name__)
+
+_DEFAULT_LIMIT = 100
+_MAX_LIMIT = 500
+
+
+def _get_db() -> DatabaseManager:
+    if "db" not in g:
+        g.db = DatabaseManager(os.environ["DATABASE_URL"])
+        g.db.connect()
+    return g.db
+
+
+def _tenant_id() -> str | None:
+    """Resolve tenant_id from the verified principal.
+
+    The 'tenant' field comes from the verified token's tenant claim.
+    For tokens without a tenant claim, admins may supply
+    X-Tenant-Id as a request header (never a query param, which leaks into
+    logs and caches). Non-admin tokens cannot override the header.
+    """
+    user = getattr(g, "user", {}) or {}
+    # OIDC path: tid claim decoded by the verifier into user["tenant"]
+    tid = user.get("tenant")
+    if tid:
+        return tid
+    # Shared-secret path: admin-only header override for multi-tenant deployments
+    if user.get("role") == "admin":
+        return request.headers.get("X-Tenant-Id") or None
+    return None
+
+
+@attack_graph_bp.teardown_request
+def _close_db(exc):
+    db = g.pop("db", None)
+    if db is not None:
+        db.close()
+
+
+@attack_graph_bp.get("/api/v1/attack-graph")
+def get_attack_graph():
+    """Return graph nodes and edges for the caller's tenant (latest snapshot).
+
+    Query params: subscription_id (optional), limit (default 100, max 500)
+    """
+    try:
+        limit = positive_integer(int(request.args.get("limit", _DEFAULT_LIMIT)), "limit")
+        if limit > _MAX_LIMIT:
+            limit = _MAX_LIMIT
+        subscription_id = request.args.get("subscription_id")
+    except (ValidationError, ValueError):
+        return jsonify({"error": "Invalid request parameters"}), 400
+
+    tenant_id = _tenant_id()
+    if not tenant_id:
+        # A caller without a verified tenant or admin fallback cannot access
+        # tenant-scoped graph evidence.
+        return jsonify({"error": "tenant_id not available; OIDC authentication required"}), 403
+
+    try:
+        conn = _get_db().conn
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            if subscription_id:
+                cur.execute(
+                    """
+                    SELECT n.node_id::text, n.resource_id, n.resource_type, n.name,
+                           n.location, n.resource_group, n.subscription_id, n.snapshot_id,
+                           n.updated_at
+                    FROM current_graph_nodes n
+                    WHERE n.tenant_id = %(tenant_id)s
+                      AND n.subscription_id = %(subscription_id)s
+                    ORDER BY n.updated_at DESC
+                    LIMIT %(limit)s
+                    """,
+                    {"tenant_id": tenant_id, "subscription_id": subscription_id, "limit": limit},
+                )
+            else:
+                cur.execute(
+                    """
+                    SELECT n.node_id::text, n.resource_id, n.resource_type, n.name,
+                           n.location, n.resource_group, n.subscription_id, n.snapshot_id,
+                           n.updated_at
+                    FROM current_graph_nodes n
+                    WHERE n.tenant_id = %(tenant_id)s
+                    ORDER BY n.updated_at DESC
+                    LIMIT %(limit)s
+                    """,
+                    {"tenant_id": tenant_id, "limit": limit},
+                )
+            nodes = cur.fetchall()
+
+            node_ids = [row["node_id"] for row in nodes]
+            edges: list = []
+            if node_ids:
+                cur.execute(
+                    """
+                    SELECT e.edge_id::text, e.source_node_id::text, e.target_node_id::text,
+                           e.relationship_type, e.confidence, e.evidence_source, e.collected_at
+                    FROM current_graph_edges e
+                    WHERE e.source_node_id = ANY(%(node_ids)s::uuid[])
+                      AND e.target_node_id = ANY(%(node_ids)s::uuid[])
+                    """,
+                    {"node_ids": node_ids},
+                )
+                edges = cur.fetchall()
+    except Exception:
+        logger.exception("get_attack_graph failed for tenant %s", tenant_id)
+        return jsonify({"error": "internal server error"}), 500
+
+    return jsonify({"nodes": [dict(r) for r in nodes], "edges": [dict(r) for r in edges]})
+
+
+@attack_graph_bp.get("/api/v1/attack-paths")
+def list_attack_paths():
+    """Return pre-computed attack paths for a scan.
+
+    Query params: scan_id (required), limit (default 100, max 500)
+    """
+    scan_id = request.args.get("scan_id")
+    if not scan_id:
+        return jsonify({"error": "scan_id is required"}), 400
+    try:
+        scan_id = uuid_string(scan_id, "scan_id")
+        limit = positive_integer(int(request.args.get("limit", _DEFAULT_LIMIT)), "limit")
+        if limit > _MAX_LIMIT:
+            limit = _MAX_LIMIT
+    except (ValidationError, ValueError):
+        return jsonify({"error": "Invalid request parameters"}), 400
+
+    tenant_id = _tenant_id()
+    if not tenant_id:
+        # A caller without a verified tenant or admin fallback cannot access
+        # tenant-scoped graph evidence.
+        return jsonify({"error": "tenant_id not available; OIDC authentication required"}), 403
+
+    try:
+        conn = _get_db().conn
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(
+                """
+                SELECT ap.path_id::text, ap.source_node_id::text, ap.target_node_id::text,
+                       ap.path_node_ids, ap.path_length, ap.min_confidence,
+                       ap.relationship_types, ap.computed_at,
+                       src.resource_type AS source_type, src.name AS source_name,
+                       tgt.resource_type AS target_type, tgt.name AS target_name
+                FROM attack_paths ap
+                JOIN graph_nodes src ON src.node_id = ap.source_node_id
+                JOIN graph_nodes tgt ON tgt.node_id = ap.target_node_id
+                WHERE ap.scan_id = %(scan_id)s
+                  AND ap.tenant_id = %(tenant_id)s
+                ORDER BY ap.path_length ASC, ap.min_confidence DESC
+                LIMIT %(limit)s
+                """,
+                {"scan_id": scan_id, "tenant_id": tenant_id, "limit": limit},
+            )
+            rows = cur.fetchall()
+    except Exception:
+        logger.exception("list_attack_paths failed for scan %s", scan_id)
+        return jsonify({"error": "internal server error"}), 500
+
+    return jsonify({"scan_id": scan_id, "paths": [dict(r) for r in rows]})
+
+
+@attack_graph_bp.get("/api/v1/attack-paths/<path_id>")
+def get_attack_path(path_id: str):
+    """Return a single attack path with full node detail for each hop."""
+    try:
+        path_id = uuid_string(path_id, "path_id")
+    except (ValidationError, ValueError):
+        return jsonify({"error": "Invalid request parameters"}), 400
+
+    tenant_id = _tenant_id()
+    if not tenant_id:
+        # A caller without a verified tenant or admin fallback cannot access
+        # tenant-scoped graph evidence.
+        return jsonify({"error": "tenant_id not available; OIDC authentication required"}), 403
+
+    try:
+        conn = _get_db().conn
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(
+                """
+                SELECT ap.path_id::text, ap.scan_id, ap.source_node_id::text,
+                       ap.target_node_id::text, ap.path_node_ids, ap.path_length,
+                       ap.min_confidence, ap.relationship_types, ap.computed_at
+                FROM attack_paths ap
+                WHERE ap.path_id = %(path_id)s::uuid
+                  AND ap.tenant_id = %(tenant_id)s
+                """,
+                {"path_id": path_id, "tenant_id": tenant_id},
+            )
+            row = cur.fetchone()
+
+        if row is None:
+            return jsonify({"error": "not found"}), 404
+
+        path = dict(row)
+
+        # Fetch full node detail for each hop
+        node_ids = [str(nid) for nid in path["path_node_ids"]]
+        if node_ids:
+            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+                cur.execute(
+                    """
+                    SELECT node_id::text, resource_id, resource_type, name,
+                           location, resource_group, subscription_id
+                    FROM graph_nodes
+                    WHERE node_id = ANY(%(ids)s::uuid[])
+                      AND tenant_id = %(tenant_id)s
+                    """,
+                    {"ids": node_ids, "tenant_id": tenant_id},
+                )
+                nodes_by_id = {r["node_id"]: dict(r) for r in cur.fetchall()}
+            path["hops"] = [nodes_by_id.get(str(nid), {"node_id": str(nid)}) for nid in path["path_node_ids"]]
+    except Exception:
+        logger.exception("get_attack_path failed for path_id %s", path_id)
+        return jsonify({"error": "internal server error"}), 500
+
+    path["path_node_ids"] = [str(nid) for nid in path["path_node_ids"]]
+    return jsonify(path)

docs/api-reference.md

@@ -697,3 +697,17 @@ The following endpoints are called by the frontend but have no backend implement
 | Endpoint | Used by | Status |
 |---|---|---|
 | `GET /api/monitoring` | Monitoring page — score trend chart, category distribution | Deferred. Score and findings data come from `GET /api/score` and `GET /api/findings` instead. |
+
+---
+
+## Attack graph endpoints
+
+`GET /api/attack-graph` returns explicitly observed nodes and relationships from the latest published inventory snapshot for each subscription in the caller's tenant. Complete snapshots expire absent resources and relationships within their tenant/subscription scope. Partial snapshots retain historical evidence in storage, while current graph queries and traversal exclude unobserved resources and relationships. Failed collection keeps the previous published snapshot. `GET /api/attack-paths` and `GET /api/attack-paths/<path_id>` expose paths computed for a requested scan. Successful subsequent scans replace prior paths for the same tenant/subscription, including scans with no findings.
+
+### Authentication requirement
+
+Callers use the tenant claim preserved by the token verifier in either authentication mode. A supplied `X-Tenant-Id` header cannot override that claim. An authenticated admin without a tenant claim may supply `X-Tenant-Id` explicitly; viewer and operator tokens cannot select a tenant. Requests without a usable tenant return `403 {"error": "tenant_id not available; OIDC authentication required"}`. Tenant query parameters never determine graph scope.
+
+### Attack-path retention
+
+Attack paths are replaced by a newer successful scan within the same tenant/subscription. Clean scans remove prior paths even when there are no findings. Failed scans and delayed older scans retain the newer published evidence. Scope publication and traversal serialize on the same database lock.

scanner/graph/graph_populator.py

@@ -11,6 +11,7 @@
 from scanner.arg_inventory import InventoryResource, InventoryStatus
 from scanner.graph.node_service import graph_connection, link_findings_to_nodes, lock_graph_scopes, populate_nodes
 from scanner.graph.edge_detector import detect_all_edges
+from scanner.graph.path_traversal import compute_attack_paths
 
 if TYPE_CHECKING:
     from scanner.arg_inventory import InventorySnapshot
@@ -189,3 +190,10 @@ def populate_graph(scan_id: str, snapshot: InventorySnapshot, dsn: str) -> None:
         )
     except Exception:
         logger.warning("graph: population failed for scan %s", scan_id, exc_info=True)
+        return
+
+    try:
+        path_count = compute_attack_paths(scan_id, snapshot.tenant_id, dsn)
+        logger.info("graph: computed %d attack paths for scan %s", path_count, scan_id)
+    except Exception as exc:
+        logger.warning("graph: path traversal failed for scan %s: %s", scan_id, exc)