Add net_alerter: new device join alerts via Matrix to condo Pi

net_alerter/net_alerter.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+"""
+net_alerter.py — Alert on new devices joining the local network.
+Runs via cron, sends Matrix messages when unknown MACs appear.
+"""
+
+import json
+import os
+import re
+import socket
+import sqlite3
+import subprocess
+import sys
+import urllib.request
+from datetime import datetime
+from pathlib import Path
+
+# --- Config (overridable via env) ---
+MATRIX_HOMESERVER = os.getenv("MATRIX_HOMESERVER", "https://m.example.org")
+MATRIX_ACCESS_TOKEN = os.getenv("MATRIX_ACCESS_TOKEN", "")
+MATRIX_ROOM_ID = os.getenv("MATRIX_ROOM_ID", "!REDACTED:example.org")
+DB_PATH = Path(os.getenv("NET_ALERTER_DB", "/opt/net_alerter/devices.db"))
+
+
+def get_db() -> sqlite3.Connection:
+    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(DB_PATH)
+    conn.execute("""
+        CREATE TABLE IF NOT EXISTS devices (
+            mac     TEXT PRIMARY KEY,
+            ip      TEXT,
+            name    TEXT,
+            first_seen TEXT,
+            last_seen  TEXT
+        )
+    """)
+    conn.commit()
+    return conn
+
+
+def get_local_subnets() -> list[str]:
+    """Return CIDR subnets for non-loopback, non-tailscale, non-wireguard interfaces."""
+    try:
+        out = subprocess.check_output(["ip", "-o", "-4", "addr", "show"], text=True)
+    except Exception:
+        return []
+
+    subnets = []
+    for line in out.splitlines():
+        # Fields: index iface inet ip/prefix ...
+        parts = line.split()
+        if len(parts) < 4:
+            continue
+        iface = parts[1]
+        cidr = parts[3]
+        # Skip loopback, tailscale, wireguard, docker
+        if any(iface.startswith(p) for p in ("lo", "tailscale", "wg", "tun", "docker", "br-")):
+            continue
+        subnets.append(cidr)
+    return subnets
+
+
+def scan_subnet(cidr: str) -> list[dict]:
+    """Run nmap ping scan, return list of {ip, mac, name}."""
+    try:
+        out = subprocess.check_output(
+            ["nmap", "-sn", "-PR", "--host-timeout", "5s", cidr],
+            text=True, stderr=subprocess.DEVNULL
+        )
+    except Exception:
+        return []
+
+    devices = []
+    current = {}
+    for line in out.splitlines():
+        if line.startswith("Nmap scan report for"):
+            if current.get("mac"):
+                devices.append(current)
+            # "Nmap scan report for hostname (ip)" or "... for ip"
+            m = re.search(r"for (.+?)(?:\s+\((.+?)\))?$", line)
+            if m:
+                if m.group(2):
+                    current = {"name": m.group(1), "ip": m.group(2), "mac": ""}
+                else:
+                    current = {"name": "", "ip": m.group(1), "mac": ""}
+        elif "MAC Address:" in line:
+            m = re.search(r"MAC Address: ([0-9A-F:]{17})", line)
+            if m:
+                current["mac"] = m.group(1).lower()
+
+    if current.get("mac"):
+        devices.append(current)
+
+    return devices
+
+
+def send_matrix_alert(msg: str) -> None:
+    if not MATRIX_ACCESS_TOKEN:
+        print(f"[warn] No MATRIX_ACCESS_TOKEN set, skipping alert: {msg}", file=sys.stderr)
+        return
+
+    url = (
+        f"{MATRIX_HOMESERVER}/_matrix/client/v3/rooms/"
+        f"{urllib.parse.quote(MATRIX_ROOM_ID, safe='')}/send/m.room.message"
+    )
+    payload = json.dumps({"msgtype": "m.text", "body": msg}).encode()
+    req = urllib.request.Request(
+        url,
+        data=payload,
+        headers={
+            "Authorization": f"Bearer {MATRIX_ACCESS_TOKEN}",
+            "Content-Type": "application/json",
+        },
+        method="POST",
+    )
+    try:
+        urllib.request.urlopen(req, timeout=10)
+    except Exception as e:
+        print(f"[error] Matrix send failed: {e}", file=sys.stderr)
+
+
+# urllib.parse needed for quote
+import urllib.parse  # noqa: E402
+
+
+def run() -> None:
+    conn = get_db()
+    now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
+    hostname = socket.gethostname()
+
+    subnets = get_local_subnets()
+    if not subnets:
+        print("[warn] No local subnets found", file=sys.stderr)
+        return
+
+    new_devices = []
+
+    for cidr in subnets:
+        devices = scan_subnet(cidr)
+        for d in devices:
+            mac = d["mac"]
+            row = conn.execute("SELECT mac FROM devices WHERE mac=?", (mac,)).fetchone()
+            if row is None:
+                conn.execute(
+                    "INSERT INTO devices (mac, ip, name, first_seen, last_seen) VALUES (?,?,?,?,?)",
+                    (mac, d["ip"], d["name"], now, now),
+                )
+                new_devices.append(d)
+            else:
+                conn.execute(
+                    "UPDATE devices SET ip=?, name=?, last_seen=? WHERE mac=?",
+                    (d["ip"], d["name"], now, mac),
+                )
+    conn.commit()
+    conn.close()
+
+    for d in new_devices:
+        label = d["name"] if d["name"] else "unknown"
+        msg = (
+            f"[{hostname}] NEW DEVICE\n"
+            f"  IP:   {d['ip']}\n"
+            f"  MAC:  {d['mac']}\n"
+            f"  Name: {label}\n"
+            f"  Time: {now}"
+        )
+        print(msg)
+        send_matrix_alert(msg)
+
+
+if __name__ == "__main__":
+    run()