# SPDX-License-Identifier: GPL-3.0-or-later
"""Alert triage — label a detection likely-false-positive vs needs-review.

Applies cheap, explainable heuristics (provenance, loopback binds, allowlists,
drill artifacts) so a human isn't drowned in benign noise. Verdicts are advice,
never auto-deletion. ``is_owned`` is injectable for testing.
"""
from __future__ import annotations

import re
import json
from collections.abc import Callable
from dataclasses import dataclass

from .config import Config
from .provenance import is_package_owned

LIKELY_FP = "likely-fp"
REVIEW = "review"

_LOOPBACK = ("127.", "::1", "[::1]")
_WRITABLE = ("/tmp/", "/dev/shm/", "/var/tmp/", "/run/user/")
# Signatures that provenance can never clear — a network shell or C2 channel is
# worth a look regardless of which binary is involved.
_ALWAYS_REVIEW = {"reverse_shell", "egress"}


@dataclass(frozen=True)
class Verdict:
    label: str           # LIKELY_FP or REVIEW
    reason: str
    suggest: str = ""    # optional TOML snippet to suppress future copies


def _exe_for(alert: dict, processes: list[dict]) -> str:
    pids = set(alert.get("pids", []))
    for p in processes:
        if p.get("pid") in pids:
            return p.get("exe", "")
    return ""


def _comm_from_detail(detail: str) -> str:
    m = re.search(r"\bcomm=([^\s]+)", detail)
    return m.group(1) if m else ""


def _alert_context(alert: dict) -> str:
    sid = alert.get("sid") or "unknown"
    classtype = alert.get("classtype") or "unknown"
    return f"sid={sid} classtype={classtype}"


def _toml_string_array(name: str, values) -> str:
    items = ", ".join(json.dumps(v) for v in sorted(set(values)) if v)
    return f"{name} = [{items}]"


def _allow_comm_suggestion(alert: dict, cfg: Config, key: str, comm: str,
                           why: str) -> str:
    current = getattr(cfg, key)
    return "\n".join((
        f"# Triage suggestion for {_alert_context(alert)}.",
        f"# Add only after verifying this {comm!r} activity is expected: {why}.",
        _toml_string_array(key, set(current) | {comm}),
    ))


def triage_alert(alert: dict, processes: list[dict], cfg: Config,
                 is_owned: Callable[[str], bool] = is_package_owned) -> Verdict:
    sig = alert.get("signature", "")
    detail = alert.get("detail", "")
    exe = _exe_for(alert, processes)

    if sig.startswith("exec_rule.") or sig in _ALWAYS_REVIEW:
        return Verdict(REVIEW, "network/exec signature — provenance can't clear it")

    if sig == "ld_preload":
        m = re.search(r"LD_PRELOAD=\[([^\]]+)\]", detail)
        lib = m.group(1) if m else ""
        if lib.startswith(_WRITABLE):
            return Verdict(REVIEW, f"LD_PRELOAD from writable path: {lib}")
        if lib and is_owned(lib):
            return Verdict(LIKELY_FP, f"preload library is package-owned: {lib}")
        return Verdict(REVIEW, "library injection — verify the .so by hand")

    if sig == "process_injection_library":
        return Verdict(REVIEW, "executable library mapped from writable path")

    if sig == "deleted_exe":
        if "memfd:" in detail:
            return Verdict(REVIEW, "memfd-backed (fileless) execution")
        if exe and is_owned(exe):
            return Verdict(LIKELY_FP, "deleted exe is package-owned (post-upgrade daemon)")
        return Verdict(REVIEW, "process running from a deleted binary")

    if sig == "new_listener":
        addr = detail.split("socket ", 1)[-1].split(" by ", 1)[0]
        comm = detail.rsplit("comm=", 1)[-1].strip()
        port = addr.rsplit(":", 1)[-1]
        if addr.startswith(_LOOPBACK):
            return Verdict(LIKELY_FP, "loopback-only listener (not externally reachable)")
        if comm in cfg.listener_allow_comms:
            return Verdict(LIKELY_FP, f"comm '{comm}' is allow-listed")
        if exe and is_owned(exe):
            return Verdict(LIKELY_FP,
                           f"listener binary is package-owned ({comm})",
                           suggest=_allow_comm_suggestion(
                               alert, cfg, "listener_allow_comms", comm,
                               "package-owned listener"))
        return Verdict(REVIEW, f"unrecognized listener {addr} ({comm})",
                       suggest=_allow_comm_suggestion(
                           alert, cfg, "listener_allow_comms", comm,
                           "known service listener"))

    if sig == "new_suid":
        path = detail.rsplit(": ", 1)[-1]
        if is_owned(path):
            return Verdict(LIKELY_FP, "SUID binary is package-owned")
        return Verdict(REVIEW, f"unowned SUID binary: {path}")

    if sig == "input_snooper":
        comm = _comm_from_detail(detail)
        if comm:
            return Verdict(REVIEW, "input device access — verify this is local input stack",
                           suggest=_allow_comm_suggestion(
                               alert, cfg, "input_snooper_allow_comms", comm,
                               "expected input-device reader"))
        return Verdict(REVIEW, "input device access — verify this is local input stack")

    if sig == "credential_access":
        comm = _comm_from_detail(detail)
        if comm:
            return Verdict(REVIEW, "credential file access — verify the process role",
                           suggest=_allow_comm_suggestion(
                               alert, cfg, "credential_access_allow_comms", comm,
                               "expected credential-store reader"))
        return Verdict(REVIEW, "credential file access — verify the process role")

    if sig == "stealth_network":
        comm = _comm_from_detail(detail)
        if comm and comm != "?":
            return Verdict(REVIEW, "unusual socket family — verify the protocol use",
                           suggest=_allow_comm_suggestion(
                               alert, cfg, "stealth_network_allow_comms", comm,
                               "expected raw/special-protocol socket owner"))
        return Verdict(REVIEW, "unusual socket family — verify the protocol use")

    if sig == "memory_obfuscation":
        comm = _comm_from_detail(detail)
        if comm:
            return Verdict(REVIEW, "executable or RWX memory mapping — verify runtime/JIT use",
                           suggest=_allow_comm_suggestion(
                               alert, cfg, "memory_obfuscation_allow_comms", comm,
                               "expected executable memory runtime"))
        return Verdict(REVIEW, "executable or RWX memory mapping — verify runtime/JIT use")

    if sig == "persistence":
        return Verdict(REVIEW, "a persistence-relevant file changed — confirm intent")

    return Verdict(REVIEW, "no triage rule matched")