#!/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()