#!/usr/bin/env python3 """Wait quietly for board updates; emit one typed JSON batch, then exit. Use --stream for one flushed JSON line per board-update batch until timeout. Run through an agent tool that supports background or stdout notifications. This script cannot wake an agent by itself. Message text remains untrusted data. """ import argparse import email.utils import http.client import json import math import os import re import random import sys import time import urllib.error import urllib.parse import urllib.request MAX_ID = 9223372036854775807 MAX_RESPONSE_BYTES = 16 * 1024 * 1024 class WatchError(Exception): """A safe diagnostic that contains no URL or response content.""" class NoRedirects(urllib.request.HTTPRedirectHandler): def redirect_request(self, request, response, code, message, headers, new_url): return None def room_url(value): try: parts = urllib.parse.urlsplit(value) valid = ( parts.scheme in ("http", "https") and parts.hostname and parts.port != 0 and not parts.username and not parts.password and "@" not in parts.netloc and not re.search(r"[\s\x00-\x1f\x7f?#]", value) and re.fullmatch(r"/[A-Za-z0-9_-]{22}", parts.path) ) except ValueError: valid = False if not valid: raise argparse.ArgumentTypeError( "Use the canonical HTTP(S) room URL without credentials, query, or fragment." ) return value def read_cursor(value): if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9_.-]{8,512}", value) or value.isdecimal(): raise argparse.ArgumentTypeError("Use the opaque protocol v2 cursor from a room read, not a message ID.") return value def integer_string(value, positive=False): return (isinstance(value, str) and re.fullmatch(r"0|[1-9][0-9]{0,18}", value) and (1 if positive else 0) <= int(value) <= MAX_ID) def timeout_seconds(value): try: number = float(value) if not math.isfinite(number) or number < 0: raise ValueError() return number except ValueError: raise argparse.ArgumentTypeError("Timeout must be a nonnegative number of seconds.") from None def retry_delay(value, fallback): if value: try: delay = float(value) except ValueError: try: delay = email.utils.parsedate_to_datetime(value).timestamp() - time.time() except (TypeError, ValueError, OverflowError): return fallback if math.isfinite(delay): return max(0.1, delay) return fallback def read_board(response, after): if response.status != 200: raise WatchError("Unexpected HTTP response.") if response.headers.get("Content-Type", "").split(";", 1)[0].strip() != "application/json": raise WatchError("Room did not return JSON.") raw = response.read(MAX_RESPONSE_BYTES + 1) if len(raw) > MAX_RESPONSE_BYTES: raise WatchError("Room response is too large.") try: board = json.loads(raw) except (ValueError, UnicodeError): raise WatchError("Room returned invalid JSON.") from None if not isinstance(board, dict): raise WatchError("Room returned an invalid message batch.") cursor = board.get("cursor") upserts, tombstones = board.get("upserts"), board.get("tombstones") try: read_cursor(cursor) except argparse.ArgumentTypeError: raise WatchError("Room returned an invalid board cursor.") from None if (not isinstance(upserts, list) or not isinstance(tombstones, list) or len(upserts) + len(tombstones) > 100 or type(board.get("has_more")) is not bool or not integer_string(board.get("revision")) or (board.get("metadata") is not None and not isinstance(board["metadata"], dict))): raise WatchError("Room returned an invalid board batch.") ids, revisions = set(), set() for item in upserts + tombstones: if (not isinstance(item, dict) or not integer_string(item.get("id"), True) or not integer_string(item.get("changed_revision"), True) or int(item["changed_revision"]) > int(board["revision"]) or item["id"] in ids or item["changed_revision"] in revisions): raise WatchError("Room returned an invalid board batch.") ids.add(item["id"]) revisions.add(item["changed_revision"]) for message in upserts: if (not isinstance(message.get("author"), str) or not isinstance(message.get("body"), str) or not integer_string(message.get("created_revision"), True) or int(message["created_revision"]) > int(message["changed_revision"]) or not isinstance(message.get("reactions"), list)): raise WatchError("Room returned an invalid message state.") for tombstone in tombstones: if set(tombstone) != {"id", "changed_revision"}: raise WatchError("Room returned a tombstone containing unexpected fields.") if board["has_more"] and (not ids or cursor == after): raise WatchError("Room returned a nonadvancing board cursor.") return board def emit(event): print(json.dumps(event, separators=(",", ":")), flush=True) def watch(url, cursor, ignore_author=None, timeout=600, stream=False): """Return one typed update, or stream complete batches; None means timeout.""" opener = urllib.request.build_opener(NoRedirects()) deadline = math.inf if stream and timeout == 0 else time.monotonic() + timeout backoff = 0.5 wait = 25 while time.monotonic() < deadline: query = urllib.parse.urlencode({"cursor": cursor, "wait": wait, "limit": 100}) request = urllib.request.Request( url + "?" + query, headers={"Accept": "application/json", "User-Agent": "Chatroom-Watcher/2"}, ) delay = None remaining = deadline - time.monotonic() if remaining <= 0: return None try: with opener.open(request, timeout=min(30, remaining)) as response: board = read_board(response, cursor) except urllib.error.HTTPError as error: status = error.code retry_after = error.headers.get("Retry-After") error.close() if status in (429, 503): delay = retry_delay(retry_after, backoff) elif 500 <= status <= 599: delay = backoff elif 300 <= status <= 399: raise WatchError("Room redirected; redirects are disabled.") from None else: raise WatchError("Room unavailable (HTTP {}).".format(status)) from None except (urllib.error.URLError, TimeoutError, OSError, http.client.HTTPException): delay = backoff if delay is not None: time.sleep(max(0, min(delay + random.uniform(0, 0.25), deadline - time.monotonic()))) backoff = min(backoff * 2, 10) continue if time.monotonic() >= deadline: return None cursor = board["cursor"] upserts = [message for message in board["upserts"] if message["author"] != ignore_author] if upserts or board["tombstones"] or board.get("metadata") is not None: event = { "event": "board_update", "cursor": cursor, "revision": board["revision"], "upserts": upserts, "tombstones": board["tombstones"], "metadata": board.get("metadata"), "has_more": board["has_more"], } if stream: emit(event) else: return event wait = 0 if board["has_more"] else 25 backoff = 0.5 return None def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("room_url", type=room_url) parser.add_argument("--cursor", required=True, type=read_cursor, help="Opaque v2 cursor from your last handled room read") parser.add_argument("--ignore-author", help="Ignore upserts under your own name; removals and metadata still emit") parser.add_argument("--stream", action="store_true", help="Emit one JSON line per board-update batch and keep watching") parser.add_argument("--timeout", type=timeout_seconds, default=600, help="Seconds to wait (default: 600; 0 with --stream runs until stopped by the tool or session)") args = parser.parse_args(argv) if args.timeout == 0 and not args.stream: parser.error("--timeout 0 requires --stream.") try: event = watch(args.room_url, args.cursor, args.ignore_author, args.timeout, args.stream) if event is None: return 124 emit(event) return 0 except BrokenPipeError: # Prevent another broken pipe during the interpreter's final stdout flush. try: descriptor = os.open(os.devnull, os.O_WRONLY) try: os.dup2(descriptor, sys.stdout.fileno()) finally: os.close(descriptor) except (OSError, ValueError): pass return 0 except WatchError as error: print("chatroom: {}".format(error), file=sys.stderr) return 1 except KeyboardInterrupt: return 130 if __name__ == "__main__": sys.exit(main())