admin / Synapse-NetscanXi

"""
SQLite persistence layer for NetScan Xi.

Stores scans, the hosts found in each scan, and a derived change-log so we can
answer "what changed since last time?" without re-deriving it on every request.

Schema is intentionally simple and self-contained (stdlib sqlite3 only).
"""

import json
import os
import sqlite3
import threading
import time

_DB_LOCK = threading.Lock()
DEFAULT_DB_PATH = os.environ.get(
    "NETSCAN_DB",
    os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "netscan.db"),
)

SCHEMA = """
CREATE TABLE IF NOT EXISTS scans (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    target      TEXT NOT NULL,
    profile     TEXT NOT NULL,
    status      TEXT NOT NULL,           -- running | done | error | cancelled
    started     REAL NOT NULL,
    finished    REAL,
    host_count  INTEGER DEFAULT 0,
    vuln_count  INTEGER DEFAULT 0,
    error       TEXT,
    triggered_by TEXT DEFAULT 'manual',  -- manual | schedule
    tenant      TEXT NOT NULL DEFAULT 'default',  -- v8: soft multi-tenancy tag
    scan_type   TEXT NOT NULL DEFAULT 'active'    -- v9: active | passive
);

-- v9: persistent asset registry. Each physical device (keyed on MAC) gets a
-- stable 8-char alphanumeric ID, so vulnerabilities/findings are not duplicated
-- across scans and a manual device-type correction sticks. Hosts with no MAC
-- (routed/remote IPs) fall back to an IP-derived key.
CREATE TABLE IF NOT EXISTS assets (
    asset_id    TEXT PRIMARY KEY,        -- 8-char [A-Z0-9] unique ID
    mac         TEXT,                    -- canonical MAC (uppercased) or NULL
    key         TEXT NOT NULL UNIQUE,    -- dedupe key: 'mac:AA:..' or 'ip:1.2.3.4'
    type_override TEXT,                  -- user-corrected device type (sticky)
    first_seen  REAL NOT NULL,
    last_seen   REAL NOT NULL,
    last_ip     TEXT,
    tenant      TEXT NOT NULL DEFAULT 'default'
);

CREATE TABLE IF NOT EXISTS hosts (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    scan_id   INTEGER NOT NULL,
    ip        TEXT NOT NULL,
    data      TEXT NOT NULL,             -- full host dict as JSON
    risk      INTEGER DEFAULT 0,
    vuln_count INTEGER DEFAULT 0,
    FOREIGN KEY (scan_id) REFERENCES scans(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS changes (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    scan_id   INTEGER NOT NULL,
    ip        TEXT NOT NULL,
    kind      TEXT NOT NULL,             -- host_new|host_gone|port_new|port_gone|service_changed|vuln_new|os_changed
    detail    TEXT NOT NULL,
    severity  TEXT DEFAULT 'info',       -- info | warn | critical
    ts        REAL NOT NULL,
    FOREIGN KEY (scan_id) REFERENCES scans(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS users (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    username      TEXT NOT NULL UNIQUE COLLATE NOCASE,
    password_hash TEXT NOT NULL,
    role          TEXT NOT NULL DEFAULT 'viewer',   -- admin | operator | viewer
    tenant        TEXT NOT NULL DEFAULT 'default',   -- v8: soft multi-tenancy
    created_at    REAL NOT NULL,
    last_login    REAL,
    mfa_required  INTEGER NOT NULL DEFAULT 0,        -- admin toggle: 1 => TOTP enforced
    mfa_secret    TEXT,                              -- base32 TOTP secret; NULL => not enrolled
    mfa_enrolled_at REAL
);

-- One-time backup/recovery codes for MFA users. Stored hashed.
CREATE TABLE IF NOT EXISTS mfa_backup_codes (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id   INTEGER NOT NULL,
    code_hash TEXT NOT NULL,
    used_at   REAL,                                 -- NULL => still usable
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS idx_hosts_scan ON hosts(scan_id);
CREATE INDEX IF NOT EXISTS idx_hosts_ip ON hosts(ip);
CREATE INDEX IF NOT EXISTS idx_changes_scan ON changes(scan_id);
-- Scheduled recurring scans, managed from the UI (v6).
CREATE TABLE IF NOT EXISTS schedules (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    name          TEXT NOT NULL,
    target        TEXT NOT NULL DEFAULT '',         -- '' => auto-detect /24
    profile       TEXT NOT NULL DEFAULT 'standard',
    options       TEXT,                             -- JSON scan-option overrides
    interval_minutes INTEGER NOT NULL,
    enabled       INTEGER NOT NULL DEFAULT 1,
    created_at    REAL NOT NULL,
    last_run      REAL,
    next_run      REAL,
    -- v9: weekly day+time schedule. days = comma list of 0-6 (Mon=0..Sun=6),
    -- hour/minute = local wall-clock time. When days is set, this overrides the
    -- plain interval and fires on the next matching day/time.
    days          TEXT,
    hour          INTEGER,
    minute        INTEGER,
    scan_type     TEXT NOT NULL DEFAULT 'active'
);

-- v8: scheduled scans also carry a tenant tag for data separation.
-- v8: per-user activity / audit log (logins, logouts, key actions).
CREATE TABLE IF NOT EXISTS activity_log (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    ts        REAL NOT NULL,
    user_id   INTEGER,                          -- NULL for anonymous/failed
    username  TEXT,                             -- snapshot at event time
    tenant    TEXT NOT NULL DEFAULT 'default',
    action    TEXT NOT NULL,                    -- login | logout | login_failed | scan | export | ...
    detail    TEXT,
    ip        TEXT                              -- client IP (best effort)
);

-- v9.1: CVE finding lifecycle. A user can mark a specific CVE (optionally
-- scoped to a single asset) as remediated / accepted_risk / false_positive.
-- Keyed on (tenant, asset_id, cve) so the disposition follows the asset across
-- scans. asset_id may be '*' for a tenant-wide disposition of a CVE.
CREATE TABLE IF NOT EXISTS cve_lifecycle (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant      TEXT NOT NULL DEFAULT 'default',
    asset_id    TEXT NOT NULL DEFAULT '*',
    cve         TEXT NOT NULL,                 -- CVE id or NSE finding id, uppercased
    state       TEXT NOT NULL,                 -- remediated | accepted_risk | false_positive | open
    note        TEXT,
    updated_by  TEXT,
    updated_at  REAL NOT NULL,
    UNIQUE(tenant, asset_id, cve)
);

-- v9.1: historical risk + compliance posture snapshots, one row per done
-- scan, so we can plot 30/90/365-day trend graphs without re-deriving.
CREATE TABLE IF NOT EXISTS risk_snapshots (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    scan_id       INTEGER,
    tenant        TEXT NOT NULL DEFAULT 'default',
    ts            REAL NOT NULL,
    risk_score    REAL NOT NULL DEFAULT 0,      -- 0-100 network risk score
    compliance_pct REAL NOT NULL DEFAULT 100,   -- 0-100 overall pass rate
    hosts         INTEGER DEFAULT 0,
    vulns         INTEGER DEFAULT 0,
    kev           INTEGER DEFAULT 0,
    high_risk     INTEGER DEFAULT 0
);

-- v9.1: key/value settings store for the Admin UI (data-retention policy etc).
CREATE TABLE IF NOT EXISTS settings (
    key         TEXT PRIMARY KEY,
    value       TEXT,
    updated_at  REAL
);

-- v9.1: service-desk integrations (Jira / ServiceNow / Zendesk). Secrets are
-- stored here; in production point NETSCAN_DB at an encrypted volume.
CREATE TABLE IF NOT EXISTS integrations (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant      TEXT NOT NULL DEFAULT 'default',
    platform    TEXT NOT NULL,                 -- jira | servicenow | zendesk
    name        TEXT NOT NULL,
    base_url    TEXT NOT NULL,
    auth_user   TEXT,
    auth_secret TEXT,                          -- API token / password
    project_key TEXT,                          -- Jira project / SNOW table / ZD group
    enabled     INTEGER NOT NULL DEFAULT 1,
    created_at  REAL NOT NULL,
    last_used   REAL
);

-- v9.1: tickets opened in a service desk for a finding (audit + dedupe).
CREATE TABLE IF NOT EXISTS tickets (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant        TEXT NOT NULL DEFAULT 'default',
    integration_id INTEGER,
    platform      TEXT NOT NULL,
    asset_id      TEXT,
    cve           TEXT,
    external_key  TEXT,                         -- e.g. JIRA-123 / INC0010001
    external_url  TEXT,
    status        TEXT DEFAULT 'open',
    created_by    TEXT,
    created_at    REAL NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_users_username ON users(username);
CREATE INDEX IF NOT EXISTS idx_backup_user ON mfa_backup_codes(user_id);
CREATE INDEX IF NOT EXISTS idx_sched_enabled ON schedules(enabled);
CREATE INDEX IF NOT EXISTS idx_scans_tenant ON scans(tenant);
CREATE INDEX IF NOT EXISTS idx_assets_mac ON assets(mac);
CREATE INDEX IF NOT EXISTS idx_assets_tenant ON assets(tenant);
CREATE INDEX IF NOT EXISTS idx_activity_ts ON activity_log(ts);
CREATE INDEX IF NOT EXISTS idx_activity_tenant ON activity_log(tenant);
CREATE INDEX IF NOT EXISTS idx_cvelc_lookup ON cve_lifecycle(tenant, asset_id, cve);
CREATE INDEX IF NOT EXISTS idx_snap_tenant_ts ON risk_snapshots(tenant, ts);
CREATE INDEX IF NOT EXISTS idx_integrations_tenant ON integrations(tenant);
CREATE INDEX IF NOT EXISTS idx_tickets_tenant ON tickets(tenant);

-- v12: asset groups (admin-defined) + membership. Tenant-separated.
CREATE TABLE IF NOT EXISTS asset_groups (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant      TEXT NOT NULL DEFAULT 'default',
    name        TEXT NOT NULL,
    description TEXT,
    created_by  TEXT,
    created_at  REAL NOT NULL,
    UNIQUE(tenant, name)
);
CREATE TABLE IF NOT EXISTS asset_group_members (
    group_id    INTEGER NOT NULL,
    asset_id    TEXT NOT NULL,
    tenant      TEXT NOT NULL DEFAULT 'default',
    added_at    REAL NOT NULL,
    PRIMARY KEY (group_id, asset_id),
    FOREIGN KEY (group_id) REFERENCES asset_groups(id) ON DELETE CASCADE
);

-- v12: remediation tracking. One row per remediation activity against an
-- asset, tenant-separated so a tenant only ever sees its own items.
CREATE TABLE IF NOT EXISTS remediation (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    tenant      TEXT NOT NULL DEFAULT 'default',
    asset_id    TEXT NOT NULL,
    title       TEXT NOT NULL,
    detail      TEXT,
    status      TEXT NOT NULL DEFAULT 'open',     -- open | in_progress | blocked | resolved
    priority    TEXT NOT NULL DEFAULT 'medium',   -- low | medium | high | critical
    assignee    TEXT,
    due_date    TEXT,                             -- ISO date 'YYYY-MM-DD' or NULL
    cve         TEXT,
    created_by  TEXT,
    created_at  REAL NOT NULL,
    updated_at  REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_agroups_tenant ON asset_groups(tenant);
CREATE INDEX IF NOT EXISTS idx_agmembers_asset ON asset_group_members(tenant, asset_id);
CREATE INDEX IF NOT EXISTS idx_agmembers_group ON asset_group_members(group_id);
CREATE INDEX IF NOT EXISTS idx_remediation_tenant ON remediation(tenant, asset_id);
-- v13: free-form comment thread attached to a remediation activity. Comments
-- are tenant-tagged (mirroring the parent) so they stay tenant-separated, and
-- are removed with their parent on delete.
CREATE TABLE IF NOT EXISTS remediation_comments (
    id             INTEGER PRIMARY KEY AUTOINCREMENT,
    remediation_id INTEGER NOT NULL,
    tenant         TEXT NOT NULL DEFAULT 'default',
    author         TEXT,
    body           TEXT NOT NULL,
    created_at     REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_remcomments_rid ON remediation_comments(remediation_id);
"""

