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.parse
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")
MATRIX_USER = os.getenv("MATRIX_USER", "")
MATRIX_PASSWORD = os.getenv("MATRIX_PASSWORD", "")
DB_PATH = Path(os.getenv("NET_ALERTER_DB", "/opt/net_alerter/devices.db"))
ENV_PATH = Path(os.getenv("NET_ALERTER_ENV", "/opt/net_alerter/.env"))


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 refresh_token() -> str:
    if not MATRIX_USER or not MATRIX_PASSWORD:
        print("[error] Cannot refresh token: MATRIX_USER/MATRIX_PASSWORD not set", file=sys.stderr)
        return ""
    payload = json.dumps({
        "type": "m.login.password",
        "identifier": {"type": "m.id.user", "user": MATRIX_USER},
        "password": MATRIX_PASSWORD,
        "initial_device_display_name": "net-alerter",
    }).encode()
    req = urllib.request.Request(
        f"{MATRIX_HOMESERVER}/_matrix/client/v3/login",
        data=payload,
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as r:
            data = json.loads(r.read())
        token = data.get("access_token", "")
        if token and ENV_PATH.exists():
            lines = ENV_PATH.read_text().splitlines()
            lines = [l if not l.startswith("MATRIX_ACCESS_TOKEN=") else f"MATRIX_ACCESS_TOKEN={token}" for l in lines]
            ENV_PATH.write_text("\n".join(lines) + "\n")
        return token
    except Exception as e:
        print(f"[error] Token refresh failed: {e}", file=sys.stderr)
        return ""


def _do_send(token: str, msg: str) -> int:
    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 {token}", "Content-Type": "application/json"},
        method="POST",
    )
    try:
        urllib.request.urlopen(req, timeout=10)
        return 200
    except urllib.error.HTTPError as e:
        return e.code
    except Exception as e:
        print(f"[error] Matrix send failed: {e}", file=sys.stderr)
        return 0


def send_matrix_alert(msg: str) -> None:
    global MATRIX_ACCESS_TOKEN
    if not MATRIX_ACCESS_TOKEN:
        print(f"[warn] No MATRIX_ACCESS_TOKEN set, skipping alert", file=sys.stderr)
        return
    code = _do_send(MATRIX_ACCESS_TOKEN, msg)
    if code == 401:
        print("[info] Token expired, refreshing...", file=sys.stderr)
        new_token = refresh_token()
        if new_token:
            MATRIX_ACCESS_TOKEN = new_token
            _do_send(MATRIX_ACCESS_TOKEN, msg)
        else:
            print("[error] Could not refresh token, alert dropped", file=sys.stderr)
    elif code == 403:
        print("[error] Room permission denied (403) — check bot membership in room", file=sys.stderr)
    elif code not in (200, 0):
        print(f"[error] Matrix send failed with HTTP {code}", file=sys.stderr)


def get_local_subnets() -> list[str]:
    try:
        out = subprocess.check_output(["ip", "-o", "-4", "addr", "show"], text=True)
    except Exception:
        return []

    subnets = []
    for line in out.splitlines():
        parts = line.split()
        if len(parts) < 4:
            continue
        iface = parts[1]
        cidr = parts[3]
        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]:
    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)
            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 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()