diff --git a/nixos/etc/anubis-watchdog/watchdog.py b/nixos/etc/anubis-watchdog/watchdog.py new file mode 100644 index 0000000..1f27050 --- /dev/null +++ b/nixos/etc/anubis-watchdog/watchdog.py @@ -0,0 +1,305 @@ +#!/usr/bin/env python3 +import os +import sys +import json +import subprocess +import threading +import time +import sqlite3 +import hashlib +import logging +import ipaddress +import re + +# --- CONFIG --- +STATE_PATH = "/var/lib/anubis-watchdog/state.json" +DB_PATH = "/var/lib/anubis-watchdog/reports.db" +SLEEP_REFRESH = 8 +DUPLICATE_TTL_SECONDS = 24 * 3600 +REPORT_FLUSH_INTERVAL = 5 # seconds +INMEM_COOLDOWN = 600 # 10 minutes per IP + +DOCKER_BIN = os.environ.get("DOCKER_BIN", "/run/current-system/sw/bin/docker") +ABUSEIPDB_KEY_FILE = os.environ.get("ABUSEIPDB_KEY_FILE", "/etc/anubis-watchdog/abuseipdb.key") +CURL_BIN = os.environ.get("CURL_BIN", "/run/current-system/sw/bin/curl") + +IP_RE = re.compile(r'\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b') + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + +# --- DB / Directories --- +def ensure_dirs(): + os.makedirs(os.path.dirname(STATE_PATH), exist_ok=True) + os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) + +def init_db(): + conn = sqlite3.connect(DB_PATH, timeout=10, check_same_thread=False) + cur = conn.cursor() + cur.execute(""" + CREATE TABLE IF NOT EXISTS reports ( + ip TEXT, + logsig TEXT, + ts INTEGER, + PRIMARY KEY (ip, logsig) + ) + """) + conn.commit() + return conn + +# --- Deduplication --- +_recent_reports = {} +_recent_lock = threading.Lock() + +def signature_for_log(container, ip, raw_log): + """Hash IP + container + first 400 chars of log to ignore timestamps.""" + m = hashlib.sha256() + core = f"{container}:{ip}" + core_part = raw_log[:400] + m.update(core.encode()) + m.update(core_part.encode()) + return m.hexdigest() + +def already_reported(conn, ip, logsig): + now = time.time() + with _recent_lock: + if ip in _recent_reports and now - _recent_reports[ip] < INMEM_COOLDOWN: + return True + + cur = conn.cursor() + cur.execute("SELECT ts FROM reports WHERE ip=? AND logsig=?", (ip, logsig)) + row = cur.fetchone() + if row and (now - row[0]) < DUPLICATE_TTL_SECONDS: + return True + return False + +def mark_reported(conn, ip, logsig): + now = int(time.time()) + with _recent_lock: + _recent_reports[ip] = now + cur = conn.cursor() + cur.execute( + "INSERT OR REPLACE INTO reports(ip, logsig, ts) VALUES(?,?,?)", + (ip, logsig, now), + ) + conn.commit() + +# --- IP extraction --- +def ip_is_public(ip_str: str) -> bool: + try: + a = ipaddress.ip_address(ip_str) + if a.is_private or a.is_loopback or a.is_unspecified or a.is_multicast: + return False + if hasattr(a, "is_global") and not a.is_global: + return False + return True + except Exception: + return False + +def extract_ips_from_text(line: str): + cand = IP_RE.findall(line) + return [ip for ip in cand if ip_is_public(ip)] + +def pick_ip_from_json(obj: dict) -> str | None: + for k in ("x-real-ip", "x_real_ip", "x-forwarded-for", "x_forwarded_for"): + v = obj.get(k) + if isinstance(v, str) and v.strip(): + if "," in v: + parts = [p.strip() for p in v.split(",") if p.strip()] + for p in parts: + if ip_is_public(p): + return p + return parts[0] + return v.strip() + return None + +# --- Reporting queue --- +report_queue = [] +queue_lock = threading.Lock() + +def queue_report(ip, comment): + with queue_lock: + report_queue.append((ip, comment)) + +def flush_reports(api_key, dry_run=False): + with queue_lock: + batch = report_queue[:] + report_queue.clear() + for ip, comment in batch: + if not os.path.exists(CURL_BIN): + logging.error("curl binary not found at %s; cannot report", CURL_BIN) + continue + if dry_run: + logging.info("[dry-run] would report %s (len comment=%d)", ip, len(comment)) + continue + cmd = [ + CURL_BIN, "-sS", "-X", "POST", "https://api.abuseipdb.com/api/v2/report", + "-H", f"Key: {api_key}", + "-H", "Accept: application/json", + "--data-urlencode", f"ip={ip}", + "--data-urlencode", "categories=19", + "--data-urlencode", f"comment={comment}" + ] + try: + p = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + if p.returncode == 0: + logging.info("Reported %s to AbuseIPDB (stdout len=%d)", ip, len(p.stdout or "")) + else: + logging.warning("curl returned %s when reporting %s: %s", p.returncode, ip, (p.stderr or p.stdout or "")[:300]) + except Exception as e: + logging.exception("curl failed for %s: %s", ip, e) + +# --- Log line processing --- +def parse_log_line(line: str): + if "|" in line: + right = line.split("|", 1)[1].strip() + else: + right = line.strip() + try: + obj = json.loads(right) + return obj, right + except Exception: + return None, right + +def process_line(container, line, conn, api_key, dry_run): + obj, raw = parse_log_line(line) + found_deny = False + if obj and isinstance(obj, dict): + cr = obj.get("check_result") or {} + if isinstance(cr, dict) and cr.get("rule") == "DENY": + found_deny = True + else: + if '"rule":"DENY"' in raw or '"rule": "DENY"' in raw: + found_deny = True + if not found_deny: + return + + ips = [] + if obj: + cand = pick_ip_from_json(obj) + if cand: + for p in re.split(r'[;,\s]+', cand): + try: + ips.append(str(ipaddress.ip_address(p.strip()))) + except Exception: + continue + if not ips: + ips = extract_ips_from_text(raw) + ips = [ip for ip in ips if ip_is_public(ip)] + if not ips: + logging.info("DENY found but no public IP: %s", raw[:200]) + return + + for ip in ips: + logsig = signature_for_log(container, ip, raw) + if already_reported(conn, ip, logsig): + continue + comment = f"Anubis DENY log\nContainer: {container}\nLog: {raw[:950]}" + queue_report(ip, comment) + mark_reported(conn, ip, logsig) + +# --- Container follower --- +class ContainerFollower(threading.Thread): + def __init__(self, name, docker_bin, conn, api_key, dry_run): + super().__init__(daemon=True) + self.name = name + self.docker_bin = docker_bin + self.conn = conn + self.api_key = api_key + self.dry_run = dry_run + self.proc = None + self.stopped = threading.Event() + + def run(self): + while not self.stopped.is_set(): + try: + # Filter DENY logs at Docker level + cmd = f'{self.docker_bin} logs -f {self.name} | grep \'"rule":"DENY"\'' + self.proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, + shell=True, text=True) + logging.info("Started DENY-only log follower: %s", self.name) + for rawline in self.proc.stdout: + if self.stopped.is_set(): + break + if not rawline.strip(): + continue + process_line(self.name, rawline.rstrip("\n"), self.conn, self.api_key, self.dry_run) + except Exception: + logging.exception("Follower for %s crashed; restarting in 2s", self.name) + time.sleep(2) + finally: + if self.proc and self.proc.poll() is None: + try: + self.proc.terminate() + except Exception: + pass + self.proc = None + time.sleep(1) + + def stop(self): + self.stopped.set() + if self.proc and self.proc.poll() is None: + try: + self.proc.terminate() + except Exception: + pass + +# --- Supervisor --- +def supervisor(dry_run=False): + ensure_dirs() + conn = init_db() + if not os.path.exists(ABUSEIPDB_KEY_FILE): + logging.warning("AbuseIPDB key file missing: %s", ABUSEIPDB_KEY_FILE) + api_key = None + else: + api_key = open(ABUSEIPDB_KEY_FILE).read().strip() + followers = {} + + # reporter thread + def reporter_loop(): + while True: + time.sleep(REPORT_FLUSH_INTERVAL) + flush_reports(api_key, dry_run=dry_run) + t = threading.Thread(target=reporter_loop, daemon=True) + t.start() + + while True: + try: + p = subprocess.run([DOCKER_BIN, "ps", "--filter", "name=anubis", "--format", "{{.Names}}"], + capture_output=True, text=True) + if p.returncode != 0: + names = [] + else: + names = [ln.strip() for ln in p.stdout.splitlines() if ln.strip()] + except Exception: + logging.exception("docker ps failed") + names = [] + current = set(names) + existing = set(followers.keys()) + + for name in current - existing: + f = ContainerFollower(name, DOCKER_BIN, conn, api_key, dry_run) + followers[name] = f + f.start() + logging.info("Started follower for %s", name) + + for name in existing - current: + logging.info("Stopping follower for %s", name) + followers[name].stop() + followers[name].join(timeout=5) + followers.pop(name, None) + + time.sleep(SLEEP_REFRESH) + +def main(): + import argparse + parser = argparse.ArgumentParser() + parser.add_argument("--dry-run", action="store_true", help="Do not actually POST") + args = parser.parse_args() + try: + supervisor(dry_run=args.dry_run) + except KeyboardInterrupt: + logging.info("Shutting down") + sys.exit(0) + +if __name__ == "__main__": + main() \ No newline at end of file