# Columns added after the original users table shipped. Applied idempotently on
# init so existing databases gain MFA support without a manual migration.
_USER_MIGRATIONS = (
    ("mfa_required", "INTEGER NOT NULL DEFAULT 0"),
    ("mfa_secret", "TEXT"),
    ("mfa_enrolled_at", "REAL"),
    ("tenant", "TEXT NOT NULL DEFAULT 'default'"),  # v8 soft multi-tenancy
)

# v8: columns added to scans / schedules after they first shipped.
_SCAN_MIGRATIONS = (
    ("tenant", "TEXT NOT NULL DEFAULT 'default'"),
    ("scan_type", "TEXT NOT NULL DEFAULT 'active'"),   # v9: active | passive
)
_SCHEDULE_MIGRATIONS = (
    ("tenant", "TEXT NOT NULL DEFAULT 'default'"),
    ("days", "TEXT"),                                  # v9: weekly day+time
    ("hour", "INTEGER"),
    ("minute", "INTEGER"),
    ("scan_type", "TEXT NOT NULL DEFAULT 'active'"),
)

# Recognised roles, ordered from most to least privileged.
# v9.1: "tenant_admin" sits between operator and admin — full control over its
# own tenant (users, schedules, scans, settings) but never across tenants.
ROLES = ("admin", "tenant_admin", "operator", "viewer")


