161 lines
6.6 KiB
Python
161 lines
6.6 KiB
Python
# 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")
|