"""Mesin deteksi: menerima peristiwa dari sensor, menjalankan aturan, menyimpan temuan & peringatan.""" from __future__ import annotations import logging import threading from collections import defaultdict, deque from datetime import datetime, timedelta from typing import Any from . import rules from .redact import redact from .repository import Repository, new_incident_code, utcnow from .rules import Finding, RISK_ORDER log = logging.getLogger("csm.engine") ALERT_MIN_RISK = "high" # temuan >= high otomatis membuat alert DEDUP_MINUTES = 10 # cegah banjir peringatan untuk kategori+IP yang sama class Engine: def __init__(self, repo: Repository, clock=utcnow): self.repo = repo self.clock = clock self._lock = threading.Lock() self._port_hits: dict[str, deque] = defaultdict(lambda: deque(maxlen=2000)) self._req_per_min: dict[datetime, int] = defaultdict(int) self._sensor_alerted: set[int] = set() # ---------- util ---------- def _record(self, f: Finding, now: datetime, source: str, sensor_id: int | None, evidence: str = "", dedup: bool = True) -> int | None: if dedup and self.repo.recent_event_exists(f.category, f.ip, now - timedelta(minutes=DEDUP_MINUTES)): return None event_id = self.repo.insert_event({ "incident_code": new_incident_code(now), "event_time": now, "event_type": f.category, "source": source, "ip_address": f.ip, "risk_level": f.risk, "summary": f.summary, "evidence": redact(evidence), "recommendation": f.recommendation, "sensor_id": sensor_id}) if f.category in ("sql_injection", "xss", "path_traversal"): self.repo.insert_attack({ "event_id": event_id, "attack_type": f.category, "endpoint": f.endpoint, "indicator": ", ".join(f.indicators), "detected_at": now, "verification_status": f.verification}) if RISK_ORDER[f.risk] >= RISK_ORDER[ALERT_MIN_RISK]: self.repo.insert_alert({"event_id": event_id, "priority": f.risk, "message": f.summary}) return event_id def _log_info(self, etype: str, summary: str, now: datetime, source: str, sensor_id: int | None, ip: str | None = None, risk: str = "normal", evidence: str = "") -> None: self.repo.insert_event({ "incident_code": new_incident_code(now), "event_time": now, "event_type": etype, "source": source, "ip_address": ip, "risk_level": risk, "summary": redact(summary), "evidence": redact(evidence), "recommendation": None, "sensor_id": sensor_id}) # ---------- ingest ---------- def process_batch(self, sensor: dict, events: list[dict[str, Any]]) -> dict: now = self.clock() settings = self.repo.load_settings() self.repo.touch_sensor(sensor["id"], sensor.get("hostname"), sensor.get("os"), now) self._sensor_alerted.discard(sensor["id"]) stats = {"processed": 0, "findings": 0} for ev in events: try: stats["findings"] += self._handle(sensor, ev, settings, now) stats["processed"] += 1 except Exception: log.exception("Gagal memproses event tipe=%s", ev.get("type")) return stats def _handle(self, sensor: dict, ev: dict, settings: dict, now: datetime) -> int: kind = ev.get("type") sid, src = sensor["id"], sensor["sensor_code"] ip = ev.get("ip") n = 0 if kind == "http_request": self._count_request(now) for f in rules.analyze_http_request(ev.get("path", ""), ev.get("query", ""), ev.get("body", ""), ip): ev_line = f'{ev.get("method", "GET")} {ev.get("path", "")}?{ev.get("query", "")} {ev.get("body", "")}' n += 1 if self._record(f, now, src, sid, ev_line) else 0 elif kind == "login": ok = bool(ev.get("success")) self.repo.insert_login({"username": ev.get("username", "?"), "ip_address": ip, "success": ok, "user_agent": ev.get("user_agent"), "source": src, "created_at": now}) if not ok: since = now - timedelta(minutes=settings["brute_force_window_min"]) cnt = self.repo.count_failed_logins(ip, None, since) f = rules.check_brute_force(cnt, int(settings["brute_force_threshold"]), ip, int(settings["brute_force_window_min"]), ev.get("username")) if f: n += 1 if self._record(f, now, src, sid, f"user={ev.get('username')}") else 0 elif kind == "firewall": if ip and ev.get("dst_port") is not None: with self._lock: self._port_hits[ip].append((int(ev["dst_port"]), now)) recs = list(self._port_hits[ip]) f = rules.check_port_scan(recs, ip, now, int(settings["port_scan_threshold"]), int(settings["port_scan_window_sec"])) if f: n += 1 if self._record(f, now, src, sid, f"dst_port={ev['dst_port']}") else 0 elif kind == "metrics": m = ev.get("metrics") or {} svc = ev.get("services") self.repo.insert_metric({"sensor_id": sid, "cpu_percent": m.get("cpu"), "mem_percent": m.get("mem"), "disk_percent": m.get("disk"), "services_json": redact(str(svc), 1000) if svc else None, "collected_at": now}) f = rules.check_system_metrics(m.get("cpu"), m.get("mem"), m.get("disk")) if f: n += 1 if self._record(f, now, src, sid) else 0 elif kind == "file_integrity": for fe in (ev.get("files") or [])[:200]: n += self._handle_file(sid, src, fe.get("path", ""), fe.get("hash"), now) elif kind == "syslog": msg = ev.get("message", "") risk = rules.classify_syslog(msg) self._log_info("system", f"[{ev.get('os', 'syslog')}] {msg}", now, src, sid, ip, risk, msg) elif kind == "admin_action": f = rules.check_admin_activity(ev.get("action", "?"), now, int(settings["work_hours_start"]), int(settings["work_hours_end"]), ip, self.repo.admin_known_ips() or None) if f: n += 1 if self._record(f, now, src, sid, ev.get("action", "")) else 0 elif kind == "heartbeat": pass return n def _handle_file(self, sid: int, src: str, path: str, current: str | None, now: datetime) -> int: last = self.repo.last_file_row(sid, path) baseline = last["baseline_hash"] if last else None status, f = rules.check_file_integrity(path, baseline, current) # baseline tetap hash lama sampai admin menyetujui (aksi "fim_accept" di dashboard) self.repo.insert_fim({"sensor_id": sid, "file_path": path, "baseline_hash": baseline if baseline else current, "current_hash": current, "status": status, "checked_at": now}) # alert hanya saat perubahan BARU (hash/status berbeda dari pemeriksaan sebelumnya), bukan setiap siklus already_reported = bool(last) and last["status"] == status and last["current_hash"] == current if f and not already_reported: f.endpoint = path return 1 if self._record(f, now, src, sid, path, dedup=False) else 0 return 0 def _count_request(self, now: datetime) -> None: minute = now.replace(second=0, microsecond=0) with self._lock: self._req_per_min[minute] += 1 for old in [k for k in self._req_per_min if k < minute - timedelta(minutes=60)]: del self._req_per_min[old] # ---------- pemeriksaan terjadwal ---------- def run_scheduled(self) -> dict: now = self.clock() s = self.repo.load_settings() out = {"sensor_offline": 0, "brute_force": 0, "traffic_spike": 0} for sensor in self.repo.stale_sensors(now, int(s["sensor_offline_sec"])): self.repo.set_sensor_status(sensor["id"], "offline") if sensor["id"] not in self._sensor_alerted: self._sensor_alerted.add(sensor["id"]) f = Finding("sensor_failure", "high", f"Sensor '{sensor['sensor_code']}' berhenti mengirim data", [f"last_seen={sensor.get('last_seen_at')}"], "Periksa layanan sensor, koneksi jaringan, dan kunci API sensor.") self._record(f, now, "engine", sensor["id"], dedup=False) out["sensor_offline"] += 1 since = now - timedelta(minutes=int(s["brute_force_window_min"])) for g in self.repo.failed_login_groups(since, int(s["brute_force_threshold"])): f = rules.check_brute_force(int(g["n"]), int(s["brute_force_threshold"]), g["ip_address"], int(s["brute_force_window_min"])) if f and self._record(f, now, "engine", None, "login_logs"): out["brute_force"] += 1 with self._lock: minute = now.replace(second=0, microsecond=0) - timedelta(minutes=1) # menit terakhir yang selesai current = self._req_per_min.get(minute, 0) history = [c for k, c in self._req_per_min.items() if k < minute] f = rules.check_traffic_spike(current, history, float(s["traffic_spike_factor"]), int(s["traffic_spike_min"])) if f and self._record(f, now, "engine", None): out["traffic_spike"] += 1 return out def purge(self) -> int: s = self.repo.load_settings() return self.repo.purge_older_than(int(s.get("retention_days", 90)), self.clock())