scanner/layer8_feedback.py

"""
Layer 8 — Feedback Store (SQLite)
Self-improving per-repo. Human corrections persist and skip re-evaluation on future scans.
"""
from __future__ import annotations
import sqlite3
import hashlib
import os
import contextlib
import threading
from .models import Finding

DEFAULT_DB_PATH = os.environ.get("FEEDBACK_DB_PATH", "/data/feedback.sqlite")
FEEDBACK_DB = DEFAULT_DB_PATH

_lock = threading.Lock()

def get_db_path(db_path: str | None = None) -> str:
    if db_path is None:
        db_path = FEEDBACK_DB
    try:
        dir_name = os.path.dirname(db_path)
        if dir_name:
            os.makedirs(dir_name, exist_ok=True)
        # Try to open/create a test connection to verify access
        conn = sqlite3.connect(db_path)
        conn.close()
        return db_path
    except (PermissionError, sqlite3.OperationalError, OSError):
        fallback_path = "./.vibesec_feedback.sqlite"
        dir_name = os.path.dirname(fallback_path)
        if dir_name:
            os.makedirs(dir_name, exist_ok=True)
        return fallback_path


def _ensure_schema(conn: sqlite3.Connection):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS dismissed_findings (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            rule_id TEXT NOT NULL,
            file_role TEXT,
            code_hash TEXT NOT NULL,
            verdict TEXT NOT NULL CHECK(verdict IN ('false_positive', 'real')),
            confirmed_by TEXT NOT NULL CHECK(confirmed_by IN ('human', 'tier1', 'tier2', 'tier3')),
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
        )
    """)
    try:
        # Deduplicate existing rows before applying unique index to prevent IntegrityError
        conn.execute("""
            DELETE FROM dismissed_findings WHERE id NOT IN (
                SELECT MAX(id) FROM dismissed_findings GROUP BY rule_id, code_hash
            )
        """)
    except Exception:
        pass
    try:
        conn.execute("DROP INDEX IF EXISTS idx_rule_hash")
    except Exception:
        pass
    conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_rule_hash ON dismissed_findings (rule_id, code_hash)")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS rule_fp_rates (
            check_id TEXT,
            file_role TEXT,
            seen_count INTEGER DEFAULT 0,
            fp_count INTEGER DEFAULT 0,
            PRIMARY KEY (check_id, file_role)
        )
    """)
    conn.commit()


@contextlib.contextmanager
def get_db_connection():
    db_path = get_db_path()
    conn = sqlite3.connect(db_path)
    _ensure_schema(conn)
    conn.row_factory = lambda cursor, row: {col[0]: row[idx] for idx, col in enumerate(cursor.description)}
    try:
        yield conn
    finally:
        conn.close()


def record_verdict(check_id: str, file_role: str, is_false_positive: bool):
    with _lock:
        with get_db_connection() as conn:
            cursor = conn.execute(
                "SELECT seen_count, fp_count FROM rule_fp_rates WHERE check_id = ? AND file_role = ?",
                (check_id, file_role),
            )
            row = cursor.fetchone()
            if row:
                seen = row["seen_count"] + 1
                fp = row["fp_count"] + (1 if is_false_positive else 0)
                conn.execute(
                    "UPDATE rule_fp_rates SET seen_count = ?, fp_count = ? WHERE check_id = ? AND file_role = ?",
                    (seen, fp, check_id, file_role),
                )
            else:
                seen = 1
                fp = 1 if is_false_positive else 0
                conn.execute(
                    "INSERT INTO rule_fp_rates (check_id, file_role, seen_count, fp_count) VALUES (?, ?, ?, ?)",
                    (check_id, file_role, seen, fp),
                )
            conn.commit()


def get_high_fp_exclusions(threshold: float = 0.90, min_samples: int = 10):
    with get_db_connection() as conn:
        cursor = conn.execute(
            "SELECT check_id, file_role FROM rule_fp_rates WHERE seen_count >= ? AND (fp_count * 1.0 / seen_count) >= ?",
            (min_samples, threshold),
        )
        return {(row["check_id"], row["file_role"]) for row in cursor.fetchall()}


class FeedbackStore:
    def __init__(self, db_path: str = DEFAULT_DB_PATH):
        resolved_path = get_db_path(db_path)
        self.conn = sqlite3.connect(resolved_path, check_same_thread=False)
        self._init_schema()

    def _init_schema(self):
        _ensure_schema(self.conn)

    def _code_hash(self, finding: Finding) -> str:
        # TODO (Priority: HIGH) - ISSUE #108: Resolve Feedback Store Line Shift Vulnerability
        # Currently, the hash includes the exact line_number. If a file is modified (e.g. inserting lines at the top),
        # the line number of a previously-dismissed finding shifts. This results in a new hash, causing the
        # dismissed finding to reappear in future scans.
        #
        # Future Mitigation Plan:
        # 1. Update the dismissed_findings database schema to store the absolute file_path and line_number columns separately.
        # 2. Store the original code snippet (normalized content of the line) as a column.
        # 3. Implement fuzzy matching during verification: if a finding matches rule_id and file_path, check if a dismissed
        #    finding exists within a +/- 5 lines tolerance range, or verify that the normalized line content is identical.
        content = f"{finding.check_id}:{finding.file_path}:{finding.line_number}"
        return hashlib.sha256(content.encode()).hexdigest()[:16]

    def is_known_false_positive(self, finding: Finding) -> bool:
        code_hash = self._code_hash(finding)
        rule_id = finding.check_id or ""
        row = self.conn.execute(
            "SELECT verdict FROM dismissed_findings WHERE rule_id=? AND code_hash=? ORDER BY timestamp DESC LIMIT 1",
            (rule_id, code_hash)
        ).fetchone()
        return row is not None and row[0] == "false_positive"

    def record_verdict(
        self,
        finding: Finding,
        verdict: str,
        confirmed_by: str = "human",
    ):
        code_hash = self._code_hash(finding)
        self.conn.execute("""
            INSERT OR REPLACE INTO dismissed_findings (rule_id, file_role, code_hash, verdict, confirmed_by)
            VALUES (?, ?, ?, ?, ?)
        """, (finding.check_id or "", finding.file_role or "", code_hash, verdict, confirmed_by))
        self.conn.commit()

    def close(self):
        self.conn.close()