210 lines
6.6 KiB
Python
210 lines
6.6 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Post-alert enrichment for captured snapshots.
|
|
|
|
This module stays cheap: it annotates already-captured process, file, network,
|
|
and integrity context without running heavyweight package verification or
|
|
rootcheck sweeps during snapshot capture.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import os
|
|
import re
|
|
import stat
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
from .config import Config
|
|
from .fim import hash_file
|
|
from .netutil import is_public_ip
|
|
from .provenance import package_owner
|
|
from .system import SystemState
|
|
|
|
_IP_RE = re.compile(r"(?<![\w.])(?:\d{1,3}\.){3}\d{1,3}(?![\w.])")
|
|
|
|
|
|
def build(
|
|
report: dict,
|
|
state: SystemState,
|
|
cfg: Config,
|
|
*,
|
|
lineage: set[int] | None = None,
|
|
recent_files: list[str] | None = None,
|
|
owner: Callable[[str], str | None] = package_owner,
|
|
) -> dict:
|
|
alerts = report.get("alerts", [])
|
|
return {
|
|
"processes": [_process_context(state, proc, cfg, owner)
|
|
for proc in report.get("processes", [])],
|
|
"remotes": [_remote_context(ip) for ip in sorted(_remote_ips(alerts))],
|
|
"files": [_file_context(path, owner) for path in sorted(_paths(alerts))],
|
|
"recent_writes": sorted(set(recent_files or [])),
|
|
"lineage": _lineage_context(state, sorted(lineage or set())),
|
|
"integrity": _integrity_context(cfg),
|
|
}
|
|
|
|
|
|
def format_text(enrichment: dict) -> str:
|
|
lines: list[str] = []
|
|
for proc in enrichment.get("processes", []):
|
|
lines.append(
|
|
"process pid={pid} comm={comm} package={pkg} sha256={sha}".format(
|
|
pid=proc.get("pid"),
|
|
comm=proc.get("comm") or "?",
|
|
pkg=proc.get("package_owner") or "unowned/unknown",
|
|
sha=(proc.get("exe_sha256") or "-")[:16],
|
|
)
|
|
)
|
|
chain = " <- ".join(
|
|
f"{p.get('pid')}:{p.get('comm') or '?'}"
|
|
for p in proc.get("parent_chain", [])
|
|
)
|
|
if chain:
|
|
lines.append(f" parent chain: {chain}")
|
|
for remote in enrichment.get("remotes", []):
|
|
lines.append(f"remote {remote['ip']}: {remote['classification']}")
|
|
for f in enrichment.get("files", []):
|
|
mode = f"0{f['mode']:o}" if f.get("mode") is not None else "-"
|
|
lines.append(
|
|
f"file {f['path']}: exists={f['exists']} mode={mode} "
|
|
f"owner={f.get('package_owner') or 'unowned/unknown'}"
|
|
)
|
|
integrity = enrichment.get("integrity", {})
|
|
if integrity:
|
|
lines.append(
|
|
"integrity: fim_baseline={fim} pkgdb_anchor={pkg} "
|
|
"rootcheck={root}".format(
|
|
fim=integrity.get("fim_baseline", {}).get("status", "unknown"),
|
|
pkg=integrity.get("pkgdb_anchor", {}).get("status", "unknown"),
|
|
root="enabled" if integrity.get("rootcheck", {}).get("enabled") else "disabled",
|
|
)
|
|
)
|
|
return "\n".join(lines) if lines else " (no enrichment available)"
|
|
|
|
|
|
def _process_context(
|
|
state: SystemState,
|
|
proc: dict,
|
|
cfg: Config,
|
|
owner: Callable[[str], str | None],
|
|
) -> dict:
|
|
exe = _clean_path(proc.get("exe", ""))
|
|
return {
|
|
"pid": proc.get("pid"),
|
|
"comm": proc.get("comm", ""),
|
|
"exe": exe,
|
|
"package_owner": owner(exe) if exe else None,
|
|
"exe_sha256": hash_file(exe) if exe and os.path.isfile(exe) else None,
|
|
"parent_chain": _parent_chain(state, proc.get("ppid", 0),
|
|
cfg.incident_lineage_depth),
|
|
}
|
|
|
|
|
|
def _parent_chain(state: SystemState, ppid: int, depth: int) -> list[dict]:
|
|
chain: list[dict] = []
|
|
seen: set[int] = set()
|
|
cur = ppid
|
|
for _ in range(max(depth, 0)):
|
|
if not cur or cur in seen:
|
|
break
|
|
seen.add(cur)
|
|
proc = state.process(cur)
|
|
if proc is None:
|
|
chain.append({"pid": cur, "comm": "", "ppid": 0})
|
|
break
|
|
chain.append({"pid": cur, "comm": proc.comm, "ppid": proc.ppid})
|
|
cur = proc.ppid
|
|
return chain
|
|
|
|
|
|
def _lineage_context(state: SystemState, lineage: list[int]) -> list[dict]:
|
|
out = []
|
|
for pid in lineage:
|
|
proc = state.process(pid)
|
|
out.append({
|
|
"pid": pid,
|
|
"comm": proc.comm if proc else "",
|
|
"present": proc is not None,
|
|
})
|
|
return out
|
|
|
|
|
|
def _remote_ips(alerts: list[dict]) -> set[str]:
|
|
out: set[str] = set()
|
|
for alert in alerts:
|
|
for ip in _IP_RE.findall(alert.get("detail", "")):
|
|
out.add(ip)
|
|
return out
|
|
|
|
|
|
def _remote_context(ip: str) -> dict:
|
|
try:
|
|
parsed = ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return {"ip": ip, "classification": "invalid"}
|
|
if is_public_ip(ip):
|
|
classification = "public"
|
|
elif parsed.is_loopback:
|
|
classification = "loopback"
|
|
elif parsed.is_private:
|
|
classification = "private"
|
|
elif parsed.is_link_local:
|
|
classification = "link-local"
|
|
else:
|
|
classification = "reserved"
|
|
return {"ip": ip, "classification": classification}
|
|
|
|
|
|
def _paths(alerts: list[dict]) -> set[str]:
|
|
paths: set[str] = set()
|
|
for alert in alerts:
|
|
for text in (alert.get("detail", ""), alert.get("key", "")):
|
|
for token in re.findall(r"(?<![\w.])/[^\s,\])]+", text):
|
|
paths.add(_clean_path(token.rstrip(":")))
|
|
return {p for p in paths if p}
|
|
|
|
|
|
def _file_context(path: str, owner: Callable[[str], str | None]) -> dict:
|
|
p = Path(path)
|
|
try:
|
|
st = p.lstat()
|
|
except OSError:
|
|
return {"path": path, "exists": False, "mode": None, "uid": None,
|
|
"gid": None, "package_owner": None}
|
|
return {
|
|
"path": path,
|
|
"exists": True,
|
|
"mode": stat.S_IMODE(st.st_mode),
|
|
"uid": st.st_uid,
|
|
"gid": st.st_gid,
|
|
"package_owner": owner(path),
|
|
}
|
|
|
|
|
|
def _integrity_context(cfg: Config) -> dict:
|
|
return {
|
|
"fim_baseline": _path_status(cfg.fim_baseline),
|
|
"pkgdb_anchor": _path_status(cfg.log_dir / "pkgdb-anchor.json"),
|
|
"package_verify": {
|
|
"enabled": bool(cfg.pkgdb_pkgverify),
|
|
"sample": cfg.pkgdb_pkgverify_sample,
|
|
},
|
|
"rootcheck": {
|
|
"enabled": bool(cfg.rootcheck_enabled),
|
|
"interval": cfg.rootcheck_interval,
|
|
},
|
|
}
|
|
|
|
|
|
def _path_status(path: Path) -> dict:
|
|
try:
|
|
st = path.stat()
|
|
except OSError:
|
|
return {"path": str(path), "status": "missing", "age": None}
|
|
return {"path": str(path), "status": "present",
|
|
"age": max(0, int(time.time() - st.st_mtime))}
|
|
|
|
|
|
def _clean_path(path: str) -> str:
|
|
return path.replace(" (deleted)", "")
|