feat: DDoS detector for ports 80/443 with Telegram alert (cron 1min, ss-based)

ddos_detector.py

@@ -0,0 +1,148 @@
+#!/usr/bin/env python3
+"""DDoS detector for public web ports (80/443).
+Runs every minute via cron. Counts active connections per IP on 80/443,
+alerts Telegram if a single IP exceeds CONN_PER_IP or total exceeds CONN_TOTAL.
+State file prevents alert spam (only alerts on threshold crossing).
+"""
+import os
+import re
+import subprocess
+import datetime
+
+# Load telegram creds from stock-monitor .env (or env)
+def load_env(path):
+    env = {}
+    try:
+        with open(path) as f:
+            for line in f:
+                line = line.strip()
+                if line and not line.startswith("#") and "=" in line:
+                    k, v = line.split("=", 1)
+                    env[k.strip()] = v.strip()
+    except FileNotFoundError:
+        pass
+    return env
+
+ENV = load_env("/root/stock-monitor/.env")
+BOT = os.environ.get("TELEGRAM_BOT_TOKEN") or ENV.get("TELEGRAM_BOT_TOKEN", "")
+CHAT = os.environ.get("TELEGRAM_CHAT_ID") or ENV.get("TELEGRAM_CHAT_ID", "5722352727")
+
+PORTS = [80, 443]
+CONN_PER_IP = int(os.environ.get("DDOS_PER_IP", "100"))   # alert if 1 IP > this
+CONN_TOTAL = int(os.environ.get("DDOS_TOTAL", "500"))      # alert if total on a port > this
+STATE_FILE = "/root/ddos_state.json"
+
+# ports currently exposed to public (from ss output)
+def get_connections():
+    """Return dict: port -> {ip: count}"""
+    out = subprocess.run(
+        ["ss", "-tn", "src", ":80", "or", "src", ":443"],
+        capture_output=True, text=True
+    ).stdout
+    result = {p: {} for p in PORTS}
+    # ss -tn src :80 or src :443  -> lines like:
+    # ESTAB 0 0 127.0.0.1:80 1.2.3.4:55555
+    for line in out.splitlines()[1:]:
+        parts = line.split()
+        if len(parts) < 4:
+            continue
+        local = parts[3]   # e.g. 0.0.0.0:80 or [::]:443
+        remote = parts[4]
+        # extract port from local
+        m = re.search(r":(\d+)$", local)
+        if not m:
+            continue
+        port = int(m.group(1))
+        if port not in result:
+            continue
+        # extract IP from remote (strip port)
+        ip = remote.rsplit(":", 1)[0]
+        ip = ip.strip("[]")
+        result[port][ip] = result[port].get(ip, 0) + 1
+    return result
+
+
+def send_telegram(msg):
+    if not BOT:
+        print("NO BOT TOKEN, skip send")
+        return
+    import urllib.request
+    import urllib.parse
+    url = f"https://api.telegram.org/bot{BOT}/sendMessage"
+    data = urllib.parse.urlencode({"chat_id": CHAT, "text": msg}).encode()
+    req = urllib.request.Request(url, data=data, method="POST")
+    try:
+        urllib.request.urlopen(req, timeout=10)
+    except Exception as e:
+        print(f"telegram send failed: {e}")
+
+
+def load_state():
+    import json
+    try:
+        with open(STATE_FILE) as f:
+            return json.load(f)
+    except (FileNotFoundError, ValueError):
+        return {}
+
+
+def save_state(s):
+    import json
+    with open(STATE_FILE, "w") as f:
+        json.dump(s, f)
+
+
+def main():
+    conns = get_connections()
+    now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    alerts = []
+    for port in PORTS:
+        ip_map = conns[port]
+        total = sum(ip_map.values())
+        # top offenders
+        top = sorted(ip_map.items(), key=lambda x: -x[1])[:5]
+        for ip, cnt in top:
+            if cnt > CONN_PER_IP:
+                alerts.append(
+                    f"⚠️ Port {port}: IP {ip} = {cnt} koneksi (batas {CONN_PER_IP})"
+                )
+        if total > CONN_TOTAL:
+            alerts.append(
+                f"⚠️ Port {port}: TOTAL {total} koneksi (batas {CONN_TOTAL})"
+            )
+
+    state = load_state()
+    new_alert = False
+    if alerts:
+        # only alert if we weren't already in alert state (avoid spam)
+        if not state.get("alerting"):
+            new_alert = True
+        state["alerting"] = True
+        state["last_alerts"] = alerts
+    else:
+        if state.get("alerting"):
+            # send "cleared" message once
+            send_telegram(
+                f"✅ DDoS alert cleared @ {now}\n"
+                f"Port 80: {sum(conns[80].values())} koneksi\n"
+                f"Port 443: {sum(conns[443].values())} koneksi\n"
+                f"(batas per-IP {CONN_PER_IP}, total {CONN_TOTAL})"
+            )
+        state["alerting"] = False
+
+    if new_alert:
+        top80 = sorted(conns[80].items(), key=lambda x: -x[1])[:5]
+        top443 = sorted(conns[443].items(), key=lambda x: -x[1])[:5]
+        msg = f"🚨 DDoS ALERT @ {now}\n\n" + "\n".join(alerts) + "\n\n"
+        msg += f"Top IP port 80: {top80}\n"
+        msg += f"Top IP port 443: {top443}\n"
+        msg += f"(batas per-IP {CONN_PER_IP}, total {CONN_TOTAL})"
+        send_telegram(msg)
+
+    save_state(state)
+    # debug log (no secrets)
+    print(f"{now} 80={sum(conns[80].values())} 443={sum(conns[443].values())} alert={state.get('alerting')}")
+
+
+if __name__ == "__main__":
+    main()