"""Lapisan penyimpanan. MySQLRepository dipakai produksi; InMemoryRepository dipakai pengujian.""" from __future__ import annotations import secrets from datetime import datetime, timedelta, timezone from typing import Any, Protocol from .rules import DEFAULT_SETTINGS def utcnow() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None, microsecond=0) def new_incident_code(now: datetime | None = None) -> str: now = now or utcnow() return f"INC-{now:%Y%m%d}-{secrets.token_hex(3).upper()}" class Repository(Protocol): def load_settings(self) -> dict[str, Any]: ... def get_sensor(self, code: str) -> dict | None: ... def touch_sensor(self, sensor_id: int, hostname: str | None, os_name: str | None, now: datetime) -> None: ... def set_sensor_status(self, sensor_id: int, status: str) -> None: ... def stale_sensors(self, now: datetime, offline_sec: int) -> list[dict]: ... def insert_event(self, ev: dict) -> int: ... def insert_attack(self, row: dict) -> None: ... def insert_login(self, row: dict) -> None: ... def insert_metric(self, row: dict) -> None: ... def last_file_row(self, sensor_id: int, path: str) -> dict | None: ... def insert_fim(self, row: dict) -> None: ... def insert_alert(self, row: dict) -> int: ... def count_failed_logins(self, ip: str | None, username: str | None, since: datetime) -> int: ... def failed_login_groups(self, since: datetime, threshold: int) -> list[dict]: ... def recent_event_exists(self, event_type: str, ip: str | None, since: datetime) -> bool: ... def admin_known_ips(self) -> set[str]: ... def purge_older_than(self, days: int, now: datetime) -> int: ... def unemailed_alerts(self) -> list[dict]: ... def mark_emailed(self, alert_id: int) -> None: ... class InMemoryRepository: """Implementasi sederhana untuk unit/integration test (tanpa database).""" def __init__(self, sensors: list[dict] | None = None, settings: dict | None = None): self.sensors = {s["id"]: dict(s) for s in (sensors or [])} self.settings = {**DEFAULT_SETTINGS, **(settings or {})} self.events: list[dict] = [] self.attacks: list[dict] = [] self.logins: list[dict] = [] self.metrics: list[dict] = [] self.fim: list[dict] = [] self.alerts: list[dict] = [] def load_settings(self): return dict(self.settings) def get_sensor(self, code): return next((dict(s) for s in self.sensors.values() if s["sensor_code"] == code), None) def touch_sensor(self, sensor_id, hostname, os_name, now): s = self.sensors[sensor_id] s.update(last_seen_at=now, status="online") if hostname: s["hostname"] = hostname if os_name: s["os"] = os_name def set_sensor_status(self, sensor_id, status): self.sensors[sensor_id]["status"] = status def stale_sensors(self, now, offline_sec): return [dict(s) for s in self.sensors.values() if s.get("status") != "offline" and ( s.get("last_seen_at") is None or (now - s["last_seen_at"]).total_seconds() > offline_sec) and s.get("last_seen_at") is not None] def insert_event(self, ev): ev = dict(ev); ev["id"] = len(self.events) + 1; self.events.append(ev); return ev["id"] def insert_attack(self, row): self.attacks.append(dict(row)) def insert_login(self, row): self.logins.append(dict(row)) def insert_metric(self, row): self.metrics.append(dict(row)) def last_file_row(self, sensor_id, path): rows = [r for r in self.fim if r["sensor_id"] == sensor_id and r["file_path"] == path] return dict(rows[-1]) if rows else None def insert_fim(self, row): self.fim.append(dict(row)) def insert_alert(self, row): row = dict(row); row["id"] = len(self.alerts) + 1; row.setdefault("emailed", 0); self.alerts.append(row) return row["id"] def count_failed_logins(self, ip, username, since): return sum(1 for r in self.logins if not r["success"] and r["created_at"] >= since and ((ip and r["ip_address"] == ip) or (username and r["username"] == username))) def failed_login_groups(self, since, threshold): counts: dict[str | None, int] = {} for r in self.logins: if not r["success"] and r["created_at"] >= since: counts[r["ip_address"]] = counts.get(r["ip_address"], 0) + 1 return [{"ip_address": ip, "n": n} for ip, n in counts.items() if n >= threshold] def recent_event_exists(self, event_type, ip, since): return any(e["event_type"] == event_type and e.get("ip_address") == ip and e["event_time"] >= since for e in self.events) def admin_known_ips(self): return set() def purge_older_than(self, days, now): return 0 def unemailed_alerts(self): return [a for a in self.alerts if not a.get("emailed")] def mark_emailed(self, alert_id): self.alerts[alert_id - 1]["emailed"] = 1 class MySQLRepository: def __init__(self, host: str, port: int, user: str, password: str, database: str): import pymysql # import lambat agar test tidak butuh koneksi self._pymysql = pymysql self._kw = dict(host=host, port=port, user=user, password=password, database=database, charset="utf8mb4", autocommit=True, connect_timeout=5, init_command="SET time_zone='+00:00'", cursorclass=pymysql.cursors.DictCursor) def _q(self, sql: str, args: tuple = (), fetch: str | None = None, last_id: bool = False): conn = self._pymysql.connect(**self._kw) try: with conn.cursor() as cur: cur.execute(sql, args) if fetch == "one": return cur.fetchone() if fetch == "all": return list(cur.fetchall()) return cur.lastrowid if last_id else cur.rowcount finally: conn.close() def load_settings(self): out = dict(DEFAULT_SETTINGS) for r in self._q("SELECT setting_key, setting_value FROM settings", fetch="all"): k, v = r["setting_key"], r["setting_value"] if k in DEFAULT_SETTINGS: try: num = float(v) out[k] = num if isinstance(DEFAULT_SETTINGS[k], float) else int(num) except ValueError: pass else: out[k] = v return out def get_sensor(self, code): return self._q("SELECT * FROM sensor_status WHERE sensor_code=%s", (code,), "one") def touch_sensor(self, sensor_id, hostname, os_name, now): self._q("UPDATE sensor_status SET last_seen_at=%s, status='online', " "hostname=COALESCE(NULLIF(%s,''),hostname), os=COALESCE(NULLIF(%s,''),os) WHERE id=%s", (now, hostname or "", os_name or "", sensor_id)) def set_sensor_status(self, sensor_id, status): self._q("UPDATE sensor_status SET status=%s WHERE id=%s", (status, sensor_id)) def stale_sensors(self, now, offline_sec): return self._q("SELECT * FROM sensor_status WHERE status<>'offline' AND last_seen_at IS NOT NULL " "AND last_seen_at < %s", (now - timedelta(seconds=offline_sec),), "all") def insert_event(self, ev): return self._q( "INSERT INTO security_events (incident_code,event_time,event_type,source,ip_address,risk_level," "summary,evidence,recommendation,sensor_id) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)", (ev["incident_code"], ev["event_time"], ev["event_type"], ev["source"], ev.get("ip_address"), ev["risk_level"], ev["summary"][:500], (ev.get("evidence") or None), (ev.get("recommendation") or None), ev.get("sensor_id")), last_id=True) def insert_attack(self, r): self._q("INSERT INTO attack_attempts (event_id,attack_type,endpoint,indicator,detected_at,verification_status)" " VALUES (%s,%s,%s,%s,%s,%s)", (r["event_id"], r["attack_type"], r.get("endpoint"), r["indicator"][:255], r["detected_at"], r.get("verification_status", "suspected"))) def insert_login(self, r): self._q("INSERT INTO login_logs (username,ip_address,success,user_agent,source,created_at) " "VALUES (%s,%s,%s,%s,%s,%s)", (r["username"][:100], r.get("ip_address"), int(r["success"]), (r.get("user_agent") or "")[:255], r.get("source", "sensor"), r["created_at"])) def insert_metric(self, r): self._q("INSERT INTO system_metrics (sensor_id,cpu_percent,mem_percent,disk_percent,services_json,collected_at)" " VALUES (%s,%s,%s,%s,%s,%s)", (r["sensor_id"], r.get("cpu_percent"), r.get("mem_percent"), r.get("disk_percent"), r.get("services_json"), r["collected_at"])) def last_file_row(self, sensor_id, path): return self._q("SELECT baseline_hash, current_hash, status FROM file_integrity_logs " "WHERE sensor_id=%s AND file_path=%s ORDER BY id DESC LIMIT 1", (sensor_id, path), "one") def insert_fim(self, r): self._q("INSERT INTO file_integrity_logs (sensor_id,file_path,baseline_hash,current_hash,status,checked_at)" " VALUES (%s,%s,%s,%s,%s,%s)", (r["sensor_id"], r["file_path"][:500], r.get("baseline_hash"), r.get("current_hash"), r["status"], r["checked_at"])) def insert_alert(self, r): return self._q("INSERT INTO alerts (event_id,priority,message) VALUES (%s,%s,%s)", (r.get("event_id"), r["priority"], r["message"][:500]), last_id=True) def count_failed_logins(self, ip, username, since): row = self._q("SELECT COUNT(*) AS n FROM login_logs WHERE success=0 AND created_at>=%s " "AND ((%s IS NOT NULL AND ip_address=%s) OR (%s IS NOT NULL AND username=%s))", (since, ip, ip, username, username), "one") return int(row["n"]) def failed_login_groups(self, since, threshold): return self._q("SELECT ip_address, COUNT(*) AS n FROM login_logs WHERE success=0 AND created_at>=%s " "GROUP BY ip_address HAVING n>=%s", (since, threshold), "all") def recent_event_exists(self, event_type, ip, since): row = self._q("SELECT 1 AS x FROM security_events WHERE event_type=%s AND event_time>=%s " "AND ((%s IS NULL AND ip_address IS NULL) OR ip_address=%s) LIMIT 1", (event_type, since, ip, ip), "one") return bool(row) def admin_known_ips(self): rows = self._q("SELECT DISTINCT ip_address FROM audit_logs WHERE ip_address IS NOT NULL", fetch="all") return {r["ip_address"] for r in rows} def purge_older_than(self, days, now): cutoff = now - timedelta(days=days) n = 0 for table, col in (("login_logs", "created_at"), ("system_metrics", "collected_at"), ("file_integrity_logs", "checked_at"), ("security_events", "event_time")): n += self._q(f"DELETE FROM {table} WHERE {col}<%s", (cutoff,)) return n def unemailed_alerts(self): return self._q("SELECT * FROM alerts WHERE emailed=0 AND priority IN ('high','critical') " "ORDER BY id LIMIT 20", fetch="all") def mark_emailed(self, alert_id): self._q("UPDATE alerts SET emailed=1 WHERE id=%s", (alert_id,))