diff --git a/alembic/versions/e1f2a3b4c5d6_finding_lifecycle.py b/alembic/versions/e1f2a3b4c5d6_finding_lifecycle.py new file mode 100644 index 00000000..c856176e --- /dev/null +++ b/alembic/versions/e1f2a3b4c5d6_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: e1f2a3b4c5d6 +Revises: d8e4f6a1b2c3 +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 = "e1f2a3b4c5d6" +down_revision: Union[str, Sequence[str], None] = "d8e4f6a1b2c3" +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") diff --git a/api/app.py b/api/app.py index a22314fa..a21f94c3 100644 --- a/api/app.py +++ b/api/app.py @@ -237,6 +237,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 @@ -248,6 +249,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) diff --git a/api/routes/patterns.py b/api/routes/patterns.py new file mode 100644 index 00000000..1c245d9c --- /dev/null +++ b/api/routes/patterns.py @@ -0,0 +1,219 @@ +"""Patterns routes: list and retrieve published security patterns.""" + +import logging +import os + +import psycopg2.extras +from flask import Blueprint, g, jsonify, request + +from api.models.finding import DatabaseManager + +patterns_bp = Blueprint("patterns", __name__) +logger = logging.getLogger(__name__) + +_ALLOWED_PATTERN_TYPES = frozenset({"persistent_finding", "cross_resource_recurrence", "reopened_finding"}) +_DEFAULT_LIMIT = 20 +_MAX_LIMIT = 100 +_MIN_LIMIT = 1 + +_VALIDATION_ERROR_MESSAGE = "Invalid request parameters" + + +class _ValidationError(ValueError): + """Raised when a client-controlled value violates the public API contract.""" + + +def _get_db() -> DatabaseManager: + if "db" not in g: + g.db = DatabaseManager(os.environ["DATABASE_URL"]) + g.db.connect() + return g.db + + +def _validate_limit(raw: str) -> int: + try: + value = int(raw) + except (ValueError, TypeError) as exc: + raise _ValidationError("limit must be an integer") from exc + if value < _MIN_LIMIT or value > _MAX_LIMIT: + raise _ValidationError(f"limit must be between {_MIN_LIMIT} and {_MAX_LIMIT}") + return value + + +def _effective_tenant(effective_sub: str) -> str: + """Return the tenant scope for this request. + + Uses OPENSHIELD_TENANT_ID env var when set (multi-tenant deployments). + Falls back to effective_sub for single-tenant deployments where + tenant_id == subscription_id by convention. + """ + return os.environ.get("OPENSHIELD_TENANT_ID", effective_sub) + + +def _effective_subscription(subscription_id_param: str | None) -> str: + """Return the subscription scope this request is authorized to see. + + The JWT subscription_id is always the authority. A query-param + subscription_id may narrow the JWT scope but never widen it. If the + JWT carries no subscription_id, the query param is accepted as the + scope (single-tenant deployments that do not embed subscription_id in + tokens). Either way the scope must be non-empty: an unscoped query + would return patterns from every subscription, which is never correct. + """ + user = getattr(g, "user", {}) or {} + jwt_sub = user.get("subscription_id") + + if jwt_sub: + if subscription_id_param and subscription_id_param != jwt_sub: + raise _ValidationError("subscription_id does not match token scope") + return jwt_sub + + if subscription_id_param: + return subscription_id_param + + raise _ValidationError("subscription_id is required") + + +def _row_to_dict(row: dict) -> dict: + """Serialise a patterns table row to a JSON-safe dict.""" + result = dict(row) + for key in ("created_at", "updated_at"): + val = result.get(key) + if val is not None and hasattr(val, "isoformat"): + result[key] = val.isoformat() + if result.get("scan_id") is not None: + result["scan_id"] = str(result["scan_id"]) + return result + + +@patterns_bp.get("/api/v1/patterns") +def list_patterns(): + """Return published security patterns, optionally filtered. + + Query parameters: + subscription_id - filter by Azure subscription (must match JWT scope) + pattern_type - one of persistent_finding, cross_resource_recurrence, + reopened_finding + limit - 1-100, default 20 + """ + try: + allowed_params = {"subscription_id", "pattern_type", "limit"} + unknown = set(request.args) - allowed_params + if unknown: + raise _ValidationError(f"Unsupported query parameter: {sorted(unknown)[0]}") + for key in request.args: + if len(request.args.getlist(key)) != 1: + raise _ValidationError(f"Query parameter {key} must be provided once") + + subscription_id_param = None + if "subscription_id" in request.args: + val = request.args["subscription_id"].strip() + if not val or len(val) > 256: + raise _ValidationError("subscription_id is invalid") + subscription_id_param = val + + pattern_type = None + if "pattern_type" in request.args: + raw_pt = request.args["pattern_type"].strip() + if raw_pt not in _ALLOWED_PATTERN_TYPES: + raise _ValidationError("Unsupported pattern_type") + pattern_type = raw_pt + + limit = _DEFAULT_LIMIT + if "limit" in request.args: + limit = _validate_limit(request.args["limit"]) + + effective_sub = _effective_subscription(subscription_id_param) + tenant_id = _effective_tenant(effective_sub) + + db = _get_db() + conn = db._get_conn() + + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """ + SELECT id, pattern_type, lifecycle_id, tenant_id, + subscription_id, scan_id, finding_ids, threshold, + algorithm_version, created_at, updated_at + FROM patterns + WHERE tenant_id = %s + AND subscription_id = %s + AND (%s IS NULL OR pattern_type = %s) + ORDER BY created_at DESC + LIMIT %s + """, + (tenant_id, effective_sub, pattern_type, pattern_type, limit), + ) + rows = cur.fetchall() + + cur.execute( + """ + SELECT COUNT(*) AS count + FROM patterns + WHERE tenant_id = %s + AND subscription_id = %s + AND (%s IS NULL OR pattern_type = %s) + """, + (tenant_id, effective_sub, pattern_type, pattern_type), + ) + total_row = cur.fetchone() + total = total_row["count"] if total_row else 0 + + return jsonify( + { + "patterns": [_row_to_dict(r) for r in rows], + "total": total, + } + ) + + except _ValidationError: + return jsonify({"error": _VALIDATION_ERROR_MESSAGE}), 400 + except Exception as exc: + logger.error("Failed to list patterns: %s", exc) + return jsonify({"error": "Failed to retrieve patterns"}), 500 + + +@patterns_bp.get("/api/v1/patterns/") +def get_pattern(pattern_id: int): + """Return a single pattern by its integer ID. + + The pattern is only returned when its subscription_id matches the + caller's authorized scope. An out-of-scope pattern returns 404 so + that the existence of other subscriptions' patterns is not disclosed. + """ + try: + if pattern_id <= 0: + raise _ValidationError("pattern_id must be a positive integer") + + # Resolve scope first so an unscoped caller cannot enumerate IDs. + effective_sub = _effective_subscription(None) + tenant_id = _effective_tenant(effective_sub) + + db = _get_db() + conn = db._get_conn() + + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """ + SELECT id, pattern_type, lifecycle_id, tenant_id, + subscription_id, scan_id, finding_ids, threshold, + algorithm_version, created_at, updated_at + FROM patterns + WHERE id = %s + AND tenant_id = %s + AND subscription_id = %s + """, + (pattern_id, tenant_id, effective_sub), + ) + row = cur.fetchone() + + if row is None: + return jsonify({"error": "Pattern not found"}), 404 + + return jsonify(_row_to_dict(row)) + + except _ValidationError: + return jsonify({"error": _VALIDATION_ERROR_MESSAGE}), 400 + except Exception as exc: + logger.error("Failed to get pattern %s: %s", pattern_id, exc) + return jsonify({"error": "Failed to retrieve pattern"}), 500 diff --git a/api/routes/scans.py b/api/routes/scans.py index 9ec2a289..2ae55b91 100644 --- a/api/routes/scans.py +++ b/api/routes/scans.py @@ -155,35 +155,6 @@ def _run_enrichment_in_background(scan_id: str, findings: list, db_url: str) -> db.close() -def _run_enrichment_in_background(scan_id: str, findings: list, db_url: str) -> None: - """Run CVE enrichment off the request thread and persist the result. - - Runs outside the Flask request/app context (it's started via - threading.Thread), so it opens its own DatabaseManager rather than - reusing flask.g. - """ - db = DatabaseManager(db_url) - try: - enriched = enrich_findings(findings) - db.update_cve_fields(enriched) - db.update_scan_enrichment_status(scan_id, "COMPLETED") - logger.info("Background CVE enrichment complete for scan %s (%d findings)", scan_id, len(enriched)) - except Exception as exc: - logger.error("Background enrichment failed for scan %s: %s", scan_id, exc) - try: - # A failed write (e.g. in update_cve_fields) can leave db.conn in - # an aborted-transaction state. Roll back first, or this status - # update itself raises InFailedSqlTransaction and gets swallowed - # below, leaving the scan stuck at ENRICHING forever. - if db.conn is not None: - db.conn.rollback() - db.update_scan_enrichment_status(scan_id, "FAILED") - except Exception as status_exc: - logger.error("Failed to record FAILED status for scan %s: %s", scan_id, status_exc) - finally: - db.close() - - @scans_bp.post("/api/scans//enrich") def enrich_scan(scan_id): """Kick off CVE enrichment for an existing scan in the background. diff --git a/api/services/lifecycle_service.py b/api/services/lifecycle_service.py new file mode 100644 index 00000000..4c9af341 --- /dev/null +++ b/api/services/lifecycle_service.py @@ -0,0 +1,371 @@ +"""LifecycleService: applies scan outcomes to finding lifecycle state machines.""" + +import hashlib +import json +import logging +from typing import Any, Dict, List, Optional + +logger = logging.getLogger(__name__) + +# Statuses that mean "we actively confirmed this rule was clean in the scan." +_RESOLVING_STATUSES = frozenset({"SUCCESS", "EMPTY_SUCCESS"}) + +# All statuses the DB constraint accepts. Any other value defaults to FAILED to +# prevent a single malformed outcome from aborting the whole lifecycle transaction. +_VALID_OUTCOME_STATUSES = frozenset( + { + "SUCCESS", + "EMPTY_SUCCESS", + "PERMISSION_DENIED", + "TIMEOUT", + "FAILED", + "NOT_APPLICABLE", + } +) + + +def _normalize_resource_id(resource_id: str) -> str: + """Return a stable, lowercased, stripped version of an ARM resource ID.""" + return resource_id.strip().lower() + + +def _compute_fingerprint_hash( + tenant_id: str, + subscription_id: str, + resource_id_normalized: str, + rule_id: str, + evidence_key: str, + normalization_version: str, +) -> str: + """Return a SHA-256 hex digest over canonical fields.""" + canonical = json.dumps( + [ + tenant_id, + subscription_id, + resource_id_normalized, + rule_id, + evidence_key, + normalization_version, + ], + separators=(",", ":"), + sort_keys=False, + ) + return hashlib.sha256(canonical.encode()).hexdigest() + + +class LifecycleService: + """Applies scan results to the finding lifecycle state machines. + + All database work for a single apply_scan call executes inside one + transaction. The idempotency sentinel (scan_lifecycle_applications) is + inserted last, so a crash before commit means the operation never happened + and can safely be retried. Any exception triggers an explicit rollback to + leave the connection in a clean state for the next operation. + """ + + def apply_scan( + self, + db_conn: Any, + scan_id: str, + subscription_id: str, + tenant_id: str, + rule_outcomes: List[Dict[str, Any]], + findings: Optional[List[Dict[str, Any]]] = None, + ) -> None: + """Apply a completed scan's findings to the lifecycle tables. + + Args: + db_conn: A psycopg2 connection (autocommit must be False). + scan_id: UUID of the completed scan. + subscription_id: Azure subscription ID. + tenant_id: Tenant identifier for isolation. + rule_outcomes: List of dicts with keys 'rule_id' and 'status'. + findings: List of finding dicts with keys 'rule_id', 'resource_id', + and optionally 'evidence_key'. Defaults to empty list. + """ + findings = findings or [] + + try: + with db_conn.cursor() as cur: + # --- Idempotency check ---------------------------------------- + cur.execute( + "SELECT scan_id FROM scan_lifecycle_applications WHERE scan_id = %s", + (scan_id,), + ) + if cur.fetchone() is not None: + logger.info("Scan %s already applied; skipping lifecycle update", scan_id) + return + + # --- Write durable per-rule outcome records ------------------- + # UNIQUE(scan_id, rule_id) prevents duplicates on double-writes. + for outcome in rule_outcomes: + rule_id_o = outcome.get("rule_id", "") + status_o = outcome.get("status", "FAILED") + if not rule_id_o: + continue + if status_o not in _VALID_OUTCOME_STATUSES: + logger.warning( + "Unknown outcome status %r for rule %s; defaulting to FAILED", + status_o, + rule_id_o, + ) + status_o = "FAILED" + cur.execute( + """ + INSERT INTO scan_rule_outcomes ( + scan_id, rule_id, status, tenant_id, subscription_id, + inventory_boundary, started_at, completed_at + ) + VALUES (%s, %s, %s, %s, %s, 'subscription', %s, NOW()) + ON CONFLICT (scan_id, rule_id) DO NOTHING + """, + ( + scan_id, + rule_id_o, + status_o, + tenant_id, + subscription_id, + outcome.get("started_at"), + ), + ) + + # Build a lookup: rule_id -> outcome status + outcome_by_rule: Dict[str, str] = { + o["rule_id"]: o["status"] for o in rule_outcomes if "rule_id" in o and "status" in o + } + + # Collect rule IDs that actively confirmed a clean result. Only + # these can trigger resolution of absent findings (fail-closed). + resolving_rule_ids = [ + rule_id for rule_id, status in outcome_by_rule.items() if status in _RESOLVING_STATUSES + ] + + # Build the set of fingerprints seen in this scan. + seen_fingerprint_keys: set = set() + fingerprints_in_scan: List[Dict[str, Any]] = [] + for finding in findings: + rule_id = finding.get("rule_id", "") + resource_id = finding.get("resource_id", "") + evidence_key = finding.get("evidence_key", "") + resource_id_normalized = _normalize_resource_id(resource_id) + normalization_version = "1" + + fp_hash = _compute_fingerprint_hash( + tenant_id, + subscription_id, + resource_id_normalized, + rule_id, + evidence_key, + normalization_version, + ) + key = (rule_id, resource_id_normalized, evidence_key) + if key not in seen_fingerprint_keys: + seen_fingerprint_keys.add(key) + fingerprints_in_scan.append( + { + "tenant_id": tenant_id, + "subscription_id": subscription_id, + "resource_id_normalized": resource_id_normalized, + "rule_id": rule_id, + "evidence_key": evidence_key, + "normalization_version": normalization_version, + "fingerprint_hash": fp_hash, + } + ) + + # --- Upsert fingerprints and lifecycles for seen findings ------- + seen_fingerprint_ids: set = set() + for fp in fingerprints_in_scan: + # ON CONFLICT DO UPDATE forces RETURNING id on pre-existing rows. + cur.execute( + """ + INSERT INTO finding_fingerprints ( + tenant_id, subscription_id, resource_id_normalized, + rule_id, evidence_key, normalization_version, + fingerprint_version, fingerprint_hash + ) + VALUES (%s, %s, %s, %s, %s, %s, '1', %s) + ON CONFLICT (fingerprint_hash) DO UPDATE + SET fingerprint_hash = EXCLUDED.fingerprint_hash + RETURNING id + """, + ( + fp["tenant_id"], + fp["subscription_id"], + fp["resource_id_normalized"], + fp["rule_id"], + fp["evidence_key"], + fp["normalization_version"], + fp["fingerprint_hash"], + ), + ) + row = cur.fetchone() + fingerprint_id = row[0] + seen_fingerprint_ids.add(fingerprint_id) + + cur.execute( + """ + SELECT id, state, occurrence_count, reopen_count, row_version + FROM finding_lifecycles + WHERE fingerprint_id = %s + FOR UPDATE + """, + (fingerprint_id,), + ) + lc_row = cur.fetchone() + + if lc_row is None: + cur.execute( + """ + INSERT INTO finding_lifecycles ( + fingerprint_id, state, first_seen_scan_id, + last_seen_scan_id, occurrence_count, + consecutive_success_count, reopen_count, row_version + ) + VALUES (%s, 'OPEN', %s, %s, 1, 0, 0, 0) + RETURNING id + """, + (fingerprint_id, scan_id, scan_id), + ) + lifecycle_id = cur.fetchone()[0] + cur.execute( + """ + INSERT INTO finding_lifecycle_transitions + (lifecycle_id, from_state, to_state, scan_id, reason) + VALUES (%s, NULL, 'OPEN', %s, 'New finding observed') + """, + (lifecycle_id, scan_id), + ) + else: + lifecycle_id, state, occurrence_count, reopen_count, row_version = lc_row + if state in ("OPEN", "REOPENED"): + cur.execute( + """ + UPDATE finding_lifecycles + SET occurrence_count = occurrence_count + 1, + last_seen_scan_id = %s, + updated_at = NOW(), + row_version = row_version + 1 + WHERE id = %s + """, + (scan_id, lifecycle_id), + ) + elif state in ("RESOLVED", "ACCEPTED", "SUPPRESSED"): + cur.execute( + """ + UPDATE finding_lifecycles + SET state = 'REOPENED', + last_seen_scan_id = %s, + occurrence_count = occurrence_count + 1, + reopen_count = reopen_count + 1, + consecutive_success_count = 0, + updated_at = NOW(), + row_version = row_version + 1 + WHERE id = %s + """, + (scan_id, lifecycle_id), + ) + cur.execute( + """ + INSERT INTO finding_lifecycle_transitions + (lifecycle_id, from_state, to_state, scan_id, reason) + VALUES (%s, %s, 'REOPENED', %s, 'Finding reappeared in scan') + """, + (lifecycle_id, state, scan_id), + ) + + # --- Resolve findings NOT seen in this scan ------------------- + # Fail-closed: skip resolution entirely if no rule confirmed a + # clean result this scan (all outcomes were FAILED/PERMISSION_DENIED). + if resolving_rule_ids: + if seen_fingerprint_ids: + # Use != ALL(%s) with a list to avoid single-element tuple + # syntax issues that occur with NOT IN %s. + cur.execute( + """ + SELECT fl.id, fl.state, fl.row_version, ff.rule_id + FROM finding_lifecycles fl + JOIN finding_fingerprints ff ON ff.id = fl.fingerprint_id + WHERE ff.tenant_id = %s + AND ff.subscription_id = %s + AND fl.state IN ('OPEN', 'REOPENED') + AND fl.fingerprint_id != ALL(%s) + AND ff.rule_id = ANY(%s) + FOR UPDATE OF fl + """, + ( + tenant_id, + subscription_id, + list(seen_fingerprint_ids), + resolving_rule_ids, + ), + ) + else: + cur.execute( + """ + SELECT fl.id, fl.state, fl.row_version, ff.rule_id + FROM finding_lifecycles fl + JOIN finding_fingerprints ff ON ff.id = fl.fingerprint_id + WHERE ff.tenant_id = %s + AND ff.subscription_id = %s + AND fl.state IN ('OPEN', 'REOPENED') + AND ff.rule_id = ANY(%s) + FOR UPDATE OF fl + """, + (tenant_id, subscription_id, resolving_rule_ids), + ) + _resolve_absent_rows(cur, scan_id, outcome_by_rule) + + # --- Idempotency sentinel (inserted last) --------------------- + cur.execute( + """ + INSERT INTO scan_lifecycle_applications (scan_id, applied_by) + VALUES (%s, 'system') + """, + (scan_id,), + ) + + db_conn.commit() + logger.info("Lifecycle application committed for scan %s", scan_id) + + except Exception: + db_conn.rollback() + logger.error("Lifecycle application rolled back for scan %s", scan_id, exc_info=True) + raise + + +def _resolve_absent_rows( + cur: Any, + scan_id: str, + outcome_by_rule: Dict[str, str], +) -> None: + """Transition OPEN/REOPENED lifecycle rows to RESOLVED for clean-outcome rules. + + The SQL query already filters by resolving_rule_ids, so outcome_status is + always in _RESOLVING_STATUSES here. The check is kept as a defensive guard. + """ + absent_rows = cur.fetchall() + for lc_id, state, row_version, rule_id in absent_rows: + outcome_status = outcome_by_rule.get(rule_id) + if outcome_status in _RESOLVING_STATUSES: + cur.execute( + """ + UPDATE finding_lifecycles + SET state = 'RESOLVED', + last_seen_scan_id = %s, + consecutive_success_count = consecutive_success_count + 1, + updated_at = NOW(), + row_version = row_version + 1 + WHERE id = %s + """, + (scan_id, lc_id), + ) + cur.execute( + """ + INSERT INTO finding_lifecycle_transitions + (lifecycle_id, from_state, to_state, scan_id, reason) + VALUES (%s, %s, 'RESOLVED', %s, + 'Rule confirmed clean; finding absent from scan') + """, + (lc_id, state, scan_id), + ) diff --git a/api/services/pattern_service.py b/api/services/pattern_service.py new file mode 100644 index 00000000..0a8eb5f5 --- /dev/null +++ b/api/services/pattern_service.py @@ -0,0 +1,171 @@ +"""PatternService: detects and publishes security patterns from lifecycle state.""" + +import json +import logging +from typing import Any + +import psycopg2.extras + +logger = logging.getLogger(__name__) + +# Hard-coded thresholds stored alongside the pattern record for traceability. +_PERSISTENT_THRESHOLD = 3 +_CROSS_RESOURCE_THRESHOLD = 2 +_REOPENED_THRESHOLD = 1 +_ALGORITHM_VERSION = "1" + + +class PatternService: + """Detects recurring patterns in finding lifecycle data and publishes records.""" + + def detect_and_publish( + self, + db_conn: Any, + scan_id: str, + subscription_id: str, + tenant_id: str, + ) -> int: + """Detect patterns for the given scan and upsert them into the patterns table. + + Returns the number of patterns upserted. + """ + count = 0 + + try: + with db_conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + # 1. persistent_finding: occurrence_count >= threshold and OPEN/REOPENED + cur.execute( + """ + SELECT fl.id AS lifecycle_id + FROM finding_lifecycles fl + JOIN finding_fingerprints ff ON ff.id = fl.fingerprint_id + WHERE ff.tenant_id = %s + AND ff.subscription_id = %s + AND fl.state IN ('OPEN', 'REOPENED') + AND fl.occurrence_count >= %s + """, + (tenant_id, subscription_id, _PERSISTENT_THRESHOLD), + ) + persistent_rows = cur.fetchall() + + # 2. cross_resource_recurrence: same rule_id >= 2 OPEN/REOPENED lifecycles + cur.execute( + """ + SELECT ff.rule_id, + array_agg(fl.id ORDER BY fl.id) AS lifecycle_ids, + COUNT(*) AS lc_count + FROM finding_lifecycles fl + JOIN finding_fingerprints ff ON ff.id = fl.fingerprint_id + WHERE ff.tenant_id = %s + AND ff.subscription_id = %s + AND fl.state IN ('OPEN', 'REOPENED') + GROUP BY ff.rule_id + HAVING COUNT(*) >= %s + """, + (tenant_id, subscription_id, _CROSS_RESOURCE_THRESHOLD), + ) + cross_rows = cur.fetchall() + + # 3. reopened_finding: reopen_count >= 1 and state == REOPENED + cur.execute( + """ + SELECT fl.id AS lifecycle_id + FROM finding_lifecycles fl + JOIN finding_fingerprints ff ON ff.id = fl.fingerprint_id + WHERE ff.tenant_id = %s + AND ff.subscription_id = %s + AND fl.state = 'REOPENED' + AND fl.reopen_count >= %s + """, + (tenant_id, subscription_id, _REOPENED_THRESHOLD), + ) + reopened_rows = cur.fetchall() + + # All detection queries finished; cursor closed. Upsert via fresh cursors + # to avoid open-cursor overlap inside _upsert_pattern. + for row in persistent_rows: + _upsert_pattern( + db_conn, + pattern_type="persistent_finding", + lifecycle_id=row["lifecycle_id"], + tenant_id=tenant_id, + subscription_id=subscription_id, + scan_id=scan_id, + finding_ids=[], + threshold=_PERSISTENT_THRESHOLD, + ) + count += 1 + + for row in cross_rows: + lifecycle_ids = row["lifecycle_ids"] + # Publish one pattern per lifecycle so each is individually traceable. + for lc_id in lifecycle_ids: + sibling_ids = [lid for lid in lifecycle_ids if lid != lc_id] + _upsert_pattern( + db_conn, + pattern_type="cross_resource_recurrence", + lifecycle_id=lc_id, + tenant_id=tenant_id, + subscription_id=subscription_id, + scan_id=scan_id, + finding_ids=sibling_ids, + threshold=_CROSS_RESOURCE_THRESHOLD, + ) + count += 1 + + for row in reopened_rows: + _upsert_pattern( + db_conn, + pattern_type="reopened_finding", + lifecycle_id=row["lifecycle_id"], + tenant_id=tenant_id, + subscription_id=subscription_id, + scan_id=scan_id, + finding_ids=[], + threshold=_REOPENED_THRESHOLD, + ) + count += 1 + + db_conn.commit() + logger.info("Pattern detection for scan %s: %d pattern(s) upserted", scan_id, count) + return count + + except Exception: + db_conn.rollback() + logger.error("Pattern detection rolled back for scan %s", scan_id, exc_info=True) + raise + + +def _upsert_pattern( + db_conn: Any, + pattern_type: str, + lifecycle_id: int, + tenant_id: str, + subscription_id: str, + scan_id: str, + finding_ids: list, + threshold: int, +) -> None: + """Insert a pattern record if the (type, lifecycle, scan) combination is new.""" + with db_conn.cursor() as cur: + cur.execute( + """ + INSERT INTO patterns ( + pattern_type, lifecycle_id, tenant_id, subscription_id, + scan_id, finding_ids, threshold, algorithm_version, + created_at, updated_at + ) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW()) + ON CONFLICT ON CONSTRAINT uq_patterns_type_lifecycle_scan DO NOTHING + """, + ( + pattern_type, + lifecycle_id, + tenant_id, + subscription_id, + scan_id, + json.dumps(finding_ids), + threshold, + _ALGORITHM_VERSION, + ), + ) diff --git a/scanner/engine.py b/scanner/engine.py index f7af9aa2..68fff16b 100644 --- a/scanner/engine.py +++ b/scanner/engine.py @@ -9,6 +9,14 @@ from api.observability import RULE_ERRORS_TOTAL from openshield.severity import CONTRACT_VERSION, SeverityContractError, normalize_severity, score_findings + +try: + import azure.core.exceptions as _azure_exc + + _AzureHttpResponseError = _azure_exc.HttpResponseError +except Exception: + _AzureHttpResponseError = None # type: ignore[assignment,misc] + from scanner.azure_client import AzureClient logger = logging.getLogger(__name__) @@ -103,15 +111,16 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: Returns: dict with keys: scan_id, subscription_id, started_at, - completed_at, total_findings, findings. + completed_at, total_findings, findings, rule_outcomes. """ scan_id = scan_id or str(uuid.uuid4()) started_at = datetime.now(timezone.utc).isoformat() findings: List[Dict[str, Any]] = [] + rule_outcomes: List[Dict[str, Any]] = [] detected_at = datetime.now(timezone.utc).isoformat() logger.info( - "Scan %s starting against subscription %s — %d rules loaded", + "Scan %s starting against subscription %s - %d rules loaded", scan_id, self.subscription_id, len(self.rules), @@ -119,32 +128,62 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: for rule in self.rules: rule_id = getattr(rule, "RULE_ID", "UNKNOWN") + rule_started = datetime.now(timezone.utc).isoformat() + outcome_status = "FAILED" try: rule_findings = rule.scan(self.client, self.subscription_id) if not isinstance(rule_findings, list): - logger.warning("Rule %s returned %s instead of list — skipped", rule_id, type(rule_findings)) - continue - - validated_findings = [] - for raw_finding in rule_findings: - finding = raw_finding - if not isinstance(finding, dict): - logger.warning("Rule %s returned a non-object finding — skipped", rule_id) - continue - finding = dict(finding) - finding["severity"] = normalize_severity(finding.get("severity")) - finding.setdefault("detected_at", detected_at) - finding.setdefault("scan_id", scan_id) - validated_findings.append(finding) - findings.extend(validated_findings) - logger.info("Rule %s produced %d finding(s)", rule_id, len(validated_findings)) + logger.warning("Rule %s returned %s instead of list - skipped", rule_id, type(rule_findings)) + outcome_status = "FAILED" + else: + validated_findings = [] + for raw_finding in rule_findings: + finding = raw_finding + if not isinstance(finding, dict): + logger.warning("Rule %s returned a non-object finding - skipped", rule_id) + continue + finding = dict(finding) + finding["severity"] = normalize_severity(finding.get("severity")) + finding.setdefault("detected_at", detected_at) + finding.setdefault("scan_id", scan_id) + validated_findings.append(finding) + findings.extend(validated_findings) + outcome_status = "SUCCESS" if validated_findings else "EMPTY_SUCCESS" + logger.info("Rule %s produced %d finding(s)", rule_id, len(validated_findings)) except SeverityContractError: RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() logger.exception("Rule %s returned an invalid severity", rule_id) raise + except PermissionError: + RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() + logger.error("Rule %s raised PermissionError", rule_id) + outcome_status = "PERMISSION_DENIED" + except TimeoutError: + RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() + logger.error("Rule %s timed out", rule_id) + outcome_status = "TIMEOUT" except Exception as exc: RULE_ERRORS_TOTAL.labels(rule_id=rule_id).inc() - logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True) + if _AzureHttpResponseError is not None and isinstance(exc, _AzureHttpResponseError): + status_code = getattr(exc, "status_code", None) + if status_code == 403: + outcome_status = "PERMISSION_DENIED" + logger.error("Rule %s got HTTP 403 from Azure", rule_id) + else: + outcome_status = "FAILED" + logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True) + else: + outcome_status = "FAILED" + logger.error("Rule %s raised an exception: %s", rule_id, exc, exc_info=True) + + rule_outcomes.append( + { + "rule_id": rule_id, + "status": outcome_status, + "started_at": rule_started, + "completed_at": datetime.now(timezone.utc).isoformat(), + } + ) completed_at = datetime.now(timezone.utc).isoformat() @@ -161,8 +200,9 @@ def run_scan(self, scan_id: Optional[str] = None) -> Dict[str, Any]: "score": score, "severity_contract_version": CONTRACT_VERSION, "findings": findings, + "rule_outcomes": rule_outcomes, } - logger.info("Scan %s complete — %d total finding(s). Normalising results...", scan_id, len(findings)) + logger.info("Scan %s complete - %d total finding(s). Normalising results...", scan_id, len(findings)) return make_serializable(result) diff --git a/scanner/worker.py b/scanner/worker.py index 21a94c6f..f89c593b 100644 --- a/scanner/worker.py +++ b/scanner/worker.py @@ -19,6 +19,8 @@ configure_logging, init_sentry, ) +from api.services.lifecycle_service import LifecycleService +from api.services.pattern_service import PatternService from scanner.engine import ScanEngine configure_logging() @@ -74,6 +76,36 @@ def run_worker(): result["status"] = "completed" db.save_scan(result) + + # Apply lifecycle tracking. Lifecycle failures are non-fatal: + # the scan is already persisted, so we log and continue rather + # than marking the scan as failed. + try: + tenant_id = os.environ.get("OPENSHIELD_TENANT_ID", subscription_id) + lc_svc = LifecycleService() + lc_svc.apply_scan( + db_conn=db._get_conn(), + scan_id=scan_id, + subscription_id=subscription_id, + tenant_id=tenant_id, + rule_outcomes=result.get("rule_outcomes", []), + findings=result.get("findings", []), + ) + pat_svc = PatternService() + pat_svc.detect_and_publish( + db_conn=db._get_conn(), + scan_id=scan_id, + subscription_id=subscription_id, + tenant_id=tenant_id, + ) + except Exception as lc_exc: + logger.error( + "Lifecycle/pattern update failed for scan %s (scan data intact): %s", + scan_id, + lc_exc, + extra={"scan_id": scan_id}, + ) + SCANS_TOTAL.labels(status="completed").inc() logger.info( "Successfully completed scan %s", diff --git a/tests/test_finding_lifecycle.py b/tests/test_finding_lifecycle.py new file mode 100644 index 00000000..23307410 --- /dev/null +++ b/tests/test_finding_lifecycle.py @@ -0,0 +1,387 @@ +"""Tests for LifecycleService using a mocked psycopg2 connection. + +All tests use in-memory state to simulate the database without requiring a +live PostgreSQL instance. +""" + +from collections import deque + +from api.services.lifecycle_service import ( + LifecycleService, + _compute_fingerprint_hash, + _normalize_resource_id, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +TENANT_ID = "tenant-abc" +SUB_ID = "sub-001" +SCAN_ID_1 = "11111111-1111-1111-1111-111111111111" +SCAN_ID_2 = "22222222-2222-2222-2222-222222222222" + + +def _make_finding(rule_id: str, resource_id: str, evidence_key: str = "") -> dict: + return { + "rule_id": rule_id, + "resource_id": resource_id, + "evidence_key": evidence_key, + } + + +def _make_outcome(rule_id: str, status: str) -> dict: + return {"rule_id": rule_id, "status": status} + + +# --------------------------------------------------------------------------- +# Fake cursor / connection. +# +# The cursor consumes from a shared deque so multiple cursor() calls on the +# same connection share one result stream. This mirrors how psycopg2 works +# (multiple cursors on one connection see the same transaction state) and +# avoids the "second cursor re-reads from the start" bug noted in code review. +# --------------------------------------------------------------------------- + + +class _FakeCursor: + """Fake psycopg2 cursor backed by a shared result deque.""" + + def __init__(self, results_deque: deque): + self._results = results_deque + self.executed: list = [] + self._current = None + + def execute(self, sql, params=None): + self.executed.append((sql.strip(), params)) + self._current = self._results.popleft() if self._results else None + + def fetchone(self): + return self._current + + def fetchall(self): + if isinstance(self._current, list): + return self._current + return [] if self._current is None else [self._current] + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +class _FakeConn: + """Fake connection whose cursor() calls share one result deque.""" + + def __init__(self, results: list): + self._deque: deque = deque(results) + self.committed = False + self._cursor_obj: _FakeCursor | None = None + + def cursor(self, **_kwargs): + # Return a new cursor object but backed by the same shared deque. + self._cursor_obj = _FakeCursor(self._deque) + return self._cursor_obj + + def commit(self): + self.committed = True + + def rollback(self): + pass + + def all_executed(self) -> list: + return self._cursor_obj.executed if self._cursor_obj else [] + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestComputeFingerprintHash: + def test_hash_is_stable(self): + h1 = _compute_fingerprint_hash(TENANT_ID, SUB_ID, "/sub/001/rg/foo", "RULE-001", "", "1") + h2 = _compute_fingerprint_hash(TENANT_ID, SUB_ID, "/sub/001/rg/foo", "RULE-001", "", "1") + assert h1 == h2 + + def test_different_tenant_different_hash(self): + h1 = _compute_fingerprint_hash("tenant-A", SUB_ID, "/rg/foo", "RULE-001", "", "1") + h2 = _compute_fingerprint_hash("tenant-B", SUB_ID, "/rg/foo", "RULE-001", "", "1") + assert h1 != h2 + + def test_tenant_id_included_in_hash(self): + """Tenant isolation: fingerprint_hash must encode the tenant_id.""" + h_a = _compute_fingerprint_hash("tenant-A", SUB_ID, "/rg/foo", "RULE-001", "", "1") + h_b = _compute_fingerprint_hash("tenant-B", SUB_ID, "/rg/foo", "RULE-001", "", "1") + assert h_a != h_b + + def test_different_rule_different_hash(self): + h1 = _compute_fingerprint_hash(TENANT_ID, SUB_ID, "/rg/foo", "RULE-001", "", "1") + h2 = _compute_fingerprint_hash(TENANT_ID, SUB_ID, "/rg/foo", "RULE-002", "", "1") + assert h1 != h2 + + +class TestNormalizeResourceId: + def test_lowercase(self): + assert _normalize_resource_id("/subscriptions/SUB/resourceGroups/RG") == ( + "/subscriptions/sub/resourcegroups/rg" + ) + + def test_strip_whitespace(self): + assert _normalize_resource_id(" /rg/foo ") == "/rg/foo" + + +class TestLifecycleServiceIdempotency: + """Applying the same scan_id twice must commit exactly once.""" + + def test_second_apply_is_no_op(self): + committed_count = 0 + + class _TrackingConn: + def __init__(self, already_applied: bool): + if already_applied: + # idempotency check returns a row -> return immediately + results = [("already-applied",)] + else: + # idempotency check None, outcome insert None, + # fingerprint upsert -> (1,), lifecycle lock -> None (new), + # lifecycle insert -> (10,), transition insert None, + # absent-findings query -> [], idempotency insert None + results = [None, None, (1,), None, (10,), None, [], None] + self._deque: deque = deque(results) + self._cursor_obj = None + + def cursor(self, **_kwargs): + self._cursor_obj = _FakeCursor(self._deque) + return self._cursor_obj + + def commit(self): + nonlocal committed_count + committed_count += 1 + + def rollback(self): + pass + + svc = LifecycleService() + svc.apply_scan( + _TrackingConn(False), + SCAN_ID_1, + SUB_ID, + TENANT_ID, + [_make_outcome("RULE-001", "SUCCESS")], + [_make_finding("RULE-001", "/rg/foo")], + ) + svc.apply_scan( + _TrackingConn(True), + SCAN_ID_1, + SUB_ID, + TENANT_ID, + [_make_outcome("RULE-001", "SUCCESS")], + [_make_finding("RULE-001", "/rg/foo")], + ) + + # Only the first call should have committed. + assert committed_count == 1 + + +class TestLifecycleStateTransitions: + """State machine correctness tests using scripted cursor results.""" + + def _run(self, results, findings, outcomes): + conn = _FakeConn(results) + svc = LifecycleService() + svc.apply_scan(conn, SCAN_ID_1, SUB_ID, TENANT_ID, outcomes, findings) + return conn + + def _all_sql(self, conn: _FakeConn) -> list[str]: + return [sql for sql, _ in conn.all_executed()] + + def test_new_finding_creates_open_lifecycle(self): + # Sequence (one shared deque, all execute() calls in order): + # 1. idempotency check -> None + # 2. scan_rule_outcomes insert (for RULE-001 outcome) -> None + # 3. fingerprint upsert RETURNING id -> (1,) + # 4. lifecycle FOR UPDATE -> None (new) + # 5. lifecycle INSERT RETURNING id -> (10,) + # 6. transition insert -> None + # 7. absent-findings query (no resolving ids after seen_fingerprint_ids) -> [] + # 8. idempotency insert -> None + results = [None, None, (1,), None, (10,), None, [], None] + conn = self._run( + results, + [_make_finding("RULE-001", "/rg/foo")], + [_make_outcome("RULE-001", "SUCCESS")], + ) + assert conn.committed + sqls = self._all_sql(conn) + assert any("INSERT INTO finding_lifecycles" in s for s in sqls) + # Transition to OPEN must be recorded. + assert any("finding_lifecycle_transitions" in s and "'OPEN'" in s for s in sqls) + + def test_scan_rule_outcomes_written(self): + """scan_rule_outcomes must be populated on every apply_scan call.""" + results = [None, None, (1,), None, (10,), None, [], None] + conn = self._run( + results, + [_make_finding("RULE-001", "/rg/foo")], + [_make_outcome("RULE-001", "SUCCESS")], + ) + sqls = self._all_sql(conn) + assert any("INSERT INTO scan_rule_outcomes" in s for s in sqls) + + def test_open_finding_not_seen_in_success_scan_is_resolved(self): + # No findings; SUCCESS outcome -> existing OPEN resolved. + # 1. idempotency check -> None + # 2. scan_rule_outcomes insert -> None + # 3. absent-findings query (empty seen_ids branch) -> [(10,'OPEN',0,'RULE-001')] + # 4. UPDATE to RESOLVED -> None + # 5. transition insert (OPEN -> RESOLVED) -> None + # 6. idempotency insert -> None + results = [ + None, + None, + [(10, "OPEN", 0, "RULE-001")], + None, + None, + None, + ] + conn = self._run(results, [], [_make_outcome("RULE-001", "SUCCESS")]) + assert conn.committed + sqls = self._all_sql(conn) + assert any("RESOLVED" in s and "UPDATE" in s.upper() for s in sqls) + + def test_open_finding_not_seen_in_failed_scan_stays_open(self): + # FAILED outcome -> fail-closed: no resolution. + # 1. idempotency check -> None + # 2. scan_rule_outcomes insert -> None + # (FAILED is not in resolving_rule_ids; no absent-findings query is issued) + # 3. idempotency insert -> None + results = [None, None, None] + conn = self._run(results, [], [_make_outcome("RULE-001", "FAILED")]) + assert conn.committed + sqls = self._all_sql(conn) + assert not any("RESOLVED" in s and "UPDATE" in s.upper() for s in sqls) + + def test_open_finding_not_seen_in_permission_denied_scan_stays_open(self): + results = [None, None, None] + conn = self._run(results, [], [_make_outcome("RULE-001", "PERMISSION_DENIED")]) + assert conn.committed + sqls = self._all_sql(conn) + assert not any("RESOLVED" in s and "UPDATE" in s.upper() for s in sqls) + + def test_resolved_finding_seen_again_becomes_reopened(self): + # Fingerprint exists (id=1), lifecycle row has state=RESOLVED -> REOPEN. + # 1. idempotency check -> None + # 2. scan_rule_outcomes insert -> None + # 3. fingerprint upsert RETURNING id -> (1,) + # 4. lifecycle FOR UPDATE -> (10,'RESOLVED',1,0,2) + # 5. UPDATE to REOPENED (resets consecutive_success_count=0) -> None + # 6. transition insert -> None + # 7. absent-findings query -> [] + # 8. idempotency insert -> None + results = [None, None, (1,), (10, "RESOLVED", 1, 0, 2), None, None, [], None] + conn = self._run( + results, + [_make_finding("RULE-001", "/rg/foo")], + [_make_outcome("RULE-001", "SUCCESS")], + ) + assert conn.committed + sqls = self._all_sql(conn) + assert any("REOPENED" in s and "UPDATE" in s.upper() for s in sqls) + # consecutive_success_count must be reset to 0 on reopen. + assert any("consecutive_success_count = 0" in s for s in sqls) + + def test_rule_a_success_does_not_resolve_rule_b_finding(self): + """Rule-A SUCCESS must not resolve a finding belonging to Rule-B. + + This is the most critical correctness invariant: a clean outcome for one + rule can only affect findings that were produced by that same rule. + """ + # Scan has no findings (empty list), two outcomes: + # RULE-001 SUCCESS -> in resolving_rule_ids + # RULE-002 FAILED -> NOT in resolving_rule_ids + # + # Sequence (empty seen_ids branch): + # 1. idempotency check -> None + # 2. scan_rule_outcomes insert RULE-001 -> None + # 3. scan_rule_outcomes insert RULE-002 -> None + # 4. absent-findings query (ff.rule_id = ANY(['RULE-001'])) -> + # returns only RULE-001's lifecycle row; RULE-002 is excluded by SQL + # 5. UPDATE RESOLVED for RULE-001 row -> None + # 6. transition insert -> None + # 7. idempotency insert -> None + results = [ + None, # idempotency check + None, # scan_rule_outcomes RULE-001 + None, # scan_rule_outcomes RULE-002 + [(10, "OPEN", 0, "RULE-001")], # absent-findings query + None, # UPDATE RESOLVED + None, # transition insert + None, # idempotency insert + ] + conn = self._run( + results, + [], + [_make_outcome("RULE-001", "SUCCESS"), _make_outcome("RULE-002", "FAILED")], + ) + assert conn.committed + # Verify resolving_rule_ids in the SQL params contains only RULE-001. + absent_query = next( + (item for item in conn.all_executed() if "ff.rule_id = ANY" in item[0]), + None, + ) + assert absent_query is not None, "absent-findings query was not issued" + _, params = absent_query + resolving_ids = params[-1] # last param is resolving_rule_ids list + assert "RULE-001" in resolving_ids + assert "RULE-002" not in resolving_ids + + def test_reopened_finding_absent_from_success_scan_is_resolved(self): + # REOPENED finding not seen in a SUCCESS scan must transition to RESOLVED. + # 1. idempotency -> None + # 2. scan_rule_outcomes insert -> None + # 3. absent-findings query (empty seen_ids branch) -> [(10,'REOPENED',0,'RULE-001')] + # 4. UPDATE to RESOLVED -> None + # 5. transition insert (REOPENED -> RESOLVED) -> None + # 6. idempotency insert -> None + results = [ + None, + None, + [(10, "REOPENED", 0, "RULE-001")], + None, + None, + None, + ] + conn = self._run(results, [], [_make_outcome("RULE-001", "SUCCESS")]) + assert conn.committed + sqls = self._all_sql(conn) + assert any("RESOLVED" in s and "UPDATE" in s.upper() for s in sqls) + # Transition from REOPENED to RESOLVED must be recorded. + assert any("finding_lifecycle_transitions" in s and "REOPENED" in s and "RESOLVED" in s for s in sqls) + + def test_reopened_finding_seen_again_increments_occurrence_stays_reopened(self): + # REOPENED + seen in scan: occurrence_count increments, no new state transition. + # 1. idempotency -> None + # 2. scan_rule_outcomes insert -> None + # 3. fingerprint upsert -> (1,) + # 4. lifecycle FOR UPDATE -> (10,'REOPENED',3,1,4) + # 5. UPDATE occurrence_count + 1 -> None + # 6. absent-findings query -> [] + # 7. idempotency insert -> None + results = [None, None, (1,), (10, "REOPENED", 3, 1, 4), None, [], None] + conn = self._run( + results, + [_make_finding("RULE-001", "/rg/foo")], + [_make_outcome("RULE-001", "SUCCESS")], + ) + assert conn.committed + sqls = self._all_sql(conn) + + # No transition record should be emitted for REOPENED->REOPENED. + transition_to_reopened = [s for s in sqls if "finding_lifecycle_transitions" in s and "REOPENED" in s] + assert not transition_to_reopened + + # occurrence_count should increment. + assert any("occurrence_count = occurrence_count + 1" in s for s in sqls) diff --git a/tests/test_patterns.py b/tests/test_patterns.py new file mode 100644 index 00000000..3d529977 --- /dev/null +++ b/tests/test_patterns.py @@ -0,0 +1,383 @@ +"""Tests for PatternService and GET /api/v1/patterns routes. + +Pattern detection tests use a mocked DB; route tests use the Flask test client +with mocked database queries. +""" + +import secrets +import time +from collections import deque +from unittest.mock import MagicMock, patch + +import jwt +import pytest + +TENANT_ID = "tenant-abc" +SUB_ID = "sub-001" +SCAN_ID = "33333333-3333-3333-3333-333333333333" + +_TEST_JWT_SECRET = secrets.token_urlsafe(32) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_token(sub_id: str | None = None) -> str: + payload = { + "sub": "test-user", + "role": "admin", + "iat": int(time.time()), + "exp": int(time.time()) + 3600, + } + if sub_id: + payload["subscription_id"] = sub_id + return jwt.encode(payload, _TEST_JWT_SECRET, algorithm="HS256") + + +def _auth_headers(sub_id: str | None = None) -> dict: + return { + "Authorization": f"Bearer {_make_token(sub_id)}", + "Content-Type": "application/json", + } + + +# --------------------------------------------------------------------------- +# Fake cursor / connection for service tests +# --------------------------------------------------------------------------- + + +class _FakeCursor: + """Fake cursor backed by a shared deque so all cursors on one connection share state.""" + + def __init__(self, results_deque: deque): + self._results = results_deque + self.executed: list = [] + self._current = None + + def execute(self, sql, params=None): + self.executed.append((sql.strip(), params)) + self._current = self._results.popleft() if self._results else None + + def fetchone(self): + return self._current + + def fetchall(self): + if isinstance(self._current, list): + return self._current + return [] if self._current is None else [self._current] + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +class _FakeConn: + """Fake connection whose cursor() calls share one result deque.""" + + def __init__(self, fetchall_pages: list): + self._deque: deque = deque(fetchall_pages) + self.committed = False + self._cursors: list = [] + self._cursor_obj = None + + def cursor(self, **_kwargs): + self._cursor_obj = _FakeCursor(self._deque) + self._cursors.append(self._cursor_obj) + return self._cursor_obj + + def commit(self): + self.committed = True + + def rollback(self): + pass + + def all_executed(self) -> list: + result = [] + for c in self._cursors: + result.extend(c.executed) + return result + + +# --------------------------------------------------------------------------- +# PatternService unit tests +# --------------------------------------------------------------------------- + + +class TestPatternServiceDetection: + def _run(self, pages, extra_pages=None): + """Run detect_and_publish with scripted DB results.""" + from api.services.pattern_service import PatternService + + all_pages = list(pages) + if extra_pages: + all_pages.extend(extra_pages) + conn = _FakeConn(all_pages) + svc = PatternService() + count = svc.detect_and_publish(conn, SCAN_ID, SUB_ID, TENANT_ID) + return count, conn + + def test_persistent_finding_detected_when_occurrence_ge_3(self): + # persistent_finding query returns 1 row -> 1 pattern upserted. + # cross_resource query returns [] -> 0. + # reopened query returns [] -> 0. + count, conn = self._run( + [ + [{"lifecycle_id": 10}], # persistent_finding + [], # cross_resource_recurrence + [], # reopened_finding + ] + ) + assert count == 1 + + def test_persistent_finding_not_detected_when_occurrence_lt_3(self): + # All queries return empty. + count, conn = self._run([[], [], []]) + assert count == 0 + + def test_cross_resource_recurrence_detected_when_2_open_lifecycles(self): + # cross_resource query returns 1 group with 2 lifecycle_ids -> 2 patterns. + count, conn = self._run( + [ + [], # persistent_finding + [{"rule_id": "RULE-001", "lifecycle_ids": [10, 11], "lc_count": 2}], + [], # reopened_finding + ] + ) + assert count == 2 + + def test_reopened_finding_detected_when_reopen_count_ge_1(self): + count, conn = self._run( + [ + [], # persistent_finding + [], # cross_resource_recurrence + [{"lifecycle_id": 20}], # reopened_finding + ] + ) + assert count == 1 + + def test_pattern_response_includes_threshold_and_algorithm_version(self): + """The upsert call must include threshold and algorithm_version.""" + from api.services.pattern_service import PatternService, _ALGORITHM_VERSION, _PERSISTENT_THRESHOLD + + conn = _FakeConn( + [ + [{"lifecycle_id": 10}], # persistent_finding + [], # cross_resource_recurrence + [], # reopened_finding + ] + ) + svc = PatternService() + svc.detect_and_publish(conn, SCAN_ID, SUB_ID, TENANT_ID) + + # Find the INSERT INTO patterns call and verify the params. + executed = conn._cursor_obj.executed + insert_sql, params = next(((sql, p) for sql, p in executed if "INSERT INTO patterns" in sql), (None, None)) + assert insert_sql is not None + # params order: pattern_type, lifecycle_id, tenant_id, subscription_id, + # scan_id, finding_ids, threshold, algorithm_version + assert params[6] == _PERSISTENT_THRESHOLD + assert params[7] == _ALGORITHM_VERSION + + +# --------------------------------------------------------------------------- +# Flask route tests for GET /api/v1/patterns +# --------------------------------------------------------------------------- + + +@pytest.fixture +def app_client(monkeypatch): + """Flask test client with JWT and mocked DB.""" + monkeypatch.setenv("DATABASE_URL", "postgresql://fake/fake") + monkeypatch.setenv("JWT_SECRET", _TEST_JWT_SECRET) + + from api.app import create_app + + application = create_app() + application.config["TESTING"] = True + application.config["JWT_SECRET"] = _TEST_JWT_SECRET + return application.test_client() + + +def _mock_db_rows(rows: list, total: int): + """Build a mock DatabaseManager whose conn returns scripted rows.""" + mock_conn = MagicMock() + mock_cursor = MagicMock() + mock_cursor.__enter__ = MagicMock(return_value=mock_cursor) + mock_cursor.__exit__ = MagicMock(return_value=False) + + # First fetchall -> rows; second fetchone -> {"count": total} + mock_cursor.fetchall.return_value = rows + mock_cursor.fetchone.return_value = {"count": total} + + mock_conn.cursor.return_value = mock_cursor + + mock_db = MagicMock() + mock_db._get_conn.return_value = mock_conn + return mock_db + + +def _sample_pattern_row(sub_id: str = SUB_ID) -> dict: + from datetime import datetime, timezone + + return { + "id": 1, + "pattern_type": "persistent_finding", + "lifecycle_id": 10, + "tenant_id": TENANT_ID, + "subscription_id": sub_id, + "scan_id": "33333333-3333-3333-3333-333333333333", + "finding_ids": [], + "threshold": 3, + "algorithm_version": "1", + "created_at": datetime(2026, 1, 1, tzinfo=timezone.utc), + "updated_at": datetime(2026, 1, 1, tzinfo=timezone.utc), + } + + +class TestPatternsRouteList: + def test_list_returns_patterns_and_total(self, app_client, monkeypatch): + row = _sample_pattern_row() + mock_db = _mock_db_rows([row], 1) + + with patch("api.routes.patterns._get_db", return_value=mock_db): + resp = app_client.get( + "/api/v1/patterns", + headers=_auth_headers(sub_id=SUB_ID), + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert "patterns" in data + assert "total" in data + assert data["total"] == 1 + assert len(data["patterns"]) == 1 + + def test_list_invalid_limit_returns_400(self, app_client): + resp = app_client.get( + "/api/v1/patterns?limit=999", + headers=_auth_headers(sub_id=SUB_ID), + ) + assert resp.status_code == 400 + + def test_list_invalid_pattern_type_returns_400(self, app_client): + resp = app_client.get( + "/api/v1/patterns?pattern_type=not_a_real_type", + headers=_auth_headers(sub_id=SUB_ID), + ) + assert resp.status_code == 400 + + def test_list_unknown_query_param_returns_400(self, app_client): + resp = app_client.get( + "/api/v1/patterns?foo=bar", + headers=_auth_headers(sub_id=SUB_ID), + ) + assert resp.status_code == 400 + + def test_list_without_subscription_returns_400(self, app_client): + # JWT with no subscription_id and no query param must be rejected. + resp = app_client.get( + "/api/v1/patterns", + headers=_auth_headers(), + ) + assert resp.status_code == 400 + + def test_list_cross_subscription_query_param_rejected(self, app_client): + # JWT scoped to sub-A must reject a query param asking for sub-B. + resp = app_client.get( + "/api/v1/patterns?subscription_id=sub-B", + headers=_auth_headers(sub_id="sub-A"), + ) + assert resp.status_code == 400 + + def test_list_returns_only_authorized_subscription(self, app_client): + """subscription_id from JWT must be passed as a SQL parameter.""" + row_authorized = _sample_pattern_row(sub_id="sub-authorized") + mock_db = _mock_db_rows([row_authorized], 1) + + with patch("api.routes.patterns._get_db", return_value=mock_db): + resp = app_client.get( + "/api/v1/patterns", + headers=_auth_headers(sub_id="sub-authorized"), + ) + + assert resp.status_code == 200 + # Verify subscription_id was enforced in the SQL WHERE clause parameters. + mock_cursor = mock_db._get_conn.return_value.cursor.return_value + first_call = mock_cursor.execute.call_args_list[0] + _, params = first_call[0] + assert "sub-authorized" in params + + def test_list_requires_auth(self, app_client): + resp = app_client.get("/api/v1/patterns") + assert resp.status_code == 401 + + +class TestPatternsRouteGet: + def _mock_get_db(self, row): + mock_db = MagicMock() + mock_conn = MagicMock() + mock_cursor = MagicMock() + mock_cursor.__enter__ = MagicMock(return_value=mock_cursor) + mock_cursor.__exit__ = MagicMock(return_value=False) + mock_cursor.fetchone.return_value = row + mock_conn.cursor.return_value = mock_cursor + mock_db._get_conn.return_value = mock_conn + return mock_db + + def test_get_existing_pattern(self, app_client): + row = _sample_pattern_row() + mock_db = self._mock_get_db(row) + + with patch("api.routes.patterns._get_db", return_value=mock_db): + resp = app_client.get( + "/api/v1/patterns/1", + headers=_auth_headers(sub_id=SUB_ID), + ) + + assert resp.status_code == 200 + data = resp.get_json() + assert data["id"] == 1 + assert data["pattern_type"] == "persistent_finding" + assert data["threshold"] == 3 + assert data["algorithm_version"] == "1" + + def test_get_nonexistent_pattern_returns_404(self, app_client): + # DB returns None when id + subscription_id don't match any row. + mock_db = self._mock_get_db(None) + + with patch("api.routes.patterns._get_db", return_value=mock_db): + resp = app_client.get( + "/api/v1/patterns/99999", + headers=_auth_headers(sub_id=SUB_ID), + ) + + assert resp.status_code == 404 + + def test_get_cross_subscription_returns_404(self, app_client): + # Pattern exists for sub-001 but caller is scoped to sub-other. + # The SQL adds subscription_id = %s to the WHERE clause so the DB + # returns None, and the caller sees 404 (not 403, to avoid leaking + # that the pattern ID exists in another subscription). + mock_db = self._mock_get_db(None) + + with patch("api.routes.patterns._get_db", return_value=mock_db): + resp = app_client.get( + "/api/v1/patterns/1", + headers=_auth_headers(sub_id="sub-other"), + ) + + assert resp.status_code == 404 + # Verify subscription_id was enforced as a SQL parameter (not just checked + # in Python), so removing the WHERE clause would break this test. + mock_cursor = mock_db._get_conn.return_value.cursor.return_value + call_args = mock_cursor.execute.call_args + _, params = call_args[0] + assert "sub-other" in params + + def test_get_requires_auth(self, app_client): + resp = app_client.get("/api/v1/patterns/1") + assert resp.status_code == 401