#!/usr/bin/env python3 """The remote INBOX: the user chats with the two Stickbash coordinators from inbox.html (local or https://bashatest.howincloud.com/inbox.html), this tiny server stores everything on the VM, and each coordinator picks its messages up over ssh (_tools/inbox.sh) and posts replies, questions and status lines back. Python stdlib only. Two coordinators ("lanes"): kills (the kill coordinator: kills, builds, engine, the game app) and live (the live-game coordinator: online, shop, seasons, saves, accounts). The user addresses a message to kills, live or both. python3 inbox_server.py serve [port] the HTTP API on 127.0.0.1 (Caddy proxies /api/inbox* to it) python3 inbox_server.py pull [--to LANE] NEW user messages for LANE (default kills) as JSON, marked seen by LANE python3 inbox_server.py reply ID [--to LANE] reply text on stdin; threads under message ID, marks it replied python3 inbox_server.py status [--to LANE] LANE's "now" status line on stdin (shown on top of the page) python3 inbox_server.py ask JSON on stdin {to, topic, text, choices[], multi?}: a question for the user (multi: true = the user may tick several choices; needs 2+ choices) python3 inbox_server.py post JSON on stdin {to, topic, text}: a note to the user (no answer needed) python3 inbox_server.py answers [--to LANE] answers to LANE's questions not fetched before, marked fetched python3 inbox_server.py unask ID withdraw an open question python3 inbox_server.py topics the topic list as JSON python3 inbox_server.py list everything, as JSON python3 inbox_server.py push-init make the VAPID key (vapid_private.pem, 0600) if missing; prints the PUBLIC key only python3 inbox_server.py push-status push on/off, subscription count, queue, last send per lane (no secrets) python3 inbox_server.py push-test a test notification to every subscribed device now INBOX_DIR=/some/dir overrides where data.json + token live (tests). PUSH (Web Push to the user's phone / desktop; the sender is inbox_push.py next to this file, which needs the `cryptography` package; without it or without vapid_private.pem push is simply off and nothing else changes): a coordinator's reply, ask or post (CLI) queues a push in push.json (0600; status lines never push); the `serve` process sends it: at most one per PUSH_GAP (90 s) per lane, the rest of that window merged into "N new from LIVE". Subscriptions that answer 404/410 are dropped. HTTP (every call needs the header X-Inbox-Token = the token in ./token): GET /api/inbox[?rev=N] -> {rev, status, statusTs, statuses:{kills:{text,ts},live:{text,ts}}, topics:[...], messages:[...]} (with ?rev=N equal to the current rev: just {rev, unchanged:true}) POST /api/inbox {text, topic?, to?, parent?} -> adds a user message (old pages send only {text}: topic general, to kills) POST /api/inbox/answer {id, answer} -> the user answers question id (a single question: answer = text, one of the choices or typed; a multi question: answer = a list of choices plus at most one typed item; one string is taken as a one-item list) POST /api/inbox/topic {title} -> creates (or returns the existing) topic POST /api/inbox/remove {id} -> the user REMOVES message id from the page: hidden: true (+ hiddenTs); an OPEN question is also withdrawn (state withdrawn, withdrawnBy user, withdrawnTs). Nothing is deleted; never pushes. POST /api/inbox/restore {id} -> undoes a remove: hidden: false; a question the USER withdrew (and never answered) opens again. Never pushes. GET /api/inbox/push/vapid -> {enabled, publicKey} (the VAPID public key for pushManager.subscribe) POST /api/inbox/push/subscribe {subscription:{endpoint, keys:{p256dh, auth}}, ua?} -> stores this device POST /api/inbox/push/unsubscribe {endpoint} -> forgets this device POST /api/inbox/push/test {endpoint?} -> a test notification now (to that device, or to all) Message fields: id, ts, kind (message|question|note), from (user|kills|live), to (kills|live|both|user), topic, parent, text, state (message: new|seen|replied; question: open|answered|withdrawn; note: posted), seenTs, seenBy{lane:ts}, replies[{by,text,ts}], reply + replyTs (the latest coordinator reply, for the old pages), choices[], answer, answeredTs; a multi-choice question also has multi: true and its answer is a list (single questions keep answer as text, no multi). Added by the page's REMOVE (3 Nov; absent = never removed): hidden (bool), hiddenTs; on a question the user withdrew: withdrawnBy: "user", withdrawnTs, withdrawnPulled (answers --to LANE reports it once, on stderr only, never as an answer). """ import fcntl, hmac, json, os, re, sys, threading, time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import urlparse, parse_qs HERE = os.environ.get('INBOX_DIR') or os.path.dirname(os.path.abspath(__file__)) DATA = os.path.join(HERE, 'data.json') TOKEN_FILE = os.path.join(HERE, 'token') PUSH_FILE = os.path.join(HERE, 'push.json') # subscriptions + the send queue (0600) VAPID_FILE = os.path.join(HERE, 'vapid_private.pem') # the VAPID signing key (0600, made on the VM by push-init) PUSH_GAP = float(os.environ.get('INBOX_PUSH_GAP') or 90) # seconds between two pushes of one lane (tests shorten it) PUSH_TICK, PUSH_STALE, MAX_SUBS, MAX_QUEUE = 2.0, 6 * 3600, 20, 200 MAX_TEXT, MAX_MESSAGES, MAX_BODY = 6000, 3000, 64 * 1024 MAX_TITLE, MAX_CHOICES, MAX_CHOICE, MAX_TOPICS = 120, 8, 200, 300 LANES = ('kills', 'live') TO_USER = LANES + ('both',) SCHEMA = 2 GENERAL = {'id': 'general', 'title': 'General', 'ts': '', 'by': 'system'} PAGES = {'id': 'page-sends', 'title': 'Sent from pages', 'ts': '', 'by': 'system'} PAGE_SEND = re.compile(r'^[A-Z][A-Z0-9 _/-]{2,40} v\d+\b') # RATINGS v1 / DOUBT v1 / STYLE PICKS v1 ... (page buttons) def now(): return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) # ISO UTC: the page shows it in the viewer's local time def slug(title): s = re.sub(r'[^a-z0-9]+', '-', title.lower()).strip('-')[:48] return s or 'topic' def migrate(d): """Bring any older store (v1: {status, statusTs, next, messages[{id,ts,text,state,reply,replyTs}]}) up to SCHEMA.""" d.setdefault('status', '') d.setdefault('statusTs', '') d.setdefault('next', 1) d.setdefault('messages', []) d.setdefault('rev', 0) st = d.setdefault('statuses', {}) for lane in LANES: st.setdefault(lane, {'text': '', 'ts': ''}) if d['status'] and not st['kills']['text']: st['kills'] = {'text': d['status'], 'ts': d['statusTs']} topics = d.setdefault('topics', []) ids = {t['id'] for t in topics} for t in (GENERAL, PAGES): if t['id'] not in ids: topics.insert(0 if t is GENERAL else 1, dict(t)) for m in d['messages']: m.setdefault('kind', 'message') m.setdefault('from', 'user') m.setdefault('to', 'kills') if 'topic' not in m: m['topic'] = PAGES['id'] if m['kind'] == 'message' and PAGE_SEND.match(m.get('text', '')) else 'general' m.setdefault('parent', None) m.setdefault('reply', '') m.setdefault('replyTs', '') if 'replies' not in m: m['replies'] = [{'by': 'kills', 'text': m['reply'], 'ts': m['replyTs']}] if m['reply'] else [] if 'seenBy' not in m: seen = m['replyTs'] or (m['ts'] if m.get('state') in ('seen', 'replied') else '') m['seenBy'] = {'kills': seen} if seen else {} m.setdefault('seenTs', min(m['seenBy'].values()) if m['seenBy'] else '') d['v'] = SCHEMA return d def load(): try: with open(DATA) as f: d = json.load(f) except (FileNotFoundError, json.JSONDecodeError): d = {} return migrate(d) def update(fn): """Read-modify-write data.json under an exclusive lock, written atomically (no torn file on a crash).""" with open(DATA + '.lock', 'w') as lk: fcntl.flock(lk, fcntl.LOCK_EX) d = load() out = fn(d) d['rev'] = d.get('rev', 0) + 1 tmp = DATA + '.tmp' with open(tmp, 'w') as f: json.dump(d, f, ensure_ascii=False, indent=1) os.replace(tmp, DATA) return out def token(): with open(TOKEN_FILE) as f: return f.read().strip() def clean(s, limit): return str(s if s is not None else '').replace('\r\n', '\n').strip()[:limit] def find_topic(d, ref, create_by=None): """A topic by id or title (case-insensitive); created when create_by is given and it does not exist.""" ref = clean(ref, MAX_TITLE) if not ref: return 'general' low = ref.lower() for t in d['topics']: if t['id'] == low or t['title'].lower() == low: return t['id'] if not create_by: return None if len(d['topics']) >= MAX_TOPICS: return 'general' base, ids = slug(ref), {t['id'] for t in d['topics']} tid, n = base, 2 while tid in ids: tid, n = '%s-%d' % (base, n), n + 1 d['topics'].append({'id': tid, 'title': ref, 'ts': now(), 'by': create_by}) return tid def new_msg(d, **kw): m = {'id': d['next'], 'ts': now(), 'kind': 'message', 'from': 'user', 'to': 'kills', 'topic': 'general', 'parent': None, 'text': '', 'state': 'new', 'reply': '', 'replyTs': '', 'replies': [], 'seenBy': {}, 'seenTs': ''} m.update(kw) d['next'] += 1 d['messages'].append(m) d['messages'] = d['messages'][-MAX_MESSAGES:] return m def add_user_message(d, body): text = clean(body.get('text'), MAX_TEXT) if not text: raise ValueError('empty message') to = body.get('to') or 'kills' if to not in TO_USER: raise ValueError('to must be kills, live or both') parent = body.get('parent') pm = None if parent not in (None, '', 0): try: parent = int(parent) except (TypeError, ValueError): raise ValueError('bad parent') pm = next((m for m in d['messages'] if m['id'] == parent), None) if not pm: raise ValueError('no such parent message') else: parent = None if pm: topic = pm['topic'] elif 'topic' in body and body.get('topic') not in (None, ''): topic = find_topic(d, body['topic']) if topic is None: raise ValueError('no such topic') else: topic = PAGES['id'] if PAGE_SEND.match(text) else 'general' return new_msg(d, text=text, to=to, topic=topic, parent=parent) def answer_question(d, body): try: qid = int(body.get('id')) except (TypeError, ValueError): raise ValueError('bad id') q = next((m for m in d['messages'] if m['id'] == qid and m['kind'] == 'question'), None) if not q: raise ValueError('no such question') if q['state'] == 'withdrawn': raise ValueError('that question was withdrawn') raw = body.get('answer') if q.get('multi'): ans = multi_answer(q, raw) else: if isinstance(raw, list): # a single question takes one answer (a one-item list is that answer) if len(raw) != 1: raise ValueError('that question takes one answer') raw = raw[0] ans = clean(raw, MAX_TEXT) if not ans: raise ValueError('empty answer') q['answer'], q['answeredTs'], q['state'], q['answerPulled'] = ans, now(), 'answered', False return q def multi_answer(q, raw): """A multi question's answer: a list of its choices (kept in the choices' order) plus at most one typed "other" answer (last), the same free text a single question takes. An old page sends one string: a one-item list.""" if isinstance(raw, str): raw = [raw] if not isinstance(raw, list): raise ValueError('answer must be a list') if len(raw) > MAX_CHOICES + 1: raise ValueError('too many answers') choices = q.get('choices') or [] picked, others = set(), [] for item in raw: if not isinstance(item, str): raise ValueError('each answer must be text') s = clean(item, MAX_TEXT) if not s: continue if s in choices: picked.add(s) elif s not in others: others.append(s) if len(others) > 1: raise ValueError('not one of the choices (only one typed answer allowed): %s' % others[1][:80]) ans = [c for c in choices if c in picked] + others if not ans: raise ValueError('pick at least one') return ans def remove_item(d, body, restore=False): """The page's REMOVE / RESTORE: a view flag only (nothing is deleted, no push). Removing an open question withdraws it, exactly like `unask`, so it stops counting as pending; restoring a question the user withdrew opens it again.""" try: mid = int(body.get('id')) except (TypeError, ValueError): raise ValueError('bad id') m = next((x for x in d['messages'] if x['id'] == mid), None) if not m: raise ValueError('no such message') t = now() if restore: m['hidden'], m['hiddenTs'] = False, '' if m['kind'] == 'question' and m['state'] == 'withdrawn' and m.get('withdrawnBy') == 'user' and not m.get('answeredTs'): m['state'], m['withdrawnBy'], m['withdrawnTs'] = 'open', '', '' m.pop('withdrawnPulled', None) else: if m['kind'] == 'question' and m['state'] == 'open': m['state'], m['withdrawnBy'], m['withdrawnTs'], m['withdrawnPulled'] = 'withdrawn', 'user', t, False m['hidden'], m['hiddenTs'] = True, t return {'ok': True, 'id': mid, 'kind': m['kind'], 'state': m['state'], 'hidden': m['hidden']} def user_withdrawn(d, lane): """LANE's questions the user removed while open and the coordinator was not told yet.""" return [m for m in d['messages'] if m['kind'] == 'question' and m['from'] == lane and m['state'] == 'withdrawn' and m.get('withdrawnBy') == 'user' and not m.get('withdrawnPulled')] # ---------- push (Web Push to the user's devices) ---------- _push_mod = [] def push_mod(): """inbox_push (needs `cryptography`), or None: then push is off and the inbox works exactly as before.""" if not _push_mod: try: sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import inbox_push _push_mod.append(inbox_push) except Exception: # noqa: BLE001 - any import problem = no push _push_mod.append(None) return _push_mod[0] def vapid_key(): m = push_mod() if not m or not os.path.exists(VAPID_FILE): return None try: return m.load_vapid(VAPID_FILE) except Exception: # noqa: BLE001 return None def write_private(path, data): """Write a 0600 file atomically (push.json, the VAPID key).""" tmp = path + '.tmp' fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(fd, 'wb') as f: f.write(data) os.chmod(tmp, 0o600) os.replace(tmp, path) def push_load(): try: with open(PUSH_FILE) as f: p = json.load(f) except (FileNotFoundError, json.JSONDecodeError): p = {} p.setdefault('subs', []) p.setdefault('queue', []) p.setdefault('sent', {}) return p def push_update(fn): with open(PUSH_FILE + '.lock', 'w') as lk: fcntl.flock(lk, fcntl.LOCK_EX) p = push_load() out = fn(p) write_private(PUSH_FILE, json.dumps(p, ensure_ascii=False, indent=1).encode()) return out def push_event(lane, kind, d, m, text): """Queue a push for a coordinator's reply / question / note (never raises: push must not break the CLI).""" try: title = next((t['title'] for t in d['topics'] if t['id'] == m['topic']), m['topic']) ev = {'lane': lane, 'kind': kind, 'topic': m['topic'], 'title': title, 'id': m['id'], 'text': text[:400], 'ts': time.time()} def add(p): p['queue'] = (p['queue'] + [ev])[-MAX_QUEUE:] push_update(add) except Exception: # noqa: BLE001 pass SECRETISH = re.compile(r'\b(?=[A-Za-z_+=-]*[0-9])(?=[0-9_+=-]*[A-Za-z])[A-Za-z0-9_+=-]{24,}') # key / token-like runs def push_text(s, n=120): """The notification body: the first ~n characters, one line, long key-like strings (and the token) blanked.""" s = re.sub(r'\s+', ' ', str(s or '')).strip() try: tok = token() if tok: s = s.replace(tok, '…') except OSError: pass s = SECRETISH.sub('…', s) return s if len(s) <= n else s[:n - 1].rstrip() + '…' def open_questions(): try: return sum(1 for m in load()['messages'] if m['kind'] == 'question' and m['state'] == 'open') except Exception: # noqa: BLE001 return None def push_payload(lane, evs): """One notification for a lane's events of one PUSH_GAP window: the message itself, or "N new from LIVE".""" L = lane.upper() qs = any(e['kind'] == 'question' for e in evs) topics = list(dict.fromkeys(e['topic'] for e in evs)) url = 'inbox.html?topic=' + (topics[0] if len(topics) == 1 else 'all') + ('&qs=1' if qs else '') if len(evs) == 1: e = evs[0] title = '%s%s · %s' % (L, ' asks' if qs else '', push_text(e['title'], 60)) body = push_text(e['text']) else: titles = list(dict.fromkeys(e['title'] for e in evs)) title = '%s · %d new' % (L, len(evs)) body = push_text('%d new from %s%s · %s: %s' % ( len(evs), L, ' (a question)' if qs else '', ', '.join('#' + t for t in titles[:3]) + (' …' if len(titles) > 3 else ''), evs[-1]['text'])) out = {'title': title, 'body': body, 'url': url, 'lane': lane, 'n': len(evs), 'tag': 'inbox-%s-%d' % (lane, evs[-1]['id']), 'ts': int(time.time() * 1000)} b = open_questions() if b is not None: out['badge'] = b return out def fake_post(): """INBOX_PUSH_FAKE=1 (tests only): no network; every send is appended to push_fake.jsonl in the inbox dir and answered 410 for an endpoint ending in -gone, 404 for -missing, 0 for -down, else 201.""" if not os.environ.get('INBOX_PUSH_FAKE'): return None def post(url, body, headers): with open(os.path.join(HERE, 'push_fake.jsonl'), 'a') as f: f.write(json.dumps({'url': url, 'body': body.hex(), 'headers': headers}) + '\n') return 410 if url.endswith('-gone') else 404 if url.endswith('-missing') else 0 if url.endswith('-down') else 201 return post def push_deliver(subs, payload, urgency='normal', ttl=86400): """Send payload to every sub; drop the ones that answer 404/410. Returns {sent, failed, pruned, statuses}.""" m, priv = push_mod(), vapid_key() res = {'sent': 0, 'failed': 0, 'pruned': 0, 'statuses': []} if not m or not priv: res['error'] = 'push is off on the server' return res done = {} for s in subs: try: st = m.send(priv, s, payload, ttl=ttl, urgency=urgency, _post=fake_post()) except Exception: # noqa: BLE001 - one bad subscription must not stop the others st = -1 done[s['endpoint']] = st res['statuses'].append(st) if 200 <= st < 300: res['sent'] += 1 else: res['failed'] += 1 t = now() def mark(p): keep = [] for s in p['subs']: st = done.get(s['endpoint']) if st in (404, 410): res['pruned'] += 1 continue if st is not None: if 200 <= st < 300: s['ok'], s['fails'] = t, 0 else: s['fails'] = s.get('fails', 0) + 1 s['lastFail'] = '%s %s' % (t, st) keep.append(s) p['subs'] = keep push_update(mark) return res def push_tick(t=None): """One dispatcher step: every lane with queued events whose last push is >= PUSH_GAP ago gets one notification.""" t = time.time() if t is None else t if not push_load()['queue']: return {} take = {} def pick(p): q = [e for e in p['queue'] if t - e.get('ts', t) < PUSH_STALE] # after a long outage, old news stays quiet for lane in LANES: evs = [e for e in q if e['lane'] == lane] if evs and t - p['sent'].get(lane, 0) >= PUSH_GAP: take[lane] = evs p['sent'][lane] = t p['queue'] = [e for e in q if e['lane'] not in take] return list(p['subs']) subs = push_update(pick) out = {} for lane, evs in take.items(): payload = push_payload(lane, evs) out[lane] = dict(push_deliver(subs, payload, urgency='high' if any(e['kind'] == 'question' for e in evs) else 'normal') if subs else {'sent': 0}, payload=payload) return out def push_loop(): while True: try: push_tick() except Exception as e: # noqa: BLE001 - the dispatcher never dies sys.stderr.write('push: %s\n' % type(e).__name__) time.sleep(PUSH_TICK) def push_subscribe(body, ua): m = push_mod() sub = body.get('subscription') if not isinstance(sub, dict): raise ValueError('no subscription') ep, keys = sub.get('endpoint'), sub.get('keys') or {} if not isinstance(ep, str) or not m or not m.endpoint_ok(ep): raise ValueError('not a push service endpoint') p256, auth = keys.get('p256dh'), keys.get('auth') try: ok = len(m.unb64u(p256)) == 65 and len(m.unb64u(auth)) >= 16 except Exception: # noqa: BLE001 ok = False if not ok: raise ValueError('bad subscription keys') def add(p): p['subs'] = [s for s in p['subs'] if s['endpoint'] != ep] p['subs'].append({'endpoint': ep, 'keys': {'p256dh': p256, 'auth': auth}, 'ts': now(), 'ua': clean(ua, 160)}) p['subs'] = p['subs'][-MAX_SUBS:] return len(p['subs']) return {'ok': True, 'devices': push_update(add)} def push_unsubscribe(body): ep = body.get('endpoint') if not isinstance(ep, str) or not ep: raise ValueError('no endpoint') def rm(p): n = len(p['subs']) p['subs'] = [s for s in p['subs'] if s['endpoint'] != ep] return n - len(p['subs']) return {'ok': True, 'removed': push_update(rm)} def push_test(body): ep = body.get('endpoint') subs = push_load()['subs'] if ep: subs = [s for s in subs if s['endpoint'] == ep] if not subs: raise ValueError('this device is not subscribed') if not subs: raise ValueError('no device is subscribed') payload = {'title': 'INBOX · test', 'body': 'Notifications work. A tap opens the inbox.', 'url': 'inbox.html', 'lane': 'test', 'n': 1, 'tag': 'inbox-test', 'ts': int(time.time() * 1000)} r = push_deliver(subs, payload, urgency='high', ttl=600) return {k: r[k] for k in ('sent', 'failed', 'pruned', 'error') if k in r} class Api(BaseHTTPRequestHandler): server_version = 'StickbashInbox/2' def _send(self, code, obj): body = json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header('Content-Type', 'application/json; charset=utf-8') self.send_header('Content-Length', str(len(body))) self.send_header('Cache-Control', 'no-store') self._cors() self.end_headers() self.wfile.write(body) def _cors(self): # the token is a header, not a cookie, so any origin may call (the local file:// page sends Origin: null) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type, X-Inbox-Token') def _authed(self): given = self.headers.get('X-Inbox-Token', '') return bool(given) and hmac.compare_digest(given.encode(), token().encode()) def _route(self): p = urlparse(self.path) path = p.path.rstrip('/') if path == '/api/inbox': return '', parse_qs(p.query) if path.startswith('/api/inbox/'): return path[len('/api/inbox/'):], parse_qs(p.query) return None, {} def do_OPTIONS(self): self.send_response(204) self._cors() self.end_headers() def do_GET(self): sub, qs = self._route() if sub not in ('', 'push/vapid'): return self._send(404, {'error': 'not found'}) if not self._authed(): return self._send(401, {'error': 'wrong or missing token'}) if sub == 'push/vapid': priv = vapid_key() return self._send(200, {'enabled': bool(priv), 'publicKey': push_mod().vapid_public_b64(priv) if priv else ''}) d = load() if qs.get('rev', [''])[0] == str(d['rev']): return self._send(200, {'rev': d['rev'], 'unchanged': True}) self._send(200, {'rev': d['rev'], 'status': d['status'], 'statusTs': d['statusTs'], 'statuses': d['statuses'], 'topics': d['topics'], 'messages': d['messages']}) def do_POST(self): sub, _ = self._route() if sub not in ('', 'answer', 'topic', 'remove', 'restore', 'push/subscribe', 'push/unsubscribe', 'push/test'): return self._send(404, {'error': 'not found'}) if not self._authed(): return self._send(401, {'error': 'wrong or missing token'}) try: n = int(self.headers.get('Content-Length', '0')) if n > MAX_BODY: return self._send(413, {'error': 'too big'}) body = json.loads(self.rfile.read(n) or b'{}') if not isinstance(body, dict): raise ValueError except (ValueError, json.JSONDecodeError): return self._send(400, {'error': 'bad json'}) try: if sub == '': out = update(lambda d: add_user_message(d, body)) elif sub == 'answer': out = update(lambda d: answer_question(d, body)) elif sub in ('remove', 'restore'): out = update(lambda d: remove_item(d, body, restore=sub == 'restore')) elif sub == 'push/subscribe': out = push_subscribe(body, self.headers.get('User-Agent', '')) elif sub == 'push/unsubscribe': out = push_unsubscribe(body) elif sub == 'push/test': out = push_test(body) else: title = clean(body.get('title'), MAX_TITLE) if not title: return self._send(400, {'error': 'empty title'}) def mk(d): tid = find_topic(d, title, create_by='user') return next(t for t in d['topics'] if t['id'] == tid) out = update(mk) except ValueError as e: return self._send(400, {'error': str(e)}) self._send(200, out) def log_message(self, fmt, *args): # keep the journal quiet: one line per request, no bodies sys.stderr.write('%s %s\n' % (self.command, self.path.split('?')[0])) def opt(args, name, default=None): """--name VALUE out of args (removed); also --as as an alias of --to.""" names = (name, '--as') if name == '--to' else (name,) for nm in names: if nm in args: i = args.index(nm) if i + 1 >= len(args): sys.exit('%s needs a value' % nm) val = args[i + 1] del args[i:i + 2] return val return default def lane_of(args): lane = opt(args, '--to', os.environ.get('INBOX_AS') or 'kills') if lane not in LANES: sys.exit('--to must be kills or live') return lane def cli(): args = sys.argv[1:] cmd = args.pop(0) if args else 'list' if cmd == 'serve': port = int(args[0]) if args else 8790 threading.Thread(target=push_loop, name='push', daemon=True).start() ThreadingHTTPServer(('127.0.0.1', port), Api).serve_forever() elif cmd == 'pull': lane = lane_of(args) def take(d): t, out = now(), [] for m in d['messages']: if m['kind'] != 'message' or m['from'] != 'user' or m['to'] not in (lane, 'both'): continue if m['seenBy'].get(lane): continue if m['to'] == lane and m['state'] != 'new': # seen by an older CLI (v1) before the lanes existed continue m['seenBy'][lane] = t m['seenTs'] = m['seenTs'] or t if m['state'] == 'new': m['state'] = 'seen' o = dict(m, topicTitle=next((t['title'] for t in d['topics'] if t['id'] == m['topic']), m['topic'])) pm = next((x for x in d['messages'] if x['id'] == m['parent']), None) if m['parent'] else None if pm: # a reply in a thread: the message it answers, so the coordinator has the context o['parentText'] = pm['text'][:400] o['parentFrom'] = pm['from'] out.append(o) pending = sum(1 for m in d['messages'] if m['kind'] == 'question' and m['from'] == lane and m['state'] == 'answered' and not m.get('answerPulled')) return out, pending, len(user_withdrawn(d, lane)) out, pending, gone = update(take) print(json.dumps(out, ensure_ascii=False, indent=1)) if pending: sys.stderr.write('inbox: %d new answer(s) to your questions: run inbox.sh answers --to %s\n' % (pending, lane)) if gone: sys.stderr.write('inbox: the user removed %d of your open question(s): run inbox.sh answers --to %s\n' % (gone, lane)) elif cmd == 'reply': if not args: sys.exit('usage: reply ID [--to LANE] (text on stdin)') lane = lane_of(args) mid, text = int(args[0]), clean(sys.stdin.read(), MAX_TEXT) if not text: sys.exit('empty reply') def rep(d): for m in d['messages']: if m['id'] == mid: t = now() m['replies'].append({'by': lane, 'text': text, 'ts': t}) m['reply'], m['replyTs'] = text, t m['seenBy'].setdefault(lane, t) m['seenTs'] = m['seenTs'] or t if m['kind'] == 'message': m['state'] = 'replied' push_event(lane, 'reply', d, m, text) return True return False sys.exit(0 if update(rep) else 1) elif cmd == 'status': lane = lane_of(args) text = clean(sys.stdin.read(), 1000) def st(d): t = now() d['statuses'][lane] = {'text': text, 'ts': t} if lane == 'kills': # the old pages read the single status line d['status'], d['statusTs'] = text, t update(st) elif cmd in ('ask', 'post'): try: body = json.loads(sys.stdin.read() or '{}') except json.JSONDecodeError: sys.exit('bad json on stdin') lane = body.get('to') or os.environ.get('INBOX_AS') or 'kills' if lane not in LANES: sys.exit('to must be kills or live') text = clean(body.get('text'), MAX_TEXT) if not text: sys.exit('empty text') choices = [clean(c, MAX_CHOICE) for c in (body.get('choices') or []) if clean(c, MAX_CHOICE)][:MAX_CHOICES] multi = cmd == 'ask' and body.get('multi') is True if multi: choices = list(dict.fromkeys(choices)) # no duplicate choices (a multi answer is a set of them) if multi and len(choices) < 2: sys.exit('--multi needs at least 2 --choices') def ask(d): topic = find_topic(d, body.get('topic') or 'general', create_by=lane) if cmd == 'ask': extra = {'multi': True, 'answer': []} if multi else {} # single questions keep their old shape m = new_msg(d, kind='question', **{'from': lane}, to='user', topic=topic, text=text, state='open', choices=choices, answer='', answeredTs='', answerPulled=False) m.update(extra) else: m = new_msg(d, kind='note', **{'from': lane}, to='user', topic=topic, text=text, state='posted') push_event(lane, 'question' if cmd == 'ask' else 'note', d, m, text) return m m = update(ask) out = {'id': m['id'], 'topic': m['topic'], 'kind': m['kind']} if m.get('multi'): out['multi'] = True print(json.dumps(out)) elif cmd == 'answers': lane = lane_of(args) def ans(d): out = [] for m in d['messages']: if m['kind'] == 'question' and m['from'] == lane and m['state'] == 'answered' and not m.get('answerPulled'): m['answerPulled'] = True a = {'id': m['id'], 'topic': m['topic'], 'question': m['text'], 'choices': m.get('choices', []), 'answer': m['answer'], 'answeredTs': m['answeredTs']} if m.get('multi'): # answer is a list here: the picked choices, a typed "other" last a['multi'] = True out.append(a) gone = [] for m in user_withdrawn(d, lane): # removed on the page while open: told once, on stderr (the JSON stays answers only) m['withdrawnPulled'] = True gone.append('#%d %s' % (m['id'], re.sub(r'\s+', ' ', m['text'])[:80])) return out, gone out, gone = update(ans) print(json.dumps(out, ensure_ascii=False, indent=1)) for g in gone: sys.stderr.write('inbox: the user REMOVED (withdrew) your open question %s\n' % g) elif cmd == 'unask': qid = int(args[0]) def wd(d): for m in d['messages']: if m['id'] == qid and m['kind'] == 'question' and m['state'] == 'open': m['state'] = 'withdrawn' return True return False sys.exit(0 if update(wd) else 1) elif cmd == 'topics': print(json.dumps(load()['topics'], ensure_ascii=False, indent=1)) elif cmd == 'list': print(json.dumps(load(), ensure_ascii=False, indent=1)) elif cmd == 'push-init': m = push_mod() if not m: sys.exit('push: the cryptography package is missing (python3-cryptography)') if not os.path.exists(VAPID_FILE): write_private(VAPID_FILE, m.new_vapid_pem()) os.chmod(VAPID_FILE, 0o600) print(json.dumps({'publicKey': m.vapid_public_b64(m.load_vapid(VAPID_FILE))})) elif cmd == 'push-status': p, priv = push_load(), vapid_key() print(json.dumps({'enabled': bool(priv), 'devices': len(p['subs']), 'queued': len(p['queue']), 'gap': PUSH_GAP, 'sent': {k: time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(v)) for k, v in p['sent'].items()}, 'subs': [{'host': s['endpoint'].split('/')[2], 'ts': s.get('ts'), 'ok': s.get('ok', ''), 'fails': s.get('fails', 0), 'ua': s.get('ua', '')[:60]} for s in p['subs']]}, indent=1)) elif cmd == 'push-test': try: print(json.dumps(push_test({}))) except ValueError as e: sys.exit('push: %s' % e) else: sys.exit('usage: inbox_server.py serve [port] | pull | reply ID | status | ask | post | answers | unask ID | topics | list' ' | push-init | push-status | push-test') if __name__ == '__main__': cli()