#!/usr/bin/env python3 """Boltwatch telemetry agent. Python 3.10+, standard library only. MIT licensed. This agent reads receiver health JSON; it does not capture RF or locate lightning. Run `python3 boltwatch_feeder.py --help` for configuration and usage. """ from __future__ import annotations import argparse import datetime as dt import getpass import json import math import os from pathlib import Path import random import re import sys import time import urllib.error import urllib.parse import urllib.request VERSION = "0.1.0" MAX_BYTES = 8192 DEFAULT_CONFIG = Path.home() / ".config" / "boltwatch-feeder" / "config.json" class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): raise ValueError("Redirect refused. Check the configured Boltwatch origin.") def origin(value: str) -> str: parsed = urllib.parse.urlsplit(value) if (parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ("", "/") or not parsed.hostname): raise ValueError("Use an origin such as https://boltwatch.live, without a path or credentials.") if parsed.scheme != "https" and not (parsed.scheme == "http" and parsed.hostname in ("localhost", "127.0.0.1", "::1")): raise ValueError("HTTPS is required, except for a local test server.") return value.rstrip("/") def read_json(path: Path): with path.open("rb") as handle: raw = handle.read(MAX_BYTES + 1) if len(raw) > MAX_BYTES: raise ValueError("JSON file exceeds 8 KiB.") return json.loads(raw, parse_constant=lambda value: (_ for _ in ()).throw(ValueError("Non-finite number."))) def validate_report(value): if not isinstance(value, dict): raise ValueError("Receiver report must be an object.") ranges = { "window_seconds": (1, 600, False), "timing_uncertainty_ns": (0, 1e9, False), "noise_floor_dbfs": (-200, 0, False), "clipped_fraction": (0, 1, False), "dropped_samples": (0, 2**53 - 1, True), "received_signals": (0, 2**53 - 1, True), "submitted_signals": (0, 2**53 - 1, True), "sample_rate_hz": (1, 1e8, False), } if set(value) - (set(ranges) | {"reported_at", "gnss_locked"}): raise ValueError("Receiver report contains unsupported fields.") timestamp = value.get("reported_at") if not isinstance(timestamp, str) or not re.fullmatch(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d{1,3})?Z", timestamp): raise ValueError("reported_at must be a UTC ISO timestamp.") dt.datetime.fromisoformat(timestamp.replace("Z", "+00:00")) if "window_seconds" not in value: raise ValueError("window_seconds is required.") for key, (minimum, maximum, integer) in ranges.items(): if key not in value: continue number = value[key] if isinstance(number, bool) or not isinstance(number, (int, float)) or not math.isfinite(number) or not minimum <= number <= maximum or (integer and int(number) != number): raise ValueError(f"Invalid {key}.") if "gnss_locked" in value and not isinstance(value["gnss_locked"], bool): raise ValueError("gnss_locked must be true or false.") if "submitted_signals" in value and "received_signals" in value and value["submitted_signals"] > value["received_signals"]: raise ValueError("Submitted signals exceed received signals.") return value def load_config(path: Path): if os.name == "posix" and path.stat().st_mode & 0o077: raise ValueError("Config contains a secret. Run chmod 600 on the config file.") value = read_json(path) if not isinstance(value, dict) or not isinstance(value.get("token"), str) or not re.fullmatch(r"bw_[A-Za-z0-9_-]{43}", value["token"]): raise ValueError("Invalid station token in config.") value["origin"] = origin(value.get("origin", "")) return value def configure(path: Path): url = origin(input("Boltwatch origin [https://boltwatch.live]: ").strip() or "https://boltwatch.live") token = getpass.getpass("Station token (hidden): ").strip() if not re.fullmatch(r"bw_[A-Za-z0-9_-]{43}", token): raise ValueError("Invalid station token.") report = input("Receiver health JSON path (leave blank for connection checks only): ").strip() config = {"origin": url, "token": token, "report_path": str(Path(report).expanduser().resolve()) if report else None} path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) # Never follow an existing symlink or silently overwrite another station's credentials. descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "w") as handle: json.dump(config, handle, indent=2) handle.write("\n") print(f"Saved config to {path}. Keep it private.") def heartbeat(config, opener=None): report = None if config.get("report_path"): try: report = validate_report(read_json(Path(config["report_path"]))) except (OSError, ValueError, TypeError): print("Receiver report unavailable or invalid; sending a connection heartbeat only.", file=sys.stderr) payload = {"schema_version": 1, "agent_version": VERSION, "receiver": report} # Preserve reported_at. Rewriting it here would make a stopped receiver look healthy. request = urllib.request.Request( config["origin"] + "/api/network/heartbeat", data=json.dumps(payload, allow_nan=False).encode(), headers={"Content-Type": "application/json", "Authorization": "Bearer " + config["token"], "User-Agent": "BoltwatchFeeder/" + VERSION}, method="POST", ) client = opener or urllib.request.build_opener(NoRedirect()) with client.open(request, timeout=15) as response: body = response.read(65537) if len(body) > 65536: raise ValueError("Server response is too large.") result = json.loads(body) if result.get("accepted") is not True: raise ValueError("Server did not acknowledge this heartbeat.") print("Heartbeat accepted" + (" with receiver measurements." if report else "; receiver measurements unknown."), flush=True) def run(config, once=False): failures = 0 while True: try: heartbeat(config) failures = 0 except urllib.error.HTTPError as error: print(f"Heartbeat rejected (HTTP {error.code}).", file=sys.stderr, flush=True) if error.code in (401, 403): print("Check the station token. Agent stopped.", file=sys.stderr) return 2 failures += 1 except (OSError, ValueError, urllib.error.URLError): # Avoid dumping request objects, credentials, or receiver contents to logs. print("Heartbeat failed. Check connectivity, origin and receiver timestamps.", file=sys.stderr, flush=True) failures += 1 if once: return 1 if failures else 0 time.sleep(min(300, 30 * (2 ** min(failures, 3))) + random.uniform(0, 3)) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("configure", "run")) parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG) parser.add_argument("--once", action="store_true", help="Send one heartbeat, then exit.") args = parser.parse_args() try: if args.command == "configure": configure(args.config.expanduser()) return 0 return run(load_config(args.config.expanduser()), args.once) except (OSError, ValueError, TypeError, KeyError): print("Configuration failed. Check the config path, permissions and values; use --help for usage.", file=sys.stderr) return 2 except KeyboardInterrupt: return 0 if __name__ == "__main__": sys.exit(main())