Auto-refresh Matrix token on 401 instead of silently dropping alerts

net_alerter/net_alerter.py

@@ -11,6 +11,7 @@ import socket
 import sqlite3
 import subprocess
 import sys
+import urllib.parse
 import urllib.request
 from datetime import datetime
 from pathlib import Path
@@ -19,7 +20,10 @@ from pathlib import Path
 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:
@@ -38,8 +42,75 @@ def get_db() -> sqlite3.Connection:
     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 in (401, 403):
+        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)
+
+
 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:
@@ -47,13 +118,11 @@ def get_local_subnets() -> list[str]:
 
     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)
@@ -61,7 +130,6 @@ def get_local_subnets() -> list[str]:
 
 
 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],
@@ -76,7 +144,6 @@ def scan_subnet(cidr: str) -> list[dict]:
         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):
@@ -94,35 +161,6 @@ def scan_subnet(cidr: str) -> list[dict]:
     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")