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: e2c4a6b8d013
+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] = "e2c4a6b8d013"
+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,259 @@
+"""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.routes.patterns import _effective_subscription, _ValidationError
+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.
+    Headers and query parameters cannot supply a missing verified tenant.
+    """
+    user = getattr(g, "user", {}) or {}
+    # OIDC path: tid claim decoded by the verifier into user["tenant"]
+    tid = user.get("tenant")
+    if tid:
+        return tid
+    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, _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:
+        subscription_id = _effective_subscription(subscription_id)
+    except _ValidationError:
+        return jsonify({"error": "Invalid request parameters"}), 400
+
+    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[])
+                    LIMIT %(edges_limit)s
+                    """,
+                    {"node_ids": node_ids, "edges_limit": limit * 10},
+                )
+                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:
+        subscription_id = _effective_subscription(request.args.get("subscription_id"))
+    except _ValidationError:
+        return jsonify({"error": "Invalid request parameters"}), 400
+
+    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 scans s ON s.scan_id::text = ap.scan_id
+                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
+                  AND s.subscription_id = %(subscription_id)s
+                  AND NOT EXISTS (
+                      SELECT 1 FROM graph_nodes hop
+                      WHERE hop.node_id = ANY(ap.path_node_ids)
+                        AND (hop.tenant_id <> %(tenant_id)s OR hop.subscription_id <> %(subscription_id)s)
+                  )
+                ORDER BY ap.path_length ASC, ap.min_confidence DESC
+                LIMIT %(limit)s
+                """,
+                {"scan_id": scan_id, "tenant_id": tenant_id, "subscription_id": subscription_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:
+        subscription_id = _effective_subscription(request.args.get("subscription_id"))
+    except _ValidationError:
+        return jsonify({"error": "Invalid request parameters"}), 400
+
+    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
+                JOIN scans s ON s.scan_id::text = ap.scan_id
+                WHERE ap.path_id = %(path_id)s::uuid
+                  AND ap.tenant_id = %(tenant_id)s
+                  AND s.subscription_id = %(subscription_id)s
+                  AND NOT EXISTS (
+                      SELECT 1 FROM graph_nodes hop
+                      WHERE hop.node_id = ANY(ap.path_node_ids)
+                        AND (hop.tenant_id <> %(tenant_id)s OR hop.subscription_id <> %(subscription_id)s)
+                  )
+                """,
+                {"path_id": path_id, "tenant_id": tenant_id, "subscription_id": subscription_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
+                      AND subscription_id = %(subscription_id)s
+                    """,
+                    {"ids": node_ids, "tenant_id": tenant_id, "subscription_id": subscription_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,19 @@ 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/v1/attack-graph` returns explicitly observed nodes and relationships from the latest published inventory snapshot for an authorized 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/v1/attack-paths` and `GET /api/v1/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 must have a verified tenant claim in either authentication mode. Headers and query parameters cannot supply a missing tenant, including for admins. Requests without a usable tenant return `403 {"error": "tenant_id not available; OIDC authentication required"}`.
+
+Subscription scope follows the patterns API: use a signed subscription claim, or the configured deployment allowlist only when the verified tenant matches the deployment tenant. When that allowlist contains several subscriptions, select one with `subscription_id`. A selection outside the authorized scope returns 400. Graph nodes, paths, path details and every returned hop are restricted to that subscription; paths crossing an unauthorized subscription are omitted.
+
+### 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,6 @@ 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
+
+    compute_attack_paths(scan_id, snapshot.tenant_id, dsn)