def _connect(db_path):
    conn = sqlite3.connect(db_path, timeout=30)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA journal_mode = WAL")
    return conn


def _apply_column_migrations(conn, table, migrations):
    """Idempotently add any missing columns to an existing table."""
    cols = {r["name"] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()}
    for name, decl in migrations:
        if name not in cols:
            conn.execute(f"ALTER TABLE {table} ADD COLUMN {name} {decl}")


def _apply_user_migrations(conn):
    """Add any columns missing from pre-existing tables."""
    _apply_column_migrations(conn, "users", _USER_MIGRATIONS)
    _apply_column_migrations(conn, "scans", _SCAN_MIGRATIONS)
    _apply_column_migrations(conn, "schedules", _SCHEDULE_MIGRATIONS)


def init_db(db_path=DEFAULT_DB_PATH):
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    with _DB_LOCK, _connect(db_path) as conn:
        conn.executescript(SCHEMA)
        _apply_user_migrations(conn)
    return db_path


# ---------------------------------------------------------------------------
# Scan lifecycle
# ---------------------------------------------------------------------------
def create_scan(target, profile, triggered_by="manual", tenant="default",
                scan_type="active", db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "INSERT INTO scans (target, profile, status, started, triggered_by, "
            "tenant, scan_type) VALUES (?, ?, 'running', ?, ?, ?, ?)",
            (target, profile, time.time(), triggered_by, tenant or "default",
             scan_type or "active"),
        )
        return cur.lastrowid


def finish_scan(scan_id, status, hosts, error=None, db_path=DEFAULT_DB_PATH):
    vuln_total = sum(h.get("vuln_count", 0) for h in hosts)
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute(
            "UPDATE scans SET status=?, finished=?, host_count=?, vuln_count=?, error=? "
            "WHERE id=?",
            (status, time.time(), len(hosts), vuln_total, error, scan_id),
        )
        for h in hosts:
            conn.execute(
                "INSERT INTO hosts (scan_id, ip, data, risk, vuln_count) "
                "VALUES (?, ?, ?, ?, ?)",
                (scan_id, h.get("ip", ""), json.dumps(h),
                 h.get("risk", 0), h.get("vuln_count", 0)),
            )


def record_changes(scan_id, changes, db_path=DEFAULT_DB_PATH):
    if not changes:
        return
    now = time.time()
    with _DB_LOCK, _connect(db_path) as conn:
        for c in changes:
            conn.execute(
                "INSERT INTO changes (scan_id, ip, kind, detail, severity, ts) "
                "VALUES (?, ?, ?, ?, ?, ?)",
                (scan_id, c["ip"], c["kind"], c["detail"],
                 c.get("severity", "info"), now),
            )


