#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Waveline relay accounts: email+password accounts with verified email, browser sessions and device linking. DESIGN SOURCES (implemented, not invented): * OWASP Password Storage Cheat Sheet - scrypt with per-user salt; parameters from the OWASP configuration table (N=2^15, r=8, p=3); constant-time verification; password input length capped (DoS). * OWASP Authentication Cheat Sheet - generic, uniform responses to prevent user enumeration (login, registration, forgot-password all return the SAME body whether or not the account exists); dummy work on unknown accounts so timing does not leak; common-password blocklist (NIST SP800-63B); TLS-only cookies. * OWASP Session Management Cheat Sheet - opaque session IDs with >=256 bits of entropy from a CSPRNG, exchanged via HttpOnly+Secure+ SameSite=Lax cookie, meaningless value, all state server-side, stored ONLY as a SHA-256 hash (relay convention), sliding idle TTL, rotation on every login, invalidation on logout and on password reset. * OWASP Forgot Password Cheat Sheet - single-use URL tokens, stored hashed, short expiry, generic responses, no lockout, the account is never changed until a valid token is presented, all sessions are invalidated after a reset, confirmation mail never contains the password, reset URL host comes from configuration (never the Host header). * Relay house rules (see relay.py): standard library only; secrets live on disk only as hashes; a corrupt db must never take the relay down. ADDITIVE BY DESIGN: accounts only ever LINK existing pairing sessions (by their token hash). Deleting the accounts db must never cost anyone their pairing, their phone control or their cloud library. """ import hashlib import hmac import json import os import re import secrets import smtplib import socket import threading import time from email.message import EmailMessage from email.utils import formatdate, make_msgid from typing import Optional _EMAIL_ADDR_RE = re.compile(r"^[\w.+-]+@[\w-]+\.[\w.-]+$") # -------------------------------------------------------------------------- # Roles (5.1): owner > admin > helper > user. The owner is the operator # account (bootstrap via env/role file on first load); admins manage # users; helpers can read feedback. Roles live on the user record and # never unlock anything the operator has not wired to them. ROLES = ("owner", "admin", "helper", "user") # What each role may do on the relay (checked by relay endpoints): ROLE_PERMS = { "owner": frozenset({"roles.assign", "roles.read", "feedback.read", "feedback.reply", "users.list"}), "admin": frozenset({"roles.assign.limited", "roles.read", "feedback.read", "feedback.reply", "users.list"}), "helper": frozenset({"roles.read", "feedback.read"}), "user": frozenset(), } OWNER_ENV = "WAVELINE_OWNER_EMAIL" # -------------------------------------------------------------------------- # Parameters and policies (with their sources, so they are auditable) # -------------------------------------------------------------------------- SCRYPT_N = 2 ** 15 # 32 MiB memory cost - OWASP scrypt table row SCRYPT_R = 8 # OWASP: block size 8 SCRYPT_P = 3 # OWASP scrypt table row (N=2^15, r=8, p=3) PASSWORD_MAX = 1024 # bytes; cap before hashing (DoS guard) PASSWORD_MIN = 10 # no MFA on this service; blocklist compensates VERIFY_TTL = 24 * 3600 # email verification token: single-use, 24 h RESET_TTL = 3600 # password reset token: single-use, 1 hour SESSION_IDLE_TTL = 30 * 24 * 3600 # inactivity backstop for normal login PERSISTENT_TTL = 365 * 24 * 3600 # "remember me": 1 year, sliding # Settings sync: one blob per account, kept SMALL (it is a JSON object of # player preferences, not a library - media files live in the cloud lib). SETTINGS_MAX_BYTES = 16 * 1024 SETTINGS_MAX_DEPTH = 6 SETTINGS_STR_MAX = 200 # per-string cap in the cleaned blob # Security alerts: a queued event is re-sent for ALERT_RESEND_HOURS, but # never more often than ALERT_COOLDOWN_S (an attacker must not be able to # make the server email-bomb the admin mailbox). ALERT_RESEND_HOURS = 24 ALERT_COOLDOWN_S = 6 * 3600 EMAIL_RE = re.compile(r"^[A-Za-z0-9._%+-]{1,64}@[A-Za-z0-9.-]{1,190}\.[A-Za-z]{2,24}$") NICK_RE = re.compile(r"^[A-Za-z0-9_\- ]{2,40}$") # NIST SP800-63B / OWASP: block common passwords. Small embedded list: # no third-party calls (relay must stay dependency-free and offline). _COMMON = """ 123456 123456789 12345678 12345 1234567 1234567890 qwerty password p@ssword passw0rd password1 password12 password123 abcdef iloveyou administrator welcome monkey dragon letmein football baseball master sunshine princess shadow qazwsx trustno1 superman batman starwars whatever hello freedom passw ord admin admin123 admin@123 root toor login guest test test123 abc123 111111 000000 121212 654321 696969 123123 123321 1234 12345 qwerty123 1q2w3e4r 1qaz2wsx zaq12wsx qwe123 asdfgh zxcvbnm jesus christ ninja azerty soleilchocolate cookie Pok?mon pokemon minecraft fortnite letmein waveline your-domain yourdomain waveline2026 changeme default sample example secret pass pass1234! password! P@ssw0rd P@ssword1 Qwertyuiop qwertyuiop asdfghjkl zxcvbn asdf1234 987654321 11223344 """.split() BLOCKLIST = frozenset(_COMMON) # Environment (self-hostable; mirrors relay.py's env philosophy) ENV_DB = "WAVELINE_ACCOUNTS_DB" ENV_PUBLIC_URL = "WAVELINE_REMOTE_PUBLIC_URL" # fixed URL base, never Host DEFAULT_PUBLIC_URL = "https://your-domain/remote" ENV_SMTP_HOST = "WAVELINE_SMTP_HOST" ENV_SMTP_PORT = "WAVELINE_SMTP_PORT" ENV_SMTP_USER = "WAVELINE_SMTP_USER" ENV_SMTP_PASS = "WAVELINE_SMTP_PASS" ENV_SMTP_FROM = "WAVELINE_SMTP_FROM" ENV_SMTP_TLS = "WAVELINE_SMTP_TLS" # "starttls" (default) | "implicit" | "none" DEFAULT_SMTP = {"host": "127.0.0.1", "port": 587, "user": "waveline@your-domain", "pass": "", "from": "Waveline ", "tls": "starttls"} def _smtp_conf(): conf = dict(DEFAULT_SMTP) conf["host"] = os.environ.get(ENV_SMTP_HOST, conf["host"]) conf["port"] = int(os.environ.get(ENV_SMTP_PORT, conf["port"])) conf["user"] = os.environ.get(ENV_SMTP_USER, conf["user"]) conf["pass"] = os.environ.get(ENV_SMTP_PASS, conf["pass"]) conf["from"] = os.environ.get(ENV_SMTP_FROM, conf["from"]) conf["tls"] = os.environ.get(ENV_SMTP_TLS, conf["tls"]) return conf # -------------------------------------------------------------------------- # Password hashing (OWASP Password Storage) # -------------------------------------------------------------------------- def hash_password(password: str) -> str: """scrypt with a per-user random salt. Input capped; unicode kept.""" if isinstance(password, str): password = password.encode("utf-8") if len(password) > PASSWORD_MAX: # never silently truncate: raise ValueError("password too long") # reject instead (OWASP) salt = secrets.token_bytes(16) digest = hashlib.scrypt(password, salt=salt, n=SCRYPT_N, r=SCRYPT_R, p=SCRYPT_P, maxmem=64 * 1024 * 1024) return "scrypt$%d$%d$%d$%s$%s" % ( SCRYPT_N, SCRYPT_R, SCRYPT_P, salt.hex(), digest.hex()) def verify_password(password: str, stored: str) -> bool: """Constant-time verify; wrong format -> False (no crash, no oracle).""" try: if isinstance(password, str): password = password.encode("utf-8") if len(password) > PASSWORD_MAX: return False algo, n, r, p, salt_hex, digest_hex = stored.split("$") if algo != "scrypt": return False digest = hashlib.scrypt(password, salt=bytes.fromhex(salt_hex), n=int(n), r=int(r), p=int(p), maxmem=64 * 1024 * 1024) return hmac.compare_digest(digest.hex(), digest_hex) except (ValueError, TypeError): return False def password_problem(password: str): """Returns a reason string or None. NIST: length + blocklist, no composition rules; all characters allowed; no silent truncation.""" if not isinstance(password, str) or len(password) < PASSWORD_MIN: return "too-short" if len(password.encode("utf-8")) > PASSWORD_MAX: return "too-long" if password.strip().lower() in BLOCKLIST: return "too-common" return None def _token() -> str: """256-bit CSPRNG token, URL-safe (>=64-bit entropy required; we ship 4x the minimum).""" return secrets.token_urlsafe(32) def _token_hash(token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() # -------------------------------------------------------------------------- # Accounts store # -------------------------------------------------------------------------- VERIFY_RESEND_MIN = 60 # anti mail-bombing: one verify mail / min class Accounts: """Users, verification tokens, sessions. Persisted as HASHES ONLY. File layout (one JSON object; atomic replace like relay.py): users: {email_lower: {nick, pw, verified, created, last_login, link_code, devices, settings, ...}} pending: {token_hash: {email, exp}} - verify emails resets: {token_hash: {email, exp}} - password resets sessions:{token_hash: {email, exp, created}} - browser sessions events: {key: {email, kind, first, last, sent}} - security alerts """ def __init__(self, persist: bool = True, mail_alert=None): self._mail_alert = mail_alert # callable(kind, email) | None self._lock = threading.Lock() self._persist = persist self._path = os.environ.get( ENV_DB, os.path.join(os.path.expanduser("~"), ".waveline-relay", "accounts.json")) self._users = {} self._pending = {} self._resets = {} self._sessions = {} self._events = {} self._notifs = {} # email -> [notification dicts] self._feedback = [] # newest-first report store self._notif_seq = 0 # Pre-computed hash for the dummy path: keeps 'no such user' logins # the same cost as real ones (uniform timing, no oracle). self._dummy_pw = hash_password("dummy-password-for-uniform-timing") if persist: self._load() try: with self._lock: self._ensure_owner_locked() except Exception: pass # -- disk ------------------------------------------------------------- def _load(self): try: with open(self._path, "r", encoding="utf-8") as f: data = json.load(f) self._users = {str(k)[:190]: v for k, v in (data.get("users") or {}).items() if isinstance(v, dict)} self._pending = self._pruned(data.get("pending") or {}) self._resets = self._pruned(data.get("resets") or {}) self._sessions = self._pruned(data.get("sessions") or {}) self._events = self._pruned(data.get("events") or {}) self._notifs = { str(k)[:190]: [n for n in (v or []) if isinstance(n, dict)] for k, v in (data.get("notifs") or {}).items()} self._notif_seq = int(data.get("notif_seq") or 0) self._feedback = self._feedback_pruned([ r for r in (data.get("feedback") or []) if isinstance(r, dict)])[:self.FEEDBACK_MAX] except FileNotFoundError: pass except Exception: try: os.replace(self._path, self._path + ".corrupt") except Exception: pass self._users, self._pending = {}, {} self._resets, self._sessions, self._events = {}, {}, {} self._notifs, self._notif_seq = {}, 0 self._feedback = [] @staticmethod def _pruned(table): now = time.time() return {th: v for th, v in table.items() if isinstance(v, dict) and float(v.get("exp", 0)) > now} def _save_locked(self): if not self._persist: return data = {"users": self._users, "pending": self._pending, "resets": self._resets, "sessions": self._sessions, "events": self._events, "notifs": self._notifs, "notif_seq": self._notif_seq, "feedback": self._feedback} try: os.makedirs(os.path.dirname(self._path), exist_ok=True) tmp = self._path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f) os.replace(tmp, self._path) except Exception: pass # best-effort, exactly like relay.py # -- in-app notifications (5.1) ---------------------------------------- NOTIF_MAX_PER_USER = 40 # ring buffer per account NOTIF_TTL = 14 * 86400 # seconds; older entries drop on drain def notify_user(self, email: str, kind: str, title: str, body: str = "", ticket: str = "") -> None: """Queue one in-app notification for an account (delivered by the app's /app/sync poll). Best-effort: never raises.""" try: email = str(email or "").lower() with self._lock: user = self._users.get(email) if user is None or not user.get("verified"): return self._notif_seq += 1 box = self._notifs.setdefault(email, []) box.append({ "id": self._notif_seq, "kind": str(kind or "info")[:20], "title": str(title or "")[:120], "body": str(body or "")[:2000], "ticket": str(ticket or "")[:24], "at": time.time(), }) del box[:-self.NOTIF_MAX_PER_USER] self._save_locked() except Exception: pass def notify_role(self, role: str, kind: str, title: str, body: str = "", ticket: str = "") -> int: """Queue for every verified account holding this role. Returns how many accounts were queued.""" n = 0 try: with self._lock: targets = [e for e, u in self._users.items() if u.get("verified") and str(u.get("role", "user")) == role] for e in targets: self.notify_user(e, kind, title, body, ticket) n += 1 except Exception: pass return n def drain_notifications(self, email: str): """Pop all pending notifications for an account (app poll). Removes delivered ones; a cursor-less firehose by design.""" out = [] try: with self._lock: box = self._notifs.get(str(email or "").lower()) if box: now = time.time() out = [n for n in box if now - n["at"] < self.NOTIF_TTL] self._notifs[str(email).lower()] = [] self._save_locked() except Exception: pass return out def count_notifications(self, email: str) -> int: try: with self._lock: return len(self._notifs.get(str(email or "").lower()) or []) except Exception: return 0 # -- feedback reports store (5.1) -------------------------------------- # Reports persist here so admins/helpers can read them in-app # without server access. Ring buffer + TTL keep the db bounded. FEEDBACK_MAX = 200 FEEDBACK_TTL = 90 * 86400 FEEDBACK_LOG_MAX = 300_000 # stored snapshot cap FEEDBACK_MAX_CHARS = 20000 # same cap Mailer enforces on send def feedback_add(self, ticket: str, kind: str, contact: str, message: str, version: str = "", log_snapshot: str = "") -> None: try: with self._lock: self._feedback.insert(0, { "ticket": str(ticket or "")[:24], "kind": str(kind or "other")[:20], "contact": str(contact or "")[:200], "message": str(message or "")[:self.FEEDBACK_MAX_CHARS], "version": str(version or "")[:40], "log": str(log_snapshot or "")[:self.FEEDBACK_LOG_MAX], "at": time.time(), }) del self._feedback[self.FEEDBACK_MAX:] self._save_locked() except Exception: pass def _feedback_pruned(self, rows): cutoff = time.time() - self.FEEDBACK_TTL return [r for r in rows if float(r.get("at", 0)) >= cutoff] def feedback_list(self, actor_email: str, limit: int = 100): """Summary rows, newest first; None when not permitted.""" if "feedback.read" not in self.perms_of(actor_email): return None with self._lock: rows = self._feedback_pruned(self._feedback)[:max(1, limit)] return [{"ticket": r["ticket"], "kind": r["kind"], "contact": r["contact"], "version": r["version"], "preview": r["message"][:120], "at": r["at"]} for r in rows] def feedback_get(self, actor_email: str, ticket: str): """One full report (incl. log snapshot); None when not permitted or missing.""" if "feedback.read" not in self.perms_of(actor_email): return None with self._lock: for r in self._feedback: if r["ticket"] == str(ticket or "").strip().upper(): out = dict(r) out["at"] = float(r.get("at", 0)) return out return None # -- roles (5.1) ------------------------------------------------------- def _ensure_owner_locked(self): """Bootstrap the OWNER account once. The env email must already be a registered+verified account; on first load it is promoted to owner. Nothing happens (silently) if it is not registered yet - the check re-runs on every load/login.""" owner = os.environ.get(OWNER_ENV, "").strip().lower() if not owner: return user = self._users.get(owner) if user is None or not user.get("verified"): return if user.get("role") != "owner": user["role"] = "owner" self._save_locked() def role_of(self, email: str) -> str: with self._lock: user = self._users.get(str(email or "").lower()) if user is None: return "user" return str(user.get("role", "user")) def perms_of(self, email: str) -> frozenset: return ROLE_PERMS.get(self.role_of(email), ROLE_PERMS["user"]) def list_users(self, email: str): """[(email, nick, role, verified)] - role-bearing eyes only.""" if "users.list" not in self.perms_of(email): return None with self._lock: return [(e, u.get("nick", ""), str(u.get("role", "user")), bool(u.get("verified"))) for e, u in sorted(self._users.items())] def set_role(self, actor_email: str, target_email: str, role: str): """Assign a role. Returns (ok, problem). Rules: only owner/ admin with the assign perm; nobody changes their own role; only owner may create/modify admins or the owner; target must exist; role must be a known one.""" actor = str(actor_email or "").lower() target = str(target_email or "").strip().lower() role = str(role or "").strip().lower() perms = self.perms_of(actor) if "roles.assign" not in perms and "roles.assign.limited" not in perms: return False, "not allowed" if role not in ROLES: return False, "unknown role" if target == actor: return False, "cannot change your own role" if role in ("admin", "owner") and "roles.assign" not in perms: return False, "only the owner can assign that role" with self._lock: tgt = self._users.get(target) if tgt is None or not tgt.get("verified"): return False, "no such verified account" if tgt.get("role") == "owner" and actor != target: return False, "the owner role cannot be changed" tgt["role"] = role self._save_locked() return True, "" # -- registration ----------------------------------------------------- def register(self, email: str, nickname: str, password: str): """Returns (token_or_None, generic_public_message). ALWAYS a generic message: creating a real user, re-sending a verification mail for an existing one, or hitting a policy failure are indistinguishable from the outside (OWASP anti- enumeration). Policy failures still return None so the caller can show a local, non-enumerating hint ONLY when the problem is with what the user just typed (password quality), never about whether an account exists.""" email = str(email or "").strip().lower() nickname = str(nickname or "").strip() password = str(password or "") if (not EMAIL_RE.match(email) or not NICK_RE.match(nickname) or len(password) > PASSWORD_MAX): return None, "" problem = password_problem(password) with self._lock: existing = self._users.get(email) if existing is not None: if existing.get("verified"): # Re-send verification would leak existence; do not. return None, "" # Unverified duplicate: re-send (same generic HTTP answer # as a fresh signup; the token travels only by email), # throttled to one mail per minute (anti mail-bombing). if time.time() - float(existing.get("last_verify_sent", 0)) < VERIFY_RESEND_MIN: return None, "" token = _token() self._pending[_token_hash(token)] = { "email": email, "exp": time.time() + VERIFY_TTL} existing["last_verify_sent"] = time.time() self._save_locked() return token, "" if problem is not None: # Policy is about the typed input, safe to surface. return None, problem self._users[email] = { "nick": nickname, "pw": hash_password(password), "verified": False, "created": time.time(), "last_login": 0.0, "last_verify_sent": time.time(), } token = _token() self._pending[_token_hash(token)] = { "email": email, "exp": time.time() + VERIFY_TTL} self._save_locked() return token, "" def resend(self, email: str): """Verification re-send. Returns a token ONLY for an existing UNVERIFIED account, throttled to one per minute; None otherwise, so the endpoint can answer generically.""" email = str(email or "").strip().lower() with self._lock: user = self._users.get(email) if user is None or user.get("verified"): return None if time.time() - float(user.get("last_verify_sent", 0)) < VERIFY_RESEND_MIN: return None token = _token() self._pending[_token_hash(token)] = { "email": email, "exp": time.time() + VERIFY_TTL} user["last_verify_sent"] = time.time() self._save_locked() return token def verify(self, token: str): """Consume a verification token (single-use). True on success.""" th = _token_hash(str(token or "")) with self._lock: row = self._pending.pop(th, None) if not row or float(row.get("exp", 0)) < time.time(): self._save_locked() return False user = self._users.get(row["email"]) if user is None: self._save_locked() return False user["verified"] = True self._record_event_locked("verify", row["email"]) # Owner bootstrap: if the just-verified account IS the # configured owner email, promote it on the spot. self._ensure_owner_locked() self._save_locked() return True # -- login / sessions ------------------------------------------------- def login(self, email: str, password: str, remember: bool = False): """Returns a NEW opaque session token or None. Unknown accounts burn the same scrypt work as real ones (uniform timing). remember=False -> browser-session lifetime (dies with the browser, 30-day inactivity backstop); remember=True -> the way every major site does it: a persistent cookie kept alive by activity (1-year sliding server-side TTL).""" email = str(email or "").strip().lower() with self._lock: user = self._users.get(email) if user is not None and user.get("verified"): ok = verify_password(password, user["pw"]) else: ok = verify_password(password, self._dummy_pw) if not ok or user is None: return None token = _token() ttl = PERSISTENT_TTL if remember else SESSION_IDLE_TTL self._sessions[_token_hash(token)] = { "email": email, "exp": time.time() + ttl, "created": time.time(), "remember": bool(remember)} user["last_login"] = time.time() self._save_locked() return token def auth_session(self, token: str): """Session -> user dict (email, nick) or None. Sliding TTL.""" th = _token_hash(str(token or "")) with self._lock: row = self._sessions.get(th) if not row: return None now = time.time() if float(row.get("exp", 0)) < now: del self._sessions[th] self._save_locked() return None user = self._users.get(row["email"]) if user is None or not user.get("verified"): return None # Sliding refresh honours the session's own kind: a remembered # login keeps rolling its year; a plain login keeps its # 30-day inactivity window. refresh = PERSISTENT_TTL if row.get("remember") else SESSION_IDLE_TTL row["exp"] = now + refresh self._save_locked() return {"email": row["email"], "nick": user.get("nick", "")} def logout(self, token: str): th = _token_hash(str(token or "")) with self._lock: existed = self._sessions.pop(th, None) is not None self._save_locked() return existed # -- forgot password (OWASP Forgot Password) -------------------------- def request_reset(self, email: str): """Token for an existing verified account, else None - the caller answers identically either way. No lockout, no change here.""" email = str(email or "").strip().lower() with self._lock: user = self._users.get(email) if user is None or not user.get("verified"): return None token = _token() self._resets[_token_hash(token)] = { "email": email, "exp": time.time() + RESET_TTL} self._save_locked() return token def reset(self, token: str, new_password: str): """Consume reset token, set the new password, kill ALL sessions for that user (OWASP: invalidate existing sessions).""" problem = password_problem(str(new_password or "")) if problem is not None: return problem th = _token_hash(str(token or "")) with self._lock: row = self._resets.pop(th, None) if not row or float(row.get("exp", 0)) < time.time(): self._save_locked() return "bad-token" user = self._users.get(row["email"]) if user is None: self._save_locked() return "bad-token" user["pw"] = hash_password(new_password) email = row["email"] self._sessions = {t: s for t, s in self._sessions.items() if s.get("email") != email} self._record_event_locked( "reset", email, "password reset; all sessions signed out") self._save_locked() return "" # -- device linking ---------------------------------------------------- def issue_link_code(self, email: str): """Short code the user types INTO THE APP to link that pairing to this account. 8 chars, 15 min, single-use, stored hashed.""" email = str(email or "").strip().lower() with self._lock: user = self._users.get(email) if user is None or not user.get("verified"): return None code = "".join(secrets.choice("ABCDEFGHJKLMNPQRSTUVWXYZ23456789") for _ in range(8)) user["link_code"] = {"hash": _token_hash(code), "exp": time.time() + 15 * 60} self._save_locked() return code def consume_link_code(self, code: str, token_hash: str, pc_name: str = ""): """App side: exchange the typed code for account linkage. The pairing is remembered by its TOKEN HASH only. pc_name labels the computer (multi-PC dashboards; defaults per-index).""" h = _token_hash(str(code or "")) with self._lock: for email, user in self._users.items(): lc = user.get("link_code") if not lc: continue if float(lc.get("exp", 0)) < time.time(): del user["link_code"] continue if hmac.compare_digest(lc["hash"], h): del user["link_code"] key = token_hash[:24] user.setdefault("devices", {}) idx = len(user["devices"]) + 1 label = (str(pc_name or "").strip()[:40] or "My computer %d" % idx) user["devices"][key] = { "th": token_hash, "at": time.time(), "label": label} self._record_event_locked( "link", email, "computer '%s' linked" % label) self._save_locked() return user.get("nick", "") self._save_locked() return None def linked(self, email: str): with self._lock: user = self._users.get(str(email or "").lower()) if not user: return [] return [v["th"] for v in (user.get("devices") or {}).values()] def linked_full_emails(self): """[(email_lower, token_hash), ...] across ALL accounts - the reverse index used for app-authed lookups (settings sync, pair alerts). No secrets: token hashes only.""" with self._lock: out = [] for email, user in self._users.items(): for v in (user.get("devices") or {}).values(): out.append((email, str(v.get("th", "")))) return out def linked_full(self, email: str): """[(token_hash, label), ...] in link order (multi-PC).""" with self._lock: user = self._users.get(str(email or "").lower()) if not user: return [] devs = sorted((user.get("devices") or {}).items(), key=lambda kv: kv[1].get("at", 0)) return [(v["th"], v.get("label", "My computer")) for _, v in devs] def rename_pc(self, email: str, token_hash: str, label: str) -> bool: """Owner renames one of their linked computers.""" clean = str(label or "").strip()[:40] if not clean: return False with self._lock: user = self._users.get(str(email or "").lower()) if not user: return False for v in (user.get("devices") or {}).values(): if hmac.compare_digest(str(v.get("th", "")), str(token_hash or "")): v["label"] = clean self._save_locked() return True return False def unlink(self, email: str, token_hash: str): with self._lock: user = self._users.get(str(email or "").lower()) if not user or "devices" not in user: return False removed = user["devices"].pop(token_hash[:24], None) self._save_locked() return removed is not None # -- per-account settings sync (small JSON blob) ----------------------- @staticmethod def _clean_settings(v, depth: int = 0): """Return a JSON-safe, size-bounded copy: dict/list scalars only, each string capped, depth-capped, bools/ints/floats kept.""" if depth > SETTINGS_MAX_DEPTH: return None if isinstance(v, bool) or isinstance(v, int) or isinstance(v, float): return v if isinstance(v, str): return v[:SETTINGS_STR_MAX] if isinstance(v, list): out = [] for x in v[:64]: cv = Accounts._clean_settings(x, depth + 1) if cv is not None: out.append(cv) return out if isinstance(v, dict): out = {} for k in list(v.keys())[:64]: cv = Accounts._clean_settings(v[k], depth + 1) if cv is not None: out[str(k)[:40]] = cv return out return None def settings_put(self, email: str, blob) -> bool: """Store the account's settings blob (whole-blob replacement). Oversized blobs are refused (the client falls back to local).""" clean = self._clean_settings(blob) if clean is None: return False raw = json.dumps(clean).encode("utf-8") if len(raw) > SETTINGS_MAX_BYTES: return False with self._lock: user = self._users.get(str(email or "").lower()) if user is None or not user.get("verified"): return False user["settings"] = clean user["settings_at"] = time.time() self._save_locked() return True def settings_get(self, email: str): """The account's stored blob, or None. Read-only: never touches what the user typed into THIS device (merge policy is client-side).""" with self._lock: user = self._users.get(str(email or "").lower()) if user is None or not user.get("verified"): return None return user.get("settings") # -- security events (admin mailbox alerts) ----------------------------- def _record_event_locked(self, kind: str, email: str, detail: str = "", sent: float = 0.0) -> None: """Core of record_security_event; the CALLER holds self._lock (threading.Lock is not reentrant - calling the public wrapper from inside a locked section would deadlock).""" kind = str(kind or "")[:40] email = str(email or "").strip().lower() if not kind or not email: return now = time.time() key = "%s|%s" % (kind, email) row = self._events.get(key) if row is None: self._events[key] = {"email": email, "kind": kind, "first": now, "last": now, "sent": float(sent or 0.0), "detail": str(detail)[:160]} else: row["last"] = now row["detail"] = str(detail or row.get("detail", ""))[:160] def record_security_event(self, kind: str, email: str, detail: str = "", sent: float = 0.0) -> None: """Remember that ``kind`` happened for ``email``. Returns nothing and never raises: alerting is best-effort by design (the real action - verify, reset, link - has already succeeded). ``sent`` pre-marks the event as mailed (immediate-queue path); the default 0.0 means "never sent" -> the drain picks it up. ``drain_security_alerts`` turns NEW events into admin emails, throttled by ALERT_COOLDOWN_S so abuse cannot email-bomb.""" with self._lock: self._record_event_locked(kind, email, detail, sent) self._save_locked() def email_for_device(self, token_hash: str) -> str: """Account email that has ``token_hash`` linked (first match), for alerting the owner when a control device pairs.""" th = str(token_hash or "") with self._lock: for email, user in self._users.items(): for d in (user.get("devices") or {}).values(): if hmac.compare_digest(str(d.get("th", "")), th): return email return "" def nick_for(self, email: str) -> str: """Account nickname for display ("" when unknown).""" with self._lock: user = self._users.get(str(email or "").lower()) return str(user.get("nick", "")) if user else "" def drain_security_alerts(self): """[(kind, email, detail), ...] for events that are new or were last sent more than ALERT_COOLDOWN_S ago AND first seen within ALERT_RESEND_HOURS. Marks them sent. Mail delivery itself is the caller's job (relay.py posts it from a background thread).""" now = time.time() out = [] with self._lock: for key, row in list(self._events.items()): if now - float(row.get("first", 0)) > ALERT_RESEND_HOURS * 3600: del self._events[key] continue if now - float(row.get("sent", 0)) < ALERT_COOLDOWN_S: continue row["sent"] = now out.append((row.get("kind", ""), row.get("email", ""), row.get("detail", ""))) if out: self._save_locked() return out def unlink_by_hash(self, token_hash: str) -> int: """Remove a pairing's linkage everywhere (app-driven unpair). Returns how many accounts it was linked from (0/1).""" th = str(token_hash or "") n = 0 with self._lock: for email, user in self._users.items(): devs = user.get("devices") or {} stale = [k for k, d in devs.items() if hmac.compare_digest(str(d.get("th", "")), th)] for k in stale: devs.pop(k) n += 1 self._record_event_locked( "unlink", email, "computer unlinked (app request)") if n: self._save_locked() return n # -------------------------------------------------------------------------- # Mailer (verification + reset emails over the operator's own SMTP) # -------------------------------------------------------------------------- class Mailer: """Plain-text mail via the operator's SMTP. Every knob is an env var so a self-hoster points it at their own server; failures are logged, never fatal (tokens can be re-sent).""" def __init__(self): self.conf = _smtp_conf() self.public_url = os.environ.get( ENV_PUBLIC_URL, DEFAULT_PUBLIC_URL).rstrip("/") def _send(self, to: str, subject: str, body: str) -> bool: conf = self.conf try: msg = EmailMessage() msg["From"] = conf["from"] msg["To"] = to msg["Subject"] = subject msg.set_content(body) if conf["tls"] == "implicit": client = smtplib.SMTP_SSL(conf["host"], conf["port"], timeout=15) else: client = smtplib.SMTP(conf["host"], conf["port"], timeout=15) try: if conf["tls"] == "starttls": client.starttls() if conf["user"] and conf["pass"]: client.login(conf["user"], conf["pass"]) client.send_message(msg) finally: try: client.quit() except Exception: pass return True except (OSError, smtplib.SMTPException, socket.timeout): return False def send_verify(self, to: str, token: str) -> bool: # public_url already ends with the ingress path (/remote): the # account routes hang directly below it. (A doubled /remote/ # remote here produced relay 404s when users clicked - fixed.) link = "%s/account/verify?token=%s" % (self.public_url, token) body = ("Welcome to Waveline!\n\n" "Confirm this email address to finish creating your account:\n\n" " %s\n\n" "The link works once and expires in 24 hours. If you did not\n" "sign up for Waveline, you can ignore this email.\n" % link) return self._send(to, "Waveline: confirm your email", body) def send_reset(self, to: str, token: str) -> bool: link = "%s/account/reset?token=%s" % (self.public_url, token) body = ("Someone (hopefully you) asked to reset the Waveline account\n" "password for this address.\n\n" "Set a new password here:\n\n" " %s\n\n" "The link works once and expires in 1 hour.\n" "If this was not you, ignore this email - nothing was changed.\n" % link) return self._send(to, "Waveline: reset your password", body) def send_reset_done(self, to: str) -> bool: return self._send(to, "Waveline: password was changed", "Your Waveline account password was just changed.\n" "All browser sessions were signed out.\n") # -- operator security alerts (admin mailbox) --------------------------- ALERT_TO = "admin@your-domain" # the operator's own mailbox _ALERT_KINDS = { "verify": "account email verified", "reset": "password reset", "link": "a computer was linked to an account", "unlink": "a computer was unlinked", "pair": "a control device paired", } def send_alert(self, kind: str, email: str, detail: str = "") -> bool: """Short plain-text notice to the OPERATOR (not the user). A burst of events is normal - delivery is throttled upstream (ALERT_COOLDOWN_S) so the mailbox cannot be bombed through us.""" label = self._ALERT_KINDS.get(kind, "security event") body = ("Waveline security notice\n\n" "Event: %s\n" "Account: %s\n" % (label, email or "(unknown)")) if detail: body += "Detail: %s\n" % detail body += ("\nRepeat notices for the same event are throttled to one\n" "per 6 hours and stop after 24 hours.\n") return self._send(self.ALERT_TO, "Waveline security: %s" % label, body) def send_digest(self, email: str, top: list, hours: float, served: int, links: int) -> bool: """The account's weekly listening digest (plain text).""" def _h(seconds: float) -> str: m = int(round(seconds // 60)) return "%d h %02d min" % (m // 60, m % 60) if m >= 60 \ else "%d min" % m lines = ["Your Waveline week", "", "Total listening: %s" % _h(hours), ""] if top: lines.append("Top tracks:") for i, (name, secs) in enumerate(top, 1): lines.append(" %d. %s - %s" % (i, name, _h(secs))) else: lines.append("(No listening recorded this week.)") lines += ["", "Guests requested %d track(s) via your jukebox " "share this week." % served, "Your account has %d computer(s) linked." % links, "", "Listen anywhere at %s/account" % self.public_url] return self._send(email, "Your Waveline week", "\n".join(lines)) # -- user feedback (5.1) --------------------------------------------- # Both destinations get the same message: your personal # mailbox and the waveline@ role mailbox. FEEDBACK_TO = ("you@your-domain", "waveline@your-domain") FEEDBACK_MAX_CHARS = 20000 FEEDBACK_MAX_LOG_CHARS = 300_000 # snapshot cap (client redacts too) @staticmethod def new_ticket_id() -> str: """Short human-friendly ticket id: WL-XXXXX-XXXXX.""" alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # no 0/O/1/I raw = "".join(secrets.choice(alphabet) for _ in range(10)) return "WL-%s-%s" % (raw[:5], raw[5:]) def send_feedback(self, ticket: str, kind: str, contact: str, message: str, version: str = "", log_snapshot: str = "") -> Optional[str]: """One feedback item to BOTH operator addresses. With a log snapshot, the mail is multipart with the snapshot attached as waveline-log.txt. Returns the notification's Message-ID on success (so the ack and later replies can thread under it), None on failure. Reply-To is the sender's contact, so hitting Reply in Gmail or webmail answers the user directly.""" subject = "Waveline %s %s" % (kind, ticket) body = ( "Waveline feedback %s\n" "Kind: %s\n" "App version: %s\n" "Reply-to contact: %s\n" "Log attached: %s\n" "\nMessage:\n%s\n" % (ticket, kind, version or "(not given)", contact or "(not given)", "yes" if log_snapshot else "no", (message or "(empty)")[:self.FEEDBACK_MAX_CHARS])) conf = self.conf try: msg = EmailMessage() msg["From"] = conf["from"] msg["To"] = ", ".join(self.FEEDBACK_TO) msg["Subject"] = subject msg["Date"] = formatdate(localtime=True) msg["Message-ID"] = make_msgid(domain="your-domain") if contact and _EMAIL_ADDR_RE.match(contact): msg["Reply-To"] = contact msg.set_content(body) if log_snapshot: snap = log_snapshot[:self.FEEDBACK_MAX_LOG_CHARS] msg.add_attachment( snap.encode("utf-8"), maintype="text", subtype="plain", filename="waveline-log.txt") if conf["tls"] == "implicit": client = smtplib.SMTP_SSL(conf["host"], conf["port"], timeout=15) else: client = smtplib.SMTP(conf["host"], conf["port"], timeout=15) try: if conf["tls"] == "starttls": client.starttls() if conf["user"] and conf["pass"]: client.login(conf["user"], conf["pass"]) client.send_message(msg) finally: try: client.quit() except Exception: pass return msg["Message-ID"] except (OSError, smtplib.SMTPException, socket.timeout): return None def send_feedback_ack(self, ticket: str, to: str, in_reply_to: str = "") -> bool: """Auto-acknowledgement to the person who sent feedback (only when they left a reply address). Carries the ticket id so both sides can reference it; In-Reply-To/References thread it under the operator's notification copy; Reply-To points at the role mailbox so a user reply lands where the operator reads mail.""" body = ("Thank you for your Waveline feedback!\n\n" "Your reference number is %s - keep it if you want to " "refer to this again.\n" "Your message is in the developer's inbox right now; " "every piece of feedback is read.\n\n" "If you did not send feedback, ignore this email and " "nothing further will happen.\n" % ticket) conf = self.conf try: msg = EmailMessage() msg["From"] = conf["from"] msg["To"] = to msg["Reply-To"] = "waveline@your-domain" msg["Subject"] = "Waveline: we got your feedback (%s)" % ticket if in_reply_to: msg["In-Reply-To"] = in_reply_to msg["References"] = in_reply_to msg.set_content(body) if conf["tls"] == "implicit": client = smtplib.SMTP_SSL(conf["host"], conf["port"], timeout=15) else: client = smtplib.SMTP(conf["host"], conf["port"], timeout=15) try: if conf["tls"] == "starttls": client.starttls() if conf["user"] and conf["pass"]: client.login(conf["user"], conf["pass"]) client.send_message(msg) finally: try: client.quit() except Exception: pass return True except (OSError, smtplib.SMTPException, socket.timeout): return False