#!/usr/bin/env python3 """Waveline Alexa relay API. A tiny token-authenticated command queue that bridges the Virtual Smart Home Node-RED flow (which receives commands from Amazon Alexa) and the Waveline desktop client (which polls for commands). Endpoints (all POST, JSON): POST /poll {"token": ""} -> {"status": "ok", "commands": [...]} Drains the client's queue, oldest first. Client token only. POST /command {"token": "", "command": "pause", "source": "alexa"} -> {"status": "ok", "queued": true} Shared secret ONLY. Used by Node-RED (or the admin via curl) to enqueue commands for the client. A stolen client token cannot enqueue anything. POST /status {"token": ""} -> {"status": "ok", "clients": {...}} Ops endpoint: queue depth, rate-limit counters, auth-failure counts. Shared secret only. (The old /register endpoint was removed: it handed the client token to anyone holding either token. The client token is distributed out of band - /etc/wrelay-alexa/client-token on the server.) Security model: - Two token tiers. The shared secret (Node-RED + admin) can enqueue. The client token (Waveline on the PC) can ONLY poll and drain. Every endpoint checks the tier explicitly; tier escalation is impossible by construction. - Commands unknown to the fixed vocabulary are rejected, so this API cannot make Waveline do anything beyond the playback command set. - Every request is rate-limited (per source for commands, global for the rest); repeated failed authentications from one IP trip a temporary block. - Queued commands carry a TTL: stale momentary commands (older than COMMAND_TTL seconds) are dropped rather than replayed much later. """ import hmac import ipaddress import json import os import secrets import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # --- configuration ----------------------------------------------------------- LISTEN_HOST = os.environ.get("WAVELINE_RELAY_HOST", "127.0.0.1") LISTEN_PORT = int(os.environ.get("WAVELINE_RELAY_PORT", "8099")) # Files are used instead of env vars so tokens survive reboots and can be # rotated without touching the unit file. CONF_DIR = os.environ.get("WAVELINE_RELAY_CONF", "/etc/wrelay-alexa") SHARED_SECRET_FILE = os.path.join(CONF_DIR, "shared-secret") CLIENT_TOKEN_FILE = os.path.join(CONF_DIR, "client-token") VALID_COMMANDS = { "play", "pause", "play_pause", "stop", "next", "previous", "rewind", "fast_forward", "volume_up", "volume_down", "mute", "volume_set", "volume_adjust", "start_over", # sleep timer (value = minutes for sleep_set; delta for sleep_adjust) "sleep_set", "sleep_cancel", "sleep_status", "sleep_adjust", # playback extras "chapter_next", "chapter_prev", "bookmark_add", "ab_set_a", "ab_set_b", "ab_clear", "fullscreen", # playback speed (value = percent: 100 = normal) "speed_set", "speed_up", "speed_down", # extended voice set (2026-09-23): SkipChannels via entertainment # devices, fixed 30 s jumps, shuffle, status queries. "rewind_30", "fast_forward_30", "toggle_shuffle", "announce_track", "announce_time", # 5.0: voice recorder control + radio song query (the # "Alexa, turn on Recorder" plug maps to record_now/stop). "record_now", "record_stop", "what_song", # control devices (voice-only, 2026-09-23): speed/repeat/seek are # thermostats (raw values); shuffle_set keeps the Alexa switch state # in sync (explicit on/off, not a toggle). "repeat_set", "seek_set", "shuffle_set", } # Queued for the remote-relay path as "name:value" strings (the app side # strips the value and routes it - see MainWindow._run_action). RELAY_VALUE_COMMANDS = {"speed_set", "volume_set", "sleep_set", "sleep_adjust", "repeat_set", "seek_set", "shuffle_set"} # Commands whose "value" is an integer with a meaning: VALUE_COMMANDS = {"volume_set", "volume_adjust", "sleep_set", "sleep_adjust", "speed_set", "repeat_set", "seek_set", "shuffle_set"} VALID_SOURCES = {"alexa", "node-red", "api"} MAX_QUEUE = 50 RATE_LIMIT_PER_MIN = 30 # per source (alexa / node-red) OTHER_RATE_LIMIT_PER_MIN = 60 # polls / status / non-source requests AUTH_FAIL_BLOCK = 10 # failed auths from one IP before blocking AUTH_BLOCK_SECONDS = 600 # how long that block lasts COMMAND_TTL = 300 # seconds a queued command stays valid WAKE_TTL = 120 # seconds a wake flag stays set POLL_ALIVE_WINDOW = 10 # client seen within this window = running _lock = threading.Lock() _queue = [] # list of {"command","source","queued_at"} _shared_secret = None # set at startup _rate = {"alexa": [0, 0], "node-red": [0, 0], "_other": [0, 0]} _auth_failures = 0 _auth_fail_by_ip = {} # ip -> [first_fail_ts, count, blocked_until] _wake = {"until": 0.0} # app-launch requested by an offline command _last_poll = 0.0 # last client /poll (app alive marker) def _load_tokens(): global _shared_secret with open(SHARED_SECRET_FILE, "r", encoding="utf-8") as fh: _shared_secret = fh.read().strip() def _rate_ok(key: str, limit: int) -> bool: """Sliding one-minute window per key. Caller holds _lock.""" now = time.time() window, count = _rate.get(key, [now, 0]) if now - window >= 60.0: window, count = now, 0 count += 1 _rate[key] = [window, count] return count <= limit def _ip_blocked(ip: str) -> bool: """True while this IP is serving an auth-failure block. Caller holds _lock.""" now = time.time() entry = _auth_fail_by_ip.get(ip) if entry and entry[2] > now: return True return False def _record_auth_failure(ip: str) -> None: """Count a failed auth; block the IP after too many. Caller holds _lock.""" global _auth_failures now = time.time() _auth_failures += 1 first, count, _ = _auth_fail_by_ip.get(ip, [now, 0, 0.0]) if now - first > AUTH_BLOCK_SECONDS: first, count = now, 0 count += 1 blocked_until = now + AUTH_BLOCK_SECONDS if count >= AUTH_FAIL_BLOCK else 0 _auth_fail_by_ip[ip] = [first, count, blocked_until] def _enqueue(command: str, source: str, value=None) -> int: """Queue a command. Returns the number of expired entries dropped. WAKE FEATURE: a command arriving while the client is NOT polling (app closed) also sets the wake flag, so a resident watcher can launch the app - which then drains the queued command normally. Wake TTL is short and the flag is one-shot (cleared by /wake).""" now = int(time.time()) dropped = 0 with _lock: # TTL sweep: momentary commands go stale; never replay them. alive = [e for e in _queue if now - e["queued_at"] <= COMMAND_TTL] dropped = len(_queue) - len(alive) _queue[:] = alive if len(_queue) >= MAX_QUEUE: _queue.pop(0) # drop oldest when full entry = { "command": command, "source": source, "queued_at": now, } if value is not None: entry["value"] = value _queue.append(entry) if now - _last_poll > POLL_ALIVE_WINDOW: # Client offline: remember to launch it (watcher consumes this). _wake["until"] = now + WAKE_TTL return dropped def _drain() -> list: now = int(time.time()) with _lock: alive = [e for e in _queue if now - e["queued_at"] <= COMMAND_TTL] _queue[:] = alive commands = _queue[:] _queue.clear() return commands class RelayHandler(BaseHTTPRequestHandler): server_version = "WavelineRelay/1.1" # -- helpers -------------------------------------------------------------- def _client_ip(self) -> str: # Behind Apache's local proxy; X-Forwarded-For is set by Apache and # only trusted because the direct peer is always 127.0.0.1 (the # relay listens on loopback only). Apache APPENDS the connecting # client's address, so the LAST element is the trustworthy one - # earlier elements come from the client's own header and can be # forged to evade per-IP blocking. fwd = self.headers.get("X-Forwarded-For", "") if fwd: parts = [p.strip() for p in fwd.split(",") if p.strip()] if parts: try: ipaddress.ip_address(parts[-1]) return parts[-1] except ValueError: pass return self.client_address[0] def _send(self, code: int, payload: dict) -> None: body = json.dumps(payload).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _read_json(self): try: length = int(self.headers.get("Content-Length", "0")) if length > 8192: return None return json.loads(self.rfile.read(length).decode("utf-8", "replace")) except (ValueError, OSError): return None def _auth(self, token: str) -> str: """Return 'shared' | 'client' | ''. Constant-time compares.""" if not token or not isinstance(token, str): return "" if _shared_secret and hmac.compare_digest(token, _shared_secret): return "shared" client_path = os.path.join(CONF_DIR, "client-token") try: with open(client_path, "r", encoding="utf-8") as fh: client_token = fh.read().strip() except OSError: return "" if client_token and hmac.compare_digest(token, client_token): return "client" return "" # -- endpoints ------------------------------------------------------------ def do_GET(self): # The API is POST-only; answering GETs keeps scanners quiet. self._send(405, {"status": "error", "error": "POST only"}) def do_POST(self): ip = self._client_ip() with _lock: blocked = _ip_blocked(ip) if blocked: self._send(429, {"status": "error", "error": "try again later"}) return path = self.path.rstrip("/") data = self._read_json() if not isinstance(data, dict): self._send(400, {"status": "error", "error": "invalid JSON body"}) return kind = self._auth(data.get("token", "")) if not kind: with _lock: _record_auth_failure(ip) fails = _auth_fail_by_ip.get(ip, [0, 0, 0])[1] self._send(401, {"status": "error", "error": "unauthorized", "auth_failures": fails}) return if path == "/poll": if kind != "client": self._send(403, {"status": "error", "error": "poll requires the client token"}) return with _lock: if not _rate_ok("_other", OTHER_RATE_LIMIT_PER_MIN): limited = True else: limited = False if limited: self._send(429, {"status": "error", "error": "rate limited"}) return commands = _drain() with _lock: global _last_poll _last_poll = time.time() _wake["until"] = 0.0 # app is alive: wake served self._send(200, {"status": "ok", "commands": commands}) return if path == "/wake": # The resident watcher long-polls this: 200 + {"wake": true} # means a voice command arrived while the app was closed - # launch Waveline now. # TIER FIX (2026-09-25, real-world bug): the app ships ONE # token - the CLIENT token (it polls /poll with it). The # watcher long-polls /wake with that same stored token, so # /wake MUST accept the client tier too; requiring only the # shared secret made every real watcher 403 forever (silent, # found because "wake doesn't work after PC restart"). # Waking is harmless (launches the owner's own app) and # one-shot; /command stays shared-only. if kind not in ("shared", "client"): self._send(403, {"status": "error", "error": "wake requires a valid token"}) return wait = min(50, max(0, int(data.get("wait", 0)))) deadline = time.time() + wait while True: with _lock: until = _wake["until"] if until > time.time(): with _lock: _wake["until"] = 0.0 # one-shot self._send(200, {"status": "ok", "wake": True}) return if time.time() >= deadline: self._send(200, {"status": "ok", "wake": False}) return time.sleep(1.0) if path == "/command": if kind != "shared": # Client tokens cannot enqueue: poll-only tier. self._send(403, {"status": "error", "error": "command requires the shared secret"}) return command = data.get("command", "") source = data.get("source", "api") if command not in VALID_COMMANDS: self._send(400, {"status": "error", "error": f"unknown command: {command!r}"}) return if source not in VALID_SOURCES: self._send(400, {"status": "error", "error": f"unknown source: {source!r}"}) return # Value-carrying commands: integers only. value = None if command in VALUE_COMMANDS: try: value = int(data.get("value", 0)) except (TypeError, ValueError): self._send(400, {"status": "error", "error": "value must be an integer"}) return with _lock: limit = (RATE_LIMIT_PER_MIN if source in _rate else OTHER_RATE_LIMIT_PER_MIN) if not _rate_ok(source if source in _rate else "_other", limit): limited = True else: limited = False if limited: self._send(429, {"status": "error", "error": "rate limited"}) return if command in RELAY_VALUE_COMMANDS: # Voice value command from the HOSTED relay path: queue it # as "name:value" so it survives the string-only command # channel; the app strips the value and routes it # (MainWindow._run_action). _enqueue("%s:%d" % (command, int(value or 0)), source) else: _enqueue(command, source, value) self._send(200, {"status": "ok", "queued": True}) return if path == "/status": if kind != "shared": self._send(403, {"status": "error", "error": "status requires the shared secret"}) return now = int(time.time()) with _lock: alive = [e for e in _queue if now - e["queued_at"] <= COMMAND_TTL] _queue[:] = alive blocked_ips = [bad_ip for bad_ip, ent in _auth_fail_by_ip.items() if ent[2] > now] self._send(200, { "status": "ok", "queue_depth": len(_queue), "rate_windows": {k: v for k, v in _rate.items()}, "auth_failures": _auth_failures, "blocked_ips": blocked_ips, }) return self._send(404, {"status": "error", "error": f"no such endpoint: {path}"}) def log_message(self, fmt, *args): # Keep journald clean: log nothing per-request (status covers ops). pass def main(): _load_tokens() server = ThreadingHTTPServer((LISTEN_HOST, LISTEN_PORT), RelayHandler) print(f"Waveline relay listening on {LISTEN_HOST}:{LISTEN_PORT}", flush=True) server.serve_forever() if __name__ == "__main__": main()