# ---------------------------------------------------------------------------
# Queries
# ---------------------------------------------------------------------------
def get_scan(scan_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM scans WHERE id=?", (scan_id,)).fetchone()
        return dict(row) if row else None


def latest_scan(status="done", tenant=None, db_path=DEFAULT_DB_PATH):
    """Most recent scan. If `tenant` is given, scope to that tenant."""
    with _DB_LOCK, _connect(db_path) as conn:
        clauses, params = [], []
        if status:
            clauses.append("status=?")
            params.append(status)
        if tenant is not None:
            clauses.append("tenant=?")
            params.append(tenant)
        where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
        row = conn.execute(
            f"SELECT * FROM scans {where} ORDER BY id DESC LIMIT 1", params
        ).fetchone()
        return dict(row) if row else None


def previous_done_scan(before_scan_id, tenant=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            row = conn.execute(
                "SELECT * FROM scans WHERE status='done' AND id<? AND tenant=? "
                "ORDER BY id DESC LIMIT 1",
                (before_scan_id, tenant),
            ).fetchone()
        else:
            row = conn.execute(
                "SELECT * FROM scans WHERE status='done' AND id<? "
                "ORDER BY id DESC LIMIT 1",
                (before_scan_id,),
            ).fetchone()
        return dict(row) if row else None


def get_hosts(scan_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT data FROM hosts WHERE scan_id=? ORDER BY id", (scan_id,)
        ).fetchall()
        return [json.loads(r["data"]) for r in rows]


def list_scans(limit=50, tenant=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                "SELECT id, target, profile, status, started, finished, host_count, "
                "vuln_count, triggered_by, tenant, scan_type FROM scans WHERE tenant=? "
                "ORDER BY id DESC LIMIT ?",
                (tenant, limit),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT id, target, profile, status, started, finished, host_count, "
                "vuln_count, triggered_by, tenant, scan_type FROM scans ORDER BY id DESC LIMIT ?",
                (limit,),
            ).fetchall()
        return [dict(r) for r in rows]


def list_tenants(db_path=DEFAULT_DB_PATH):
    """Distinct tenant tags seen across users and scans."""
    with _DB_LOCK, _connect(db_path) as conn:
        tenants = set()
        for tbl in ("users", "scans"):
            try:
                for r in conn.execute(f"SELECT DISTINCT tenant FROM {tbl}").fetchall():
                    if r["tenant"]:
                        tenants.add(r["tenant"])
            except sqlite3.OperationalError:
                pass
    tenants.add("default")
    return sorted(tenants)


def get_changes(scan_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT ip, kind, detail, severity, ts FROM changes "
            "WHERE scan_id=? ORDER BY "
            "CASE severity WHEN 'critical' THEN 0 WHEN 'warn' THEN 1 ELSE 2 END, ip",
            (scan_id,),
        ).fetchall()
        return [dict(r) for r in rows]


def top_vulnerabilities(scan_id, limit=10, db_path=DEFAULT_DB_PATH):
    """Aggregate the highest-severity vulns across all hosts in a scan."""
    hosts = get_hosts(scan_id, db_path=db_path)
    agg = {}
    for h in hosts:
        for v in h.get("vulns", []):
            vid = v.get("id", "")
            if not vid:
                continue
            cvss = 0.0
            try:
                cvss = float(v.get("cvss") or 0)
            except (TypeError, ValueError):
                cvss = 0.0
            entry = agg.setdefault(vid, {
                "id": vid,
                "cve": v.get("cve", ""),
                "cvss": cvss,
                "exploit": False,
                "kev": v.get("kev", False),
                "description": v.get("description", ""),
                "mitigation": v.get("mitigation", ""),
                "hosts": set(),
            })
            entry["cvss"] = max(entry["cvss"], cvss)
            entry["exploit"] = entry["exploit"] or bool(v.get("exploit"))
            entry["kev"] = entry["kev"] or bool(v.get("kev"))
            if v.get("description") and not entry["description"]:
                entry["description"] = v["description"]
            if v.get("mitigation") and not entry["mitigation"]:
                entry["mitigation"] = v["mitigation"]
            if v.get("cve") and not entry["cve"]:
                entry["cve"] = v["cve"]
            entry["hosts"].add(h.get("ip", ""))

    items = []
    for e in agg.values():
        e["host_count"] = len(e["hosts"])
        e["hosts"] = sorted(x for x in e["hosts"] if x)
        items.append(e)

    # Rank: KEV first, then exploit, then CVSS, then host spread.
    items.sort(key=lambda e: (
        e["kev"], e["exploit"], e["cvss"], e["host_count"]
    ), reverse=True)
    return items[:limit]


# ---------------------------------------------------------------------------
# User management (v4)
# ---------------------------------------------------------------------------
def _row_to_user(row):
    if not row:
        return None
    keys = row.keys()
    return {
        "id": row["id"],
        "username": row["username"],
        "password_hash": row["password_hash"],
        "role": row["role"],
        "tenant": row["tenant"] if "tenant" in keys else "default",
        "created_at": row["created_at"],
        "last_login": row["last_login"],
        "mfa_required": bool(row["mfa_required"]) if "mfa_required" in keys else False,
        "mfa_secret": row["mfa_secret"] if "mfa_secret" in keys else None,
        "mfa_enrolled_at": row["mfa_enrolled_at"] if "mfa_enrolled_at" in keys else None,
    }


def create_user(username, password_hash, role="viewer", mfa_required=False,
                tenant="default", db_path=DEFAULT_DB_PATH):
    """Insert a user. Returns new id, or None if the username already exists."""
    if role not in ROLES:
        role = "viewer"
    try:
        with _DB_LOCK, _connect(db_path) as conn:
            cur = conn.execute(
                "INSERT INTO users (username, password_hash, role, tenant, created_at, "
                "mfa_required) VALUES (?, ?, ?, ?, ?, ?)",
                (username.strip(), password_hash, role, tenant or "default",
                 time.time(), 1 if mfa_required else 0),
            )
            return cur.lastrowid
    except sqlite3.IntegrityError:
        return None


def get_user(username, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute(
            "SELECT * FROM users WHERE username = ? COLLATE NOCASE",
            (username.strip(),),
        ).fetchone()
        return _row_to_user(row)


def get_user_by_id(user_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
        return _row_to_user(row)


def list_users(db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT id, username, role, tenant, created_at, last_login, "
            "mfa_required, mfa_secret FROM users ORDER BY username COLLATE NOCASE"
        ).fetchall()
        return [{"id": r["id"], "username": r["username"], "role": r["role"],
                 "tenant": r["tenant"] if "tenant" in r.keys() else "default",
                 "created_at": r["created_at"], "last_login": r["last_login"],
                 "mfa_required": bool(r["mfa_required"]),
                 "mfa_enrolled": r["mfa_secret"] is not None}
                for r in rows]


def count_users(db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        return conn.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"]


def count_admins(db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        return conn.execute(
            "SELECT COUNT(*) AS c FROM users WHERE role = 'admin'"
        ).fetchone()["c"]


def set_password(user_id, password_hash, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("UPDATE users SET password_hash = ? WHERE id = ?",
                     (password_hash, user_id))


def set_username(user_id, username, db_path=DEFAULT_DB_PATH):
    """v8-r2: rename a user. Returns False if the new name collides (unique)."""
    username = (username or "").strip()
    if not username:
        return False
    try:
        with _DB_LOCK, _connect(db_path) as conn:
            cur = conn.execute("UPDATE users SET username = ? WHERE id = ?",
                               (username, user_id))
            return cur.rowcount > 0
    except sqlite3.IntegrityError:
        return False


def set_role(user_id, role, db_path=DEFAULT_DB_PATH):
    if role not in ROLES:
        return False
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("UPDATE users SET role = ? WHERE id = ?", (role, user_id))
    return True


def set_tenant(user_id, tenant, db_path=DEFAULT_DB_PATH):
    """v8: assign a user to a tenant (soft multi-tenancy)."""
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("UPDATE users SET tenant = ? WHERE id = ?",
                     (tenant or "default", user_id))
    return True


def set_mfa_required(user_id, required, db_path=DEFAULT_DB_PATH):
    """Toggle whether TOTP is enforced for a user (admin control)."""
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("UPDATE users SET mfa_required = ? WHERE id = ?",
                     (1 if required else 0, user_id))


def set_mfa_secret(user_id, secret, db_path=DEFAULT_DB_PATH):
    """Store (or clear, with secret=None) a user's TOTP secret + enroll time."""
    enrolled = time.time() if secret else None
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute(
            "UPDATE users SET mfa_secret = ?, mfa_enrolled_at = ? WHERE id = ?",
            (secret, enrolled, user_id),
        )


def disable_mfa(user_id, db_path=DEFAULT_DB_PATH):
    """Reset a user's enrollment: clear the secret and all backup codes."""
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute(
            "UPDATE users SET mfa_secret = NULL, mfa_enrolled_at = NULL WHERE id = ?",
            (user_id,),
        )
        conn.execute("DELETE FROM mfa_backup_codes WHERE user_id = ?", (user_id,))


def replace_backup_codes(user_id, code_hashes, db_path=DEFAULT_DB_PATH):
    """Replace a user's backup codes with a fresh hashed set."""
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("DELETE FROM mfa_backup_codes WHERE user_id = ?", (user_id,))
        conn.executemany(
            "INSERT INTO mfa_backup_codes (user_id, code_hash) VALUES (?, ?)",
            [(user_id, h) for h in code_hashes],
        )


def list_unused_backup_hashes(user_id, db_path=DEFAULT_DB_PATH):
    """Return (id, code_hash) for a user's still-usable backup codes."""
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT id, code_hash FROM mfa_backup_codes "
            "WHERE user_id = ? AND used_at IS NULL",
            (user_id,),
        ).fetchall()
        return [(r["id"], r["code_hash"]) for r in rows]


def consume_backup_code(code_id, db_path=DEFAULT_DB_PATH):
    """Mark one backup code as used."""
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("UPDATE mfa_backup_codes SET used_at = ? WHERE id = ?",
                     (time.time(), code_id))


def count_unused_backup_codes(user_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        return conn.execute(
            "SELECT COUNT(*) AS c FROM mfa_backup_codes "
            "WHERE user_id = ? AND used_at IS NULL",
            (user_id,),
        ).fetchone()["c"]


def touch_login(user_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("UPDATE users SET last_login = ? WHERE id = ?",
                     (time.time(), user_id))


def delete_user(user_id, db_path=DEFAULT_DB_PATH):
    """Delete a user. Refuses to remove the last remaining admin."""
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT role FROM users WHERE id = ?",
                           (user_id,)).fetchone()
        if not row:
            return False
        if row["role"] == "admin":
            admins = conn.execute(
                "SELECT COUNT(*) AS c FROM users WHERE role = 'admin'"
            ).fetchone()["c"]
            if admins <= 1:
                return False  # never delete the last admin
        conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
        return True


# ---------------------------------------------------------------------------
# Asset registry (v9) - stable 8-char IDs bound to MAC, sticky type override
# ---------------------------------------------------------------------------
import random
import string

_ID_ALPHABET = string.ascii_uppercase + string.digits  # [A-Z0-9]


def _new_asset_id(conn):
    """Generate a unique 8-char alphanumeric asset ID not already in use."""
    for _ in range(50):
        candidate = "".join(random.choice(_ID_ALPHABET) for _ in range(8))
        hit = conn.execute(
            "SELECT 1 FROM assets WHERE asset_id = ?", (candidate,)
        ).fetchone()
        if not hit:
            return candidate
    # Extremely unlikely fallback: timestamp-derived.
    return ("".join(random.choice(_ID_ALPHABET) for _ in range(4))
            + format(int(time.time()) % 1000000, "06d"))[:8]


def asset_key_for(mac, ip):
    """Canonical dedupe key for an asset: prefer MAC, else IP."""
    mac = (mac or "").strip().upper()
    if mac and mac not in ("-", ""):
        return "mac:" + mac, mac
    ip = (ip or "").strip()
    return "ip:" + ip, None


def resolve_asset(mac, ip, tenant="default", db_path=DEFAULT_DB_PATH):
    """Return the persistent asset record for a (mac, ip), creating it if new.

    The 8-char `asset_id` is bound to the MAC (or IP when MAC is unknown) and
    reused on every future scan, so vulnerabilities/findings tied to that ID are
    never duplicated for the same physical device. Returns a dict with
    asset_id / type_override / first_seen.
    """
    key, canon_mac = asset_key_for(mac, ip)
    now = time.time()
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute(
            "SELECT * FROM assets WHERE key = ?", (key,)
        ).fetchone()
        if row:
            conn.execute(
                "UPDATE assets SET last_seen = ?, last_ip = ? WHERE asset_id = ?",
                (now, ip, row["asset_id"]),
            )
            return {"asset_id": row["asset_id"],
                    "type_override": row["type_override"],
                    "first_seen": row["first_seen"],
                    "is_new": False}
        aid = _new_asset_id(conn)
        conn.execute(
            "INSERT INTO assets (asset_id, mac, key, type_override, first_seen, "
            "last_seen, last_ip, tenant) VALUES (?, ?, ?, NULL, ?, ?, ?, ?)",
            (aid, canon_mac, key, now, now, ip, tenant or "default"),
        )
        return {"asset_id": aid, "type_override": None,
                "first_seen": now, "is_new": True}


def get_asset(asset_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute(
            "SELECT * FROM assets WHERE asset_id = ?", (asset_id,)
        ).fetchone()
        return dict(row) if row else None


def set_asset_type_override(asset_id, device_type, db_path=DEFAULT_DB_PATH):
    """v9: store a user correction of an asset's device type. Empty clears it."""
    dt = (device_type or "").strip() or None
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "UPDATE assets SET type_override = ? WHERE asset_id = ?", (dt, asset_id)
        )
        return cur.rowcount > 0


def list_assets(tenant=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                "SELECT * FROM assets WHERE tenant = ? ORDER BY last_seen DESC",
                (tenant,),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT * FROM assets ORDER BY last_seen DESC"
            ).fetchall()
        return [dict(r) for r in rows]


# ---------------------------------------------------------------------------
# Scheduled scans (v6) - UI-managed, persisted
# ---------------------------------------------------------------------------
def _row_to_schedule(row):
    if not row:
        return None
    opts = None
    if row["options"]:
        try:
            opts = json.loads(row["options"])
        except (ValueError, TypeError):
            opts = None
    keys = row.keys()
    days = None
    if "days" in keys and row["days"]:
        days = [int(d) for d in str(row["days"]).split(",") if d != ""]
    return {
        "id": row["id"],
        "name": row["name"],
        "target": row["target"],
        "profile": row["profile"],
        "options": opts,
        "interval_minutes": row["interval_minutes"],
        "enabled": bool(row["enabled"]),
        "created_at": row["created_at"],
        "last_run": row["last_run"],
        "next_run": row["next_run"],
        # v9: weekly day+time schedule
        "days": days,
        "hour": row["hour"] if "hour" in keys else None,
        "minute": row["minute"] if "minute" in keys else None,
        "scan_type": (row["scan_type"] if "scan_type" in keys else "active") or "active",
    }


def create_schedule(name, interval_minutes, target="", profile="standard",
                    options=None, enabled=True, days=None, hour=None,
                    minute=None, scan_type="active", db_path=DEFAULT_DB_PATH):
    opts_json = json.dumps(options) if options else None
    days_str = ",".join(str(int(d)) for d in days) if days else None
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "INSERT INTO schedules (name, target, profile, options, "
            "interval_minutes, enabled, created_at, days, hour, minute, scan_type) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (name.strip() or "Schedule", target.strip(), profile,
             opts_json, int(interval_minutes), 1 if enabled else 0, time.time(),
             days_str,
             int(hour) if hour is not None else None,
             int(minute) if minute is not None else None,
             scan_type or "active"),
        )
        return cur.lastrowid


def list_schedules(only_enabled=False, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        q = "SELECT * FROM schedules"
        if only_enabled:
            q += " WHERE enabled = 1"
        q += " ORDER BY id"
        rows = conn.execute(q).fetchall()
        return [_row_to_schedule(r) for r in rows]


def get_schedule(schedule_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM schedules WHERE id = ?",
                           (schedule_id,)).fetchone()
        return _row_to_schedule(row)


def update_schedule(schedule_id, db_path=DEFAULT_DB_PATH, **fields):
    """Patch allowed schedule fields. Returns True if a row was updated."""
    allowed = {"name", "target", "profile", "options", "interval_minutes",
               "enabled", "last_run", "next_run", "days", "hour", "minute",
               "scan_type"}
    sets, vals = [], []
    for k, v in fields.items():
        if k not in allowed:
            continue
        if k == "options":
            v = json.dumps(v) if v else None
        elif k == "enabled":
            v = 1 if v else 0
        elif k == "interval_minutes":
            v = int(v)
        elif k == "days":
            v = ",".join(str(int(d)) for d in v) if v else None
        elif k in ("hour", "minute"):
            v = int(v) if v is not None else None
        sets.append(f"{k} = ?")
        vals.append(v)
    if not sets:
        return False
    vals.append(schedule_id)
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            f"UPDATE schedules SET {', '.join(sets)} WHERE id = ?", vals)
        return cur.rowcount > 0


def delete_schedule(schedule_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute("DELETE FROM schedules WHERE id = ?", (schedule_id,))
        return cur.rowcount > 0


# ---------------------------------------------------------------------------
# Activity / audit log (v8) - login/logout times + user activity
# ---------------------------------------------------------------------------
def log_activity(action, user_id=None, username=None, tenant="default",
                 detail=None, ip=None, db_path=DEFAULT_DB_PATH):
    """Record a single audit event. Best-effort; never raises to the caller."""
    try:
        with _DB_LOCK, _connect(db_path) as conn:
            conn.execute(
                "INSERT INTO activity_log (ts, user_id, username, tenant, action, "
                "detail, ip) VALUES (?, ?, ?, ?, ?, ?, ?)",
                (time.time(), user_id, username, tenant or "default",
                 action, detail, ip),
            )
    except Exception:
        pass


def list_activity(limit=200, tenant=None, user_id=None, action=None,
                  db_path=DEFAULT_DB_PATH):
    """Return recent activity events, newest first, optionally scoped."""
    clauses, params = [], []
    if tenant is not None:
        clauses.append("tenant = ?")
        params.append(tenant)
    if user_id is not None:
        clauses.append("user_id = ?")
        params.append(user_id)
    if action:
        clauses.append("action = ?")
        params.append(action)
    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
    params.append(limit)
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            f"SELECT id, ts, user_id, username, tenant, action, detail, ip "
            f"FROM activity_log {where} ORDER BY id DESC LIMIT ?", params
        ).fetchall()
        return [dict(r) for r in rows]


# ---------------------------------------------------------------------------
# CVE finding lifecycle (v9.1)
# ---------------------------------------------------------------------------
LIFECYCLE_STATES = ("open", "remediated", "accepted_risk", "false_positive")


def set_cve_state(tenant, asset_id, cve, state, note=None, updated_by=None,
                  db_path=DEFAULT_DB_PATH):
    """Upsert a lifecycle disposition for a CVE on an asset ('*' => tenant-wide).
    Setting state='open' clears any prior disposition."""
    tenant = tenant or "default"
    asset_id = (asset_id or "*").strip() or "*"
    cve = (cve or "").strip().upper()
    if state not in LIFECYCLE_STATES:
        return False
    now = time.time()
    with _DB_LOCK, _connect(db_path) as conn:
        if state == "open":
            conn.execute(
                "DELETE FROM cve_lifecycle WHERE tenant=? AND asset_id=? AND cve=?",
                (tenant, asset_id, cve))
            return True
        conn.execute(
            "INSERT INTO cve_lifecycle (tenant, asset_id, cve, state, note, "
            "updated_by, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(tenant, asset_id, cve) DO UPDATE SET "
            "state=excluded.state, note=excluded.note, "
            "updated_by=excluded.updated_by, updated_at=excluded.updated_at",
            (tenant, asset_id, cve, state, note, updated_by, now))
        return True


def get_cve_states(tenant=None, db_path=DEFAULT_DB_PATH):
    """Return {(asset_id, cve): {state, note, updated_by, updated_at}}."""
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                "SELECT * FROM cve_lifecycle WHERE tenant=?", (tenant,)).fetchall()
        else:
            rows = conn.execute("SELECT * FROM cve_lifecycle").fetchall()
    out = {}
    for r in rows:
        out[(r["asset_id"], r["cve"])] = {
            "state": r["state"], "note": r["note"],
            "updated_by": r["updated_by"], "updated_at": r["updated_at"]}
    return out


def list_cve_states(tenant=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                "SELECT * FROM cve_lifecycle WHERE tenant=? ORDER BY updated_at DESC",
                (tenant,)).fetchall()
        else:
            rows = conn.execute(
                "SELECT * FROM cve_lifecycle ORDER BY updated_at DESC").fetchall()
        return [dict(r) for r in rows]


# ---------------------------------------------------------------------------
# Settings store (v9.1) - used by data-retention policy in the Admin UI
# ---------------------------------------------------------------------------
def get_setting(key, default=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT value FROM settings WHERE key=?",
                           (key,)).fetchone()
    if not row or row["value"] is None:
        return default
    try:
        return json.loads(row["value"])
    except (ValueError, TypeError):
        return row["value"]


def set_setting(key, value, db_path=DEFAULT_DB_PATH):
    payload = json.dumps(value)
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute(
            "INSERT INTO settings (key, value, updated_at) VALUES (?, ?, ?) "
            "ON CONFLICT(key) DO UPDATE SET value=excluded.value, "
            "updated_at=excluded.updated_at",
            (key, payload, time.time()))
    return True


def delete_setting(key, db_path=DEFAULT_DB_PATH):
    """Remove a single settings row. Returns True if a row was deleted."""
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute("DELETE FROM settings WHERE key=?", (key,))
        return cur.rowcount > 0


def find_settings_by_prefix(prefix, db_path=DEFAULT_DB_PATH):
    """Return {key: parsed_value} for every settings row whose key starts with
    `prefix`. Used to enumerate per-tenant records (e.g. outbound API keys)."""
    out = {}
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT key, value FROM settings WHERE key LIKE ?",
            (prefix + "%",)).fetchall()
    for row in rows:
        val = row["value"]
        if val is None:
            out[row["key"]] = None
            continue
        try:
            out[row["key"]] = json.loads(val)
        except (ValueError, TypeError):
            out[row["key"]] = val
    return out


def purge_old_scans(max_age_days=None, max_scans=None, tenant=None,
                    db_path=DEFAULT_DB_PATH):
    """Apply a data-retention policy. Deletes scans (and, via FK cascade, their
    hosts/changes) older than max_age_days and/or beyond the newest max_scans.
    Returns the number of scans removed. Asset registry + audit log are kept."""
    removed = 0
    with _DB_LOCK, _connect(db_path) as conn:
        tclause = " AND tenant=?" if tenant else ""
        tparams = (tenant,) if tenant else ()
        if max_age_days and max_age_days > 0:
            cutoff = time.time() - max_age_days * 86400
            cur = conn.execute(
                f"DELETE FROM scans WHERE started < ?{tclause}",
                (cutoff, *tparams))
            removed += cur.rowcount
        if max_scans and max_scans > 0:
            keep_ids = [r["id"] for r in conn.execute(
                f"SELECT id FROM scans WHERE 1=1{tclause} ORDER BY id DESC LIMIT ?",
                (*tparams, max_scans)).fetchall()]
            if keep_ids:
                placeholders = ",".join("?" for _ in keep_ids)
                cur = conn.execute(
                    f"DELETE FROM scans WHERE id NOT IN ({placeholders}){tclause}",
                    (*keep_ids, *tparams))
                removed += cur.rowcount
        # Prune orphaned snapshots whose scan no longer exists.
        conn.execute("DELETE FROM risk_snapshots WHERE scan_id IS NOT NULL AND "
                     "scan_id NOT IN (SELECT id FROM scans)")
    return removed


# ---------------------------------------------------------------------------
# Risk / compliance trend snapshots (v9.1)
# ---------------------------------------------------------------------------
def add_risk_snapshot(scan_id, tenant, risk_score, compliance_pct, hosts=0,
                      vulns=0, kev=0, high_risk=0, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute(
            "INSERT INTO risk_snapshots (scan_id, tenant, ts, risk_score, "
            "compliance_pct, hosts, vulns, kev, high_risk) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (scan_id, tenant or "default", time.time(), float(risk_score),
             float(compliance_pct), int(hosts), int(vulns), int(kev),
             int(high_risk)))


def get_risk_snapshots(tenant=None, since_ts=None, db_path=DEFAULT_DB_PATH):
    clauses, params = [], []
    if tenant is not None:
        clauses.append("tenant=?"); params.append(tenant)
    if since_ts is not None:
        clauses.append("ts>=?"); params.append(since_ts)
    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            f"SELECT * FROM risk_snapshots {where} ORDER BY ts ASC", params
        ).fetchall()
        return [dict(r) for r in rows]


# ---------------------------------------------------------------------------
# Service-desk integrations + tickets (v9.1)
# ---------------------------------------------------------------------------
INTEGRATION_PLATFORMS = ("jira", "servicenow", "zendesk", "cortex")


def create_integration(tenant, platform, name, base_url, auth_user=None,
                       auth_secret=None, project_key=None, enabled=True,
                       db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "INSERT INTO integrations (tenant, platform, name, base_url, "
            "auth_user, auth_secret, project_key, enabled, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (tenant or "default", platform, name, base_url, auth_user,
             auth_secret, project_key, 1 if enabled else 0, time.time()))
        return cur.lastrowid


def _row_to_integration(r, redact=True):
    d = dict(r)
    if redact:
        d["auth_secret"] = "••••••" if d.get("auth_secret") else None
    d["enabled"] = bool(d.get("enabled"))
    return d


def list_integrations(tenant=None, redact=True, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                "SELECT * FROM integrations WHERE tenant=? ORDER BY id",
                (tenant,)).fetchall()
        else:
            rows = conn.execute(
                "SELECT * FROM integrations ORDER BY id").fetchall()
        return [_row_to_integration(r, redact=redact) for r in rows]


def get_integration(integration_id, redact=False, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM integrations WHERE id=?",
                           (integration_id,)).fetchone()
        return _row_to_integration(row, redact=redact) if row else None


def update_integration(integration_id, db_path=DEFAULT_DB_PATH, **fields):
    allowed = {"name", "base_url", "auth_user", "auth_secret", "project_key",
               "enabled", "last_used", "platform"}
    sets, vals = [], []
    for k, v in fields.items():
        if k not in allowed:
            continue
        if k == "enabled":
            v = 1 if v else 0
        sets.append(f"{k} = ?"); vals.append(v)
    if not sets:
        return False
    vals.append(integration_id)
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            f"UPDATE integrations SET {', '.join(sets)} WHERE id=?", vals)
        return cur.rowcount > 0


