feat(automation): finding lifecycle engine, scan outcome contracts, and pattern detection [1/5]

.env.example

@@ -54,3 +54,10 @@ SENTRY_DSN=
 # Optional - enables AZ-SC-007/008 (Azure DevOps pipeline scanning)
 AZURE_DEVOPS_ORG_URL=
 AZURE_DEVOPS_PROJECT=
+
+# Optional - tenant identifier used for lifecycle tracking (defaults to subscription_id if unset).
+# Set to your Entra tenant ID (lowercase UUID), e.g. 12345678-abcd-ef01-2345-6789abcdef01.
+# If unset, patterns will be scoped by subscription_id and the /patterns endpoint will return
+# empty results for OIDC callers whose token carries a different tenant claim.
+OPENSHIELD_TENANT_ID=
+

.github/workflows/ci.yml

@@ -5,6 +5,7 @@ on:
     branches:
       - dev
       - main
+
   # 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:

.github/workflows/codeql.yml

@@ -5,6 +5,7 @@ on:
     branches:
       - dev
       - main
+
   push:
     branches:
       - dev

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/e1b3f7a9c2d5_durable_lifecycle_jobs.py

@@ -0,0 +1,89 @@
+"""Add durable, fenced lifecycle application jobs.
+
+Revision ID: e1b3f7a9c2d5
+Revises: a8d9c2e4f601
+Create Date: 2026-10-15 00:00:00.000000
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "e1b3f7a9c2d5"
+down_revision: Union[str, Sequence[str], None] = "a8d9c2e4f601"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create one resumable lifecycle job per scan, and persist rule outcomes."""
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    existing_cols = {c["name"] for c in inspector.get_columns("scans")}
+    if "rule_outcomes_json" not in existing_cols:
+        op.add_column(
+            "scans",
+            sa.Column("rule_outcomes_json", postgresql.JSONB(), nullable=True),
+        )
+    if not inspector.has_table("lifecycle_jobs"):
+        op.create_table(
+            "lifecycle_jobs",
+            sa.Column("job_id", postgresql.UUID(), nullable=False),
+            sa.Column("scan_id", postgresql.UUID(), nullable=False),
+            sa.Column("status", sa.Text(), nullable=False, server_default=sa.text("'pending'")),
+            sa.Column("lease_owner", sa.Text(), nullable=True),
+            sa.Column("lease_expires_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("last_heartbeat_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("fencing_token", sa.BigInteger(), nullable=False, server_default=sa.text("0")),
+            sa.Column("attempt_count", sa.Integer(), nullable=False, server_default=sa.text("0")),
+            sa.Column(
+                "next_retry_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.text("CURRENT_TIMESTAMP"),
+            ),
+            sa.Column("error_message", sa.Text(), nullable=True),
+            sa.Column(
+                "created_at",
+                sa.DateTime(timezone=True),
+                nullable=False,
+                server_default=sa.text("CURRENT_TIMESTAMP"),
+            ),
+            sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+            sa.ForeignKeyConstraint(["scan_id"], ["scans.scan_id"], name="lifecycle_jobs_scan_id_fkey"),
+            sa.PrimaryKeyConstraint("job_id", name="lifecycle_jobs_pkey"),
+            sa.UniqueConstraint("scan_id", name="uq_lifecycle_jobs_scan_id"),
+            sa.CheckConstraint(
+                "status IN ('pending', 'running', 'completed', 'failed')",
+                name="ck_lifecycle_jobs_status",
+            ),
+        )
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_lifecycle_jobs_pending_retry")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_lifecycle_jobs_pending_retry
+            ON lifecycle_jobs (next_retry_at ASC)
+            WHERE status = 'pending'
+            """
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_lifecycle_jobs_running_lease")
+        op.execute(
+            """
+            CREATE INDEX CONCURRENTLY idx_lifecycle_jobs_running_lease
+            ON lifecycle_jobs (lease_expires_at ASC)
+            WHERE status = 'running'
+            """
+        )
+
+
+def downgrade() -> None:
+    """Remove durable lifecycle work state."""
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_lifecycle_jobs_running_lease")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_lifecycle_jobs_pending_retry")
+    op.drop_table("lifecycle_jobs")
+    op.drop_column("scans", "rule_outcomes_json")

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") or "").lower() or None,
             "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 {}