vanlink/modem/van-sms-watch

#!/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)