def delete_integration(integration_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute("DELETE FROM integrations WHERE id=?",
                           (integration_id,))
        return cur.rowcount > 0


def record_ticket(tenant, integration_id, platform, asset_id, cve, external_key,
                  external_url, created_by=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "INSERT INTO tickets (tenant, integration_id, platform, asset_id, "
            "cve, external_key, external_url, created_by, created_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (tenant or "default", integration_id, platform, asset_id,
             (cve or "").upper(), external_key, external_url, created_by,
             time.time()))
        conn.execute("UPDATE integrations SET last_used=? WHERE id=?",
                     (time.time(), integration_id))
        return cur.lastrowid


def list_tickets(tenant=None, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                "SELECT * FROM tickets WHERE tenant=? ORDER BY id DESC",
                (tenant,)).fetchall()
        else:
            rows = conn.execute(
                "SELECT * FROM tickets ORDER BY id DESC").fetchall()
        return [dict(r) for r in rows]


# ---------------------------------------------------------------------------
# Asset groups (v12) - admin-defined groupings, tenant-separated
# ---------------------------------------------------------------------------
def create_asset_group(tenant, name, description=None, created_by=None,
                       db_path=DEFAULT_DB_PATH):
    """Create a group. Returns its id, or None if the name already exists
    within the tenant."""
    tenant = tenant or "default"
    name = (name or "").strip()
    if not name:
        return None
    with _DB_LOCK, _connect(db_path) as conn:
        try:
            cur = conn.execute(
                "INSERT INTO asset_groups (tenant, name, description, "
                "created_by, created_at) VALUES (?, ?, ?, ?, ?)",
                (tenant, name, (description or "").strip() or None,
                 created_by, time.time()))
            return cur.lastrowid
        except sqlite3.IntegrityError:
            return None


def list_asset_groups(tenant=None, db_path=DEFAULT_DB_PATH):
    """Groups (with member counts). Scoped to a tenant when given."""
    with _DB_LOCK, _connect(db_path) as conn:
        clause = "WHERE g.tenant=?" if tenant is not None else ""
        params = (tenant,) if tenant is not None else ()
        rows = conn.execute(
            "SELECT g.*, (SELECT COUNT(*) FROM asset_group_members m "
            "WHERE m.group_id=g.id) AS member_count "
            f"FROM asset_groups g {clause} ORDER BY g.name COLLATE NOCASE",
            params).fetchall()
        return [dict(r) for r in rows]


def get_asset_group(group_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM asset_groups WHERE id=?",
                           (group_id,)).fetchone()
        return dict(row) if row else None


def update_asset_group(group_id, db_path=DEFAULT_DB_PATH, **fields):
    allowed = {"name", "description"}
    sets, params = [], []
    for k, v in fields.items():
        if k in allowed:
            sets.append(f"{k}=?")
            params.append((v or "").strip() or None if isinstance(v, str) else v)
    if not sets:
        return False
    params.append(group_id)
    with _DB_LOCK, _connect(db_path) as conn:
        try:
            conn.execute(f"UPDATE asset_groups SET {', '.join(sets)} WHERE id=?",
                         params)
            return True
        except sqlite3.IntegrityError:
            return False


def delete_asset_group(group_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("DELETE FROM asset_group_members WHERE group_id=?",
                     (group_id,))
        conn.execute("DELETE FROM asset_groups WHERE id=?", (group_id,))
        return True


def set_asset_groups_for_asset(tenant, asset_id, group_ids,
                               db_path=DEFAULT_DB_PATH):
    """Replace the full set of group memberships for one asset (within tenant).
    Only groups that belong to the tenant are honoured."""
    tenant = tenant or "default"
    asset_id = (asset_id or "").strip()
    if not asset_id:
        return False
    wanted = {int(g) for g in (group_ids or []) if str(g).strip().isdigit()}
    now = time.time()
    with _DB_LOCK, _connect(db_path) as conn:
        valid = {r["id"] for r in conn.execute(
            "SELECT id FROM asset_groups WHERE tenant=?", (tenant,)).fetchall()}
        wanted &= valid
        conn.execute("DELETE FROM asset_group_members WHERE asset_id=? AND tenant=?",
                     (asset_id, tenant))
        for gid in wanted:
            conn.execute(
                "INSERT OR IGNORE INTO asset_group_members "
                "(group_id, asset_id, tenant, added_at) VALUES (?, ?, ?, ?)",
                (gid, asset_id, tenant, now))
    return True


def get_asset_group_map(tenant=None, db_path=DEFAULT_DB_PATH):
    """Return {asset_id: [{'id':..,'name':..}, ...]} for the tenant."""
    with _DB_LOCK, _connect(db_path) as conn:
        clause = "WHERE m.tenant=?" if tenant is not None else ""
        params = (tenant,) if tenant is not None else ()
        rows = conn.execute(
            "SELECT m.asset_id, g.id, g.name FROM asset_group_members m "
            f"JOIN asset_groups g ON g.id=m.group_id {clause} "
            "ORDER BY g.name COLLATE NOCASE", params).fetchall()
    out = {}
    for r in rows:
        out.setdefault(r["asset_id"], []).append({"id": r["id"], "name": r["name"]})
    return out


# ---------------------------------------------------------------------------
# Remediation tracking (v12) - tenant-separated
# ---------------------------------------------------------------------------
REMEDIATION_STATES = ("open", "in_progress", "blocked", "resolved")
REMEDIATION_PRIORITIES = ("low", "medium", "high", "critical")


def create_remediation(tenant, asset_id, title, detail=None, status="open",
                       priority="medium", assignee=None, due_date=None,
                       cve=None, created_by=None, db_path=DEFAULT_DB_PATH):
    tenant = tenant or "default"
    asset_id = (asset_id or "").strip()
    title = (title or "").strip()
    if not asset_id or not title:
        return None
    if status not in REMEDIATION_STATES:
        status = "open"
    if priority not in REMEDIATION_PRIORITIES:
        priority = "medium"
    now = time.time()
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "INSERT INTO remediation (tenant, asset_id, title, detail, status, "
            "priority, assignee, due_date, cve, created_by, created_at, updated_at) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            (tenant, asset_id, title, (detail or "").strip() or None, status,
             priority, (assignee or "").strip() or None,
             (due_date or "").strip() or None, (cve or "").strip().upper() or None,
             created_by, now, now))
        return cur.lastrowid


def list_remediation(tenant=None, asset_id=None, db_path=DEFAULT_DB_PATH):
    clauses, params = [], []
    if tenant is not None:
        clauses.append("tenant=?")
        params.append(tenant)
    if asset_id is not None:
        clauses.append("asset_id=?")
        params.append(asset_id)
    where = ("WHERE " + " AND ".join(clauses)) if clauses else ""
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT r.*, (SELECT COUNT(*) FROM remediation_comments c "
            "WHERE c.remediation_id = r.id) AS comment_count "
            f"FROM remediation r {where} "
            "ORDER BY (r.status='resolved'), r.updated_at DESC", params).fetchall()
        return [dict(r) for r in rows]


def get_remediation(rid, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM remediation WHERE id=?",
                           (rid,)).fetchone()
        return dict(row) if row else None


def update_remediation(rid, db_path=DEFAULT_DB_PATH, **fields):
    allowed = {"title", "detail", "status", "priority", "assignee",
               "due_date", "cve"}
    sets, params = [], []
    for k, v in fields.items():
        if k not in allowed:
            continue
        if k == "status" and v not in REMEDIATION_STATES:
            continue
        if k == "priority" and v not in REMEDIATION_PRIORITIES:
            continue
        sets.append(f"{k}=?")
        params.append((v or "").strip() or None if isinstance(v, str) else v)
    if not sets:
        return False
    sets.append("updated_at=?")
    params.append(time.time())
    params.append(rid)
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute(f"UPDATE remediation SET {', '.join(sets)} WHERE id=?",
                     params)
        return True


def delete_remediation(rid, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("DELETE FROM remediation_comments WHERE remediation_id=?", (rid,))
        conn.execute("DELETE FROM remediation WHERE id=?", (rid,))
        return True


def delete_remediations(rids, tenant=None, db_path=DEFAULT_DB_PATH):
    """Bulk-delete remediation items by id (v13 multi-select delete).

    When ``tenant`` is given, only rows belonging to that tenant are removed, so
    a scoped user can never delete another tenant's items. Returns the count of
    rows deleted."""
    ids = [int(r) for r in (rids or []) if str(r).strip().lstrip("-").isdigit()]
    if not ids:
        return 0
    qmarks = ",".join("?" for _ in ids)
    with _DB_LOCK, _connect(db_path) as conn:
        if tenant is not None:
            rows = conn.execute(
                f"SELECT id FROM remediation WHERE id IN ({qmarks}) AND tenant=?",
                (*ids, tenant)).fetchall()
            ids = [r["id"] for r in rows]
            if not ids:
                return 0
            qmarks = ",".join("?" for _ in ids)
        conn.execute(
            f"DELETE FROM remediation_comments WHERE remediation_id IN ({qmarks})", ids)
        cur = conn.execute(
            f"DELETE FROM remediation WHERE id IN ({qmarks})", ids)
        return cur.rowcount


# ---------------------------------------------------------------------------
# Remediation comments (v13) - a free-form thread per remediation item.
# ---------------------------------------------------------------------------
def add_remediation_comment(remediation_id, tenant, body, author=None,
                            db_path=DEFAULT_DB_PATH):
    body = (body or "").strip()
    if not body:
        return None
    now = time.time()
    with _DB_LOCK, _connect(db_path) as conn:
        cur = conn.execute(
            "INSERT INTO remediation_comments (remediation_id, tenant, author, "
            "body, created_at) VALUES (?, ?, ?, ?, ?)",
            (int(remediation_id), tenant or "default",
             (author or "").strip() or None, body, now))
        # Touch the parent so the item sorts as recently updated.
        conn.execute("UPDATE remediation SET updated_at=? WHERE id=?",
                     (now, int(remediation_id)))
        return cur.lastrowid


def list_remediation_comments(remediation_id, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        rows = conn.execute(
            "SELECT * FROM remediation_comments WHERE remediation_id=? "
            "ORDER BY created_at ASC, id ASC", (int(remediation_id),)).fetchall()
        return [dict(r) for r in rows]


def get_remediation_comment(cid, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        row = conn.execute("SELECT * FROM remediation_comments WHERE id=?",
                           (int(cid),)).fetchone()
        return dict(row) if row else None


def delete_remediation_comment(cid, db_path=DEFAULT_DB_PATH):
    with _DB_LOCK, _connect(db_path) as conn:
        conn.execute("DELETE FROM remediation_comments WHERE id=?", (int(cid),))
        return True