feat(graph): evidence graph foundation - resource inventory schema and scan wiring

.github/workflows/ci.yml

@@ -5,6 +5,10 @@ on:
     branches:
       - dev
       - main
+      - feat/311-finding-lifecycle
+      - feat/331-evidence-graph-foundation
+      - feat/332-evidence-graph-node-edge-population
+
   # Post-merge assurance: re-run the full gate on the merged result, so a green
   # PR run against an older base is not the only evidence for dev and main.
   push:
@@ -573,7 +577,7 @@ jobs:
 
       - name: Start image
         env:
-          POSTGRES_PORT: ${{ job.services.postgres.ports[5432] }}
+          POSTGRES_PORT: ${{ job.services.postgres.ports['5432'] }}
         run: |
           set -euo pipefail
           JWT_SECRET="$(python3 -c 'import secrets; print(secrets.token_urlsafe(32))')"

.github/workflows/codeql.yml

@@ -5,6 +5,10 @@ on:
     branches:
       - dev
       - main
+      - feat/311-finding-lifecycle
+      - feat/331-evidence-graph-foundation
+      - feat/332-evidence-graph-node-edge-population
+
   push:
     branches:
       - dev

.github/workflows/dco.yml

@@ -2,7 +2,7 @@ name: Developer Certificate of Origin
 
 on:
   pull_request:
-    branches: [dev, main]
+    branches: [dev, main, feat/311-finding-lifecycle, feat/331-evidence-graph-foundation, feat/332-evidence-graph-node-edge-population]
 
 permissions:
   contents: read

.github/workflows/dependency-review.yml

@@ -2,7 +2,7 @@ name: Dependency Review
 
 on:
   pull_request:
-    branches: [dev, main]
+    branches: [dev, main, feat/311-finding-lifecycle, feat/331-evidence-graph-foundation, feat/332-evidence-graph-node-edge-population]
 
 permissions:
   contents: read

.github/workflows/website.yml

@@ -14,7 +14,7 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [dev, main]
+    branches: [dev, main, feat/311-finding-lifecycle, feat/331-evidence-graph-foundation, feat/332-evidence-graph-node-edge-population]
   workflow_dispatch:
 
 permissions:

alembic/env.py

@@ -9,7 +9,7 @@
 config = context.config
 
 if config.config_file_name is not None:
-    fileConfig(config.config_file_name)
+    fileConfig(config.config_file_name, disable_existing_loggers=False)
 
 # OpenShield intentionally uses raw SQL rather than ORM metadata. Migration
 # revisions therefore define every schema operation explicitly.

alembic/versions/a8d9c2e4f601_finding_lifecycle.py

@@ -0,0 +1,163 @@
+"""Add finding lifecycle tables: scan_rule_outcomes, scan_lifecycle_applications,
+finding_fingerprints, finding_lifecycles, finding_lifecycle_transitions, patterns.
+
+Revision ID: a8d9c2e4f601
+Revises: b6d2f8a4c1e7
+Create Date: 2026-08-30 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+# Revision identifiers, used by Alembic.
+revision: str = "a8d9c2e4f601"
+down_revision: Union[str, Sequence[str], None] = "b6d2f8a4c1e7"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE scan_rule_outcomes (
+            id BIGSERIAL PRIMARY KEY,
+            scan_id UUID NOT NULL REFERENCES scans(scan_id),
+            rule_id TEXT NOT NULL,
+            rule_version TEXT NOT NULL DEFAULT '1',
+            status TEXT NOT NULL,
+            error_category TEXT,
+            collector_version TEXT NOT NULL DEFAULT '1',
+            inventory_boundary TEXT NOT NULL,
+            tenant_id TEXT NOT NULL,
+            subscription_id TEXT NOT NULL,
+            started_at TIMESTAMPTZ,
+            completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            CONSTRAINT ck_scan_rule_outcomes_status
+                CHECK (status IN (
+                    'SUCCESS','EMPTY_SUCCESS','PERMISSION_DENIED',
+                    'TIMEOUT','FAILED','NOT_APPLICABLE'
+                )),
+            CONSTRAINT uq_scan_rule_outcomes_scan_rule
+                UNIQUE (scan_id, rule_id)
+        )
+        """
+    )
+
+    op.execute(
+        """
+        CREATE TABLE scan_lifecycle_applications (
+            scan_id UUID PRIMARY KEY REFERENCES scans(scan_id),
+            applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            applied_by TEXT NOT NULL DEFAULT 'system'
+        )
+        """
+    )
+
+    op.execute(
+        """
+        CREATE TABLE finding_fingerprints (
+            id BIGSERIAL PRIMARY KEY,
+            tenant_id TEXT NOT NULL,
+            subscription_id TEXT NOT NULL,
+            resource_id_normalized TEXT NOT NULL,
+            rule_id TEXT NOT NULL,
+            evidence_key TEXT NOT NULL DEFAULT '',
+            normalization_version TEXT NOT NULL DEFAULT '1',
+            fingerprint_version TEXT NOT NULL DEFAULT '1',
+            fingerprint_hash TEXT NOT NULL,
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            CONSTRAINT uq_finding_fingerprints_hash UNIQUE (fingerprint_hash),
+            CONSTRAINT uq_finding_fingerprints_identity
+                UNIQUE (
+                    tenant_id,
+                    subscription_id,
+                    resource_id_normalized,
+                    rule_id,
+                    evidence_key,
+                    normalization_version
+                )
+        )
+        """
+    )
+
+    op.execute(
+        """
+        CREATE TABLE finding_lifecycles (
+            id BIGSERIAL PRIMARY KEY,
+            fingerprint_id BIGINT NOT NULL REFERENCES finding_fingerprints(id),
+            state TEXT NOT NULL,
+            first_seen_scan_id UUID,
+            last_seen_scan_id UUID,
+            occurrence_count INTEGER NOT NULL DEFAULT 1,
+            consecutive_success_count INTEGER NOT NULL DEFAULT 0,
+            reopen_count INTEGER NOT NULL DEFAULT 0,
+            row_version INTEGER NOT NULL DEFAULT 0,  -- monotonic update counter (FOR UPDATE handles serializability)
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            CONSTRAINT ck_finding_lifecycles_state
+                CHECK (state IN ('OPEN','RESOLVED','ACCEPTED','SUPPRESSED','REOPENED')),
+            CONSTRAINT uq_finding_lifecycles_fingerprint
+                UNIQUE (fingerprint_id)
+        )
+        """
+    )
+
+    op.execute(
+        """
+        CREATE TABLE finding_lifecycle_transitions (
+            id BIGSERIAL PRIMARY KEY,
+            lifecycle_id BIGINT NOT NULL REFERENCES finding_lifecycles(id),
+            from_state TEXT,
+            to_state TEXT NOT NULL,
+            scan_id UUID,
+            reason TEXT NOT NULL DEFAULT '',
+            transitioned_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+        )
+        """
+    )
+
+    op.execute(
+        """
+        CREATE TABLE patterns (
+            id BIGSERIAL PRIMARY KEY,
+            pattern_type TEXT NOT NULL,
+            lifecycle_id BIGINT REFERENCES finding_lifecycles(id),
+            tenant_id TEXT NOT NULL,
+            subscription_id TEXT NOT NULL,
+            scan_id UUID,
+            finding_ids JSONB NOT NULL DEFAULT '[]',
+            threshold INTEGER NOT NULL,
+            algorithm_version TEXT NOT NULL DEFAULT '1',
+            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+            CONSTRAINT ck_patterns_type
+                CHECK (pattern_type IN (
+                    'persistent_finding',
+                    'cross_resource_recurrence',
+                    'reopened_finding'
+                )),
+            CONSTRAINT uq_patterns_type_lifecycle_scan
+                UNIQUE (pattern_type, lifecycle_id, scan_id)
+        )
+        """
+    )
+
+    # Indexes for hot query paths
+    op.execute("CREATE INDEX ix_finding_fingerprints_tenant_sub ON finding_fingerprints (tenant_id, subscription_id)")
+    op.execute(
+        "CREATE INDEX ix_finding_lifecycles_state ON finding_lifecycles (state) WHERE state IN ('OPEN', 'REOPENED')"
+    )
+    op.execute(
+        "CREATE INDEX ix_finding_lifecycle_transitions_lifecycle_id ON finding_lifecycle_transitions (lifecycle_id)"
+    )
+    op.execute("CREATE INDEX ix_patterns_sub_created ON patterns (tenant_id, subscription_id, created_at DESC)")
+
+
+def downgrade() -> None:
+    op.execute("DROP TABLE IF EXISTS patterns CASCADE")
+    op.execute("DROP TABLE IF EXISTS finding_lifecycle_transitions CASCADE")
+    op.execute("DROP TABLE IF EXISTS finding_lifecycles CASCADE")
+    op.execute("DROP TABLE IF EXISTS finding_fingerprints CASCADE")
+    op.execute("DROP TABLE IF EXISTS scan_lifecycle_applications CASCADE")
+    op.execute("DROP TABLE IF EXISTS scan_rule_outcomes CASCADE")

alembic/versions/e1f2a3b4c5d6_graph_schema.py

@@ -0,0 +1,101 @@
+"""Add graph_nodes, graph_edges, and finding_graph_nodes for attack graph.
+
+Revision ID: e1f2a3b4c5d6
+Revises: a8d9c2e4f601
+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 = "e1f2a3b4c5d6"
+down_revision: str = "a8d9c2e4f601"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "graph_nodes",
+        sa.Column("node_id", postgresql.UUID(), nullable=False),
+        sa.Column("tenant_id", sa.Text(), nullable=False),
+        sa.Column("subscription_id", sa.Text(), nullable=False),
+        sa.Column("resource_id", sa.Text(), nullable=False),
+        sa.Column("resource_type", sa.Text(), nullable=False),
+        sa.Column("name", sa.Text(), nullable=False),
+        sa.Column("location", sa.Text(), nullable=False, server_default=sa.text("''")),
+        sa.Column("resource_group", sa.Text(), nullable=False, server_default=sa.text("''")),
+        sa.Column("snapshot_id", sa.Text(), nullable=False),
+        sa.Column("properties", postgresql.JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb")),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.PrimaryKeyConstraint("node_id", name="graph_nodes_pkey"),
+    )
+    op.create_index("idx_graph_nodes_resource_id", "graph_nodes", ["resource_id"])
+    op.create_index("idx_graph_nodes_tenant_subscription", "graph_nodes", ["tenant_id", "subscription_id"])
+    op.create_index(
+        "uq_graph_nodes_tenant_resource",
+        "graph_nodes",
+        ["tenant_id", "resource_id"],
+        unique=True,
+    )
+
+    op.create_table(
+        "graph_edges",
+        sa.Column("edge_id", postgresql.UUID(), nullable=False),
+        sa.Column("source_node_id", postgresql.UUID(), nullable=False),
+        sa.Column("target_node_id", postgresql.UUID(), nullable=False),
+        sa.Column("relationship_type", sa.Text(), nullable=False),
+        sa.Column("evidence_source", sa.Text(), nullable=False),
+        sa.Column("evidence_snapshot_id", sa.Text(), nullable=False),
+        sa.Column("confidence", sa.Float(), nullable=False, server_default=sa.text("1.0")),
+        sa.Column("collected_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.Column("properties", postgresql.JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb")),
+        sa.ForeignKeyConstraint(
+            ["source_node_id"], ["graph_nodes.node_id"], name="graph_edges_source_fkey", ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["target_node_id"], ["graph_nodes.node_id"], name="graph_edges_target_fkey", ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("edge_id", name="graph_edges_pkey"),
+    )
+    op.create_index("idx_graph_edges_source", "graph_edges", ["source_node_id"])
+    op.create_index("idx_graph_edges_target", "graph_edges", ["target_node_id"])
+    op.create_index(
+        "uq_graph_edges_source_target_type",
+        "graph_edges",
+        ["source_node_id", "target_node_id", "relationship_type"],
+        unique=True,
+    )
+
+    op.create_table(
+        "finding_graph_nodes",
+        sa.Column("finding_id", sa.Integer(), nullable=False),
+        sa.Column("node_id", postgresql.UUID(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["finding_id"], ["findings.id"], name="finding_graph_nodes_finding_fkey", ondelete="CASCADE"
+        ),
+        sa.ForeignKeyConstraint(
+            ["node_id"], ["graph_nodes.node_id"], name="finding_graph_nodes_node_fkey", ondelete="CASCADE"
+        ),
+        sa.PrimaryKeyConstraint("finding_id", "node_id", name="finding_graph_nodes_pkey"),
+    )
+    op.create_index("idx_finding_graph_nodes_node_id", "finding_graph_nodes", ["node_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("idx_finding_graph_nodes_node_id", table_name="finding_graph_nodes")
+    op.drop_table("finding_graph_nodes")
+
+    op.drop_index("uq_graph_edges_source_target_type", table_name="graph_edges")
+    op.drop_index("idx_graph_edges_target", table_name="graph_edges")
+    op.drop_index("idx_graph_edges_source", table_name="graph_edges")
+    op.drop_table("graph_edges")
+
+    op.drop_index("uq_graph_nodes_tenant_resource", table_name="graph_nodes")
+    op.drop_index("idx_graph_nodes_tenant_subscription", table_name="graph_nodes")
+    op.drop_index("idx_graph_nodes_resource_id", table_name="graph_nodes")
+    op.drop_table("graph_nodes")

api/app.py

@@ -260,6 +260,7 @@ def verify_jwt() -> None:
     from api.routes.compliance import compliance_bp
     from api.routes.drift import drift_bp
     from api.routes.findings import findings_bp
+    from api.routes.patterns import patterns_bp
     from api.routes.prioritization import prioritization_bp
     from api.routes.resources import resources_bp
     from api.routes.scans import scans_bp
@@ -271,6 +272,7 @@ def verify_jwt() -> None:
     app.register_blueprint(compliance_bp)
     app.register_blueprint(drift_bp)
     app.register_blueprint(findings_bp)
+    app.register_blueprint(patterns_bp)
     app.register_blueprint(prioritization_bp)
     app.register_blueprint(resources_bp)
     app.register_blueprint(scans_bp)

api/auth.py

@@ -200,9 +200,10 @@ def _verify_shared_secret(self, token: str) -> Dict[str, Any]:
         return {
             "sub": claims["sub"],
             "role": role,
-            "tenant": None,
+            "tenant": _verified_scope_claim(claims, "tid"),
             "issuer": claims.get("iss"),
             "auth_mode": SHARED_SECRET_MODE,
+            **_subscription_scope(claims),
         }
 
     def _get_jwks_client(self) -> Any:
@@ -247,7 +248,8 @@ def _verify_oidc(self, token: str) -> Dict[str, Any]:
             logger.warning("OIDC authorization rejected: %s", type(exc).__name__)
             raise TokenRejected("Invalid token") from exc
 
-        tenant = str(claims.get("tid") or "").lower() or None
+        tenant_claim = _verified_scope_claim(claims, "tid")
+        tenant = tenant_claim.lower() if tenant_claim else None
         if settings.allowed_tenants and tenant not in settings.allowed_tenants:
             logger.warning("OIDC token rejected: tenant %r is not allowed", tenant)
             raise TokenRejected("Invalid token")
@@ -266,6 +268,7 @@ def _verify_oidc(self, token: str) -> Dict[str, Any]:
             "tenant": tenant,
             "issuer": claims["iss"],
             "auth_mode": AUTH_MODE_OIDC,
+            **_subscription_scope(claims),
         }
 
 
@@ -274,3 +277,18 @@ def build_verifier(shared_secret: Callable[[], str], env: Mapping[str, str] = os
     mode = load_auth_mode(env)
     oidc = load_oidc_settings(env) if mode == AUTH_MODE_OIDC else None
     return TokenVerifier(mode, shared_secret, oidc=oidc, env=env)
+
+
+def _verified_scope_claim(claims: Dict[str, Any], key: str) -> Optional[str]:
+    """Validate scope only after signature, issuer and audience verification."""
+    value = claims.get(key)
+    if value is None:
+        return None
+    if not isinstance(value, str) or not value.strip() or len(value) > 256:
+        raise TokenRejected("Invalid token")
+    return value.strip()
+
+
+def _subscription_scope(claims: Dict[str, Any]) -> Dict[str, str]:
+    subscription = _verified_scope_claim(claims, "subscription_id")
+    return {"subscription_id": subscription} if subscription else {}