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
@@ -267,6 +268,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,235 @@
+"""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.
+
+    OIDC mode: the 'tenant' field is populated from the 'tid' claim in the token.
+    Shared-secret mode: 'tenant' is always None; 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/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 as exc:
+        return jsonify({"error": str(exc)}), 400
+
+    tenant_id = _tenant_id()
+    if not tenant_id:
+        # Shared-secret callers have no tenant claim; OIDC is required for
+        # tenant-scoped graph endpoints. Return 403 (not 400) since the
+        # request is well-formed but the auth method is insufficient.
+        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 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 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 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/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 as exc:
+        return jsonify({"error": str(exc)}), 400
+
+    tenant_id = _tenant_id()
+    if not tenant_id:
+        # Shared-secret callers have no tenant claim; OIDC is required for
+        # tenant-scoped graph endpoints. Return 403 (not 400) since the
+        # request is well-formed but the auth method is insufficient.
+        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/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 as exc:
+        return jsonify({"error": str(exc)}), 400
+
+    tenant_id = _tenant_id()
+    if not tenant_id:
+        # Shared-secret callers have no tenant claim; OIDC is required for
+        # tenant-scoped graph endpoints. Return 403 (not 400) since the
+        # request is well-formed but the auth method is insufficient.
+        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`, `GET /api/attack-paths`, and `GET /api/attack-paths/<path_id>` expose the attack graph computed from the most recent scan for the caller's tenant.
+
+### Authentication requirement
+
+These endpoints are **OIDC-only**. In `shared_secret` mode no tenant claim is present in the token, so the endpoints return `403 {"error": "tenant_id not available; OIDC authentication required"}`. This is intentional: the graph is strictly scoped per tenant and there is no safe way to infer a tenant from a shared-secret token.
+
+### Attack-path retention
+
+Attack paths are written per scan and are not automatically pruned. In long-running deployments, old scan paths accumulate in `attack_paths`. Retention policy (e.g. keep only the N most recent scans per tenant) is tracked in issue #333 and will be addressed in a follow-up.

scanner/graph/graph_populator.py

@@ -13,6 +13,7 @@
 from scanner.arg_inventory import InventoryResource
 from scanner.graph.node_service import link_findings_to_nodes, 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
@@ -139,3 +140,9 @@ def populate_graph(scan_id: str, snapshot: InventorySnapshot, dsn: str) -> None:
         logger.info("graph: linked %d findings to nodes for scan %s", link_count, scan_id)
     except Exception as exc:
         logger.warning("graph: finding link failed for scan %s: %s", scan_id, exc)
+
+    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)