bigbrother/net_alerter/net_alerter.py

#!/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(
            ["sudo", "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()