#!/usr/bin/env python3 """van-sms-watch — archives inbound SMS on the EC25 modem and pages via Pushover. Polls ModemManager (via `mmcli -J`, no dbus dependency) for SMS sitting in the modem's "mt" storage. Each *received* message (pdu-type "deliver" — this skips the modem's own "submit"/"sent" records) is appended as one JSON line to /var/log/van-sms.jsonl, pushed via Pushover, then deleted from modem storage — that storage is small flash on the EC25 itself and fills up silently over time, so anything present in a poll is by definition new and needs archiving before it's gone. MMS is not decodable here: MMS arrives over SMS only as a WAP-push binary notification (a URL to fetch over the carrier's MMS APN with carrier-specific auth) — ModemManager hands it back with no `text`, just raw `data` hex. Decoding WSP and fetching that URL is real added complexity with no stdlib support and no guarantee the MMS APN is even reachable from this modem's general-internet APN. Instead: detect it (deliver + no text + non-empty data), save the raw record, and page with a note that the content itself wasn't retrieved — better than silently dropping it. Pushover credentials shared with van-battery/van-thermal/van-modem-watch (/etc/van-battery/pushover.json, 0600). Publishes /run/van-sms-watch/ state.json (same convention as the other watchdogs). Stdlib only (mmcli does the ModemManager talking over subprocess). """ import json import socket import subprocess import sys import time import urllib.parse import urllib.request from datetime import datetime, timezone from pathlib import Path INTERVAL = 15 # seconds between polls LOG_PATH = Path("/var/log/van-sms.jsonl") STATE_DIR = Path("/run/van-sms-watch") STATE_PATH = STATE_DIR / "state.json" CREDENTIALS_PATH = Path("/etc/van-battery/pushover.json") HOST = socket.gethostname() PLACEHOLDERS = {"", "REPLACE_ME", "your-token-here", "your-user-key-here"} PUSHOVER_MAX = 900 # Pushover message cap is 1024 bytes; leave headroom for the prefix def log(msg, level="info"): # systemd journal severity prefixes (sd-daemon), same convention as the other watchdogs. pri = {"info": "<6>", "warn": "<4>", "crit": "<2>"}.get(level, "<6>") print(pri + msg, flush=True) def load_creds(): try: c = json.loads(CREDENTIALS_PATH.read_text()) token, user = str(c.get("token", "")).strip(), str(c.get("user", "")).strip() if token in PLACEHOLDERS or user in PLACEHOLDERS: return None return token, user except FileNotFoundError: return None except Exception as e: log(f"credentials {CREDENTIALS_PATH} unreadable ({e})", "warn") return None def pushover(title, message, attempts=3, retry_delay=15): creds = load_creds() if not creds: log(f"pushover skipped (no credentials): {title} — {message}", "warn") return False token, user = creds data = urllib.parse.urlencode({ "token": token, "user": user, "title": title, "message": message, }).encode() req = urllib.request.Request("https://api.pushover.net/1/messages.json", data=data) for attempt in range(1, attempts + 1): try: with urllib.request.urlopen(req, timeout=10) as resp: ok = resp.status == 200 if not ok: log(f"pushover HTTP {resp.status}", "warn") return ok except Exception as e: last = attempt == attempts log(f"pushover send failed ({attempt}/{attempts}): {e}", "warn") if not last: time.sleep(retry_delay) return False def mmcli_json(args): """Run mmcli with -J and return the parsed dict, or None on any failure.""" try: out = subprocess.run(["mmcli", "-J", *args], capture_output=True, text=True, timeout=15) except Exception as e: log(f"mmcli {' '.join(args)} failed to run ({e})", "warn") return None if out.returncode != 0: log(f"mmcli {' '.join(args)} exited {out.returncode}: {out.stderr.strip()}", "warn") return None try: return json.loads(out.stdout) except Exception as e: log(f"mmcli {' '.join(args)} gave unparseable JSON ({e})", "warn") return None def find_modem(): """Path of the first modem ModemManager knows about, or None. Re-resolved each poll — the modem index can shift across a ModemManager restart.""" d = mmcli_json(["-L"]) if not d: return None modems = d.get("modem-list") or [] return modems[0] if modems else None def list_sms(modem_path): d = mmcli_json(["-m", modem_path, "--messaging-list-sms"]) if d is None: return [] return d.get("modem.messaging.sms") or [] def fetch_sms(modem_path, sms_path): d = mmcli_json(["-m", modem_path, "-s", sms_path]) if d is None: return None return d.get("sms") def delete_sms(modem_path, sms_path): try: out = subprocess.run( ["mmcli", "-m", modem_path, f"--messaging-delete-sms={sms_path}"], capture_output=True, text=True, timeout=15) except Exception as e: log(f"delete {sms_path} failed to run ({e})", "warn") return False if out.returncode != 0: log(f"delete {sms_path} exited {out.returncode}: {out.stderr.strip()}", "warn") return False return True def archive(record): try: LOG_PATH.parent.mkdir(parents=True, exist_ok=True) with open(LOG_PATH, "a") as f: f.write(json.dumps(record) + "\n") return True except OSError as e: log(f"archive write failed: {e}", "warn") return False def process(modem_path, sms_path): sms = fetch_sms(modem_path, sms_path) if sms is None: return props = sms.get("properties", {}) content = sms.get("content", {}) if props.get("pdu-type") != "deliver": return # not an inbound message (e.g. the modem's own "submit"/"sent" records) number = content.get("number", "unknown") text = content.get("text", "--") data = content.get("data", "--") timestamp = props.get("timestamp", "--") is_mms = text in ("--", "") and data not in ("--", "") record = { "time": datetime.now(timezone.utc).isoformat(timespec="seconds"), "modem_timestamp": timestamp, "from": number, "kind": "mms-notification" if is_mms else "sms", "text": None if is_mms else text, "data_hex": data if is_mms else None, "smsc": props.get("smsc", "--"), } saved = archive(record) if saved: log(f"archived {record['kind']} from {number}") else: log(f"failed to archive {record['kind']} from {number} — not deleting from modem", "warn") if is_mms: title = f"📎 {HOST}: MMS notification from {number}" body = ("MMS received but content can't be fetched on this device " "(needs the carrier's MMS APN + WAP-push decode, not implemented). " f"Raw notification archived to {LOG_PATH}.") else: title = f"📩 {HOST}: SMS from {number}" body = text if len(text) <= PUSHOVER_MAX else text[:PUSHOVER_MAX] + "… (truncated)" pushover(title, body) if saved: if delete_sms(modem_path, sms_path): log(f"deleted {sms_path} from modem storage") else: log(f"left {sms_path} on modem storage (delete failed) — will retry next poll", "warn") def write_state(payload): STATE_DIR.mkdir(parents=True, exist_ok=True) tmp = STATE_PATH.with_suffix(".tmp") tmp.write_text(json.dumps(payload)) tmp.replace(STATE_PATH) def main(): log(f"van-sms-watch up: polling every {INTERVAL}s, archiving to {LOG_PATH}") modem_present = None # unknown yet, so the first observation always logs while True: modem_path = find_modem() present = modem_path is not None if present != modem_present: log("modem present" if present else "modem not found (skipping poll)", "info" if present else "warn") modem_present = present count = 0 if modem_path: for sms_path in list_sms(modem_path): process(modem_path, sms_path) count += 1 write_state({ "updated": datetime.now(timezone.utc).isoformat(timespec="seconds"), "modem_present": present, "last_poll_messages": count, }) time.sleep(INTERVAL) if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(0)