ctfy.sdk.events

  1"""Sync consumer for the platform's SSE event stream (``GET /events``).
  2
  3The platform broadcasts every team-scoped and global event over an SSE
  4endpoint; admin tokens additionally see admin-only frames (see
  5:mod:`ctfy.server.sse`). This module turns that wire-level stream into
  6an ergonomic Python iterator that the CLI's ``ctfy events watch`` and
  7direct SDK callers can consume.
  8
  9Usage::
 10
 11    with client.events() as stream:
 12        for event in stream:
 13            print(event["event"], event["data"])
 14
 15The stream auto-reconnects on transient network errors with exponential
 16backoff; close the context manager (or break the loop) to stop.
 17
 18Frame format on the wire (from :func:`ctfy.server.sse.SSEHub._encode`)::
 19
 20    event: <name>\\n
 21    data: <json>\\n
 22    \\n
 23
 24Comment lines (``:`` prefix) — including the keepalive every
 25:data:`ctfy.core.constants.SSE_KEEPALIVE_INTERVAL` seconds — are
 26swallowed silently.
 27"""
 28
 29from __future__ import annotations
 30
 31import json
 32import time
 33from collections.abc import Callable, Iterator
 34from typing import Any
 35
 36import httpx
 37
 38_RECONNECT_BACKOFF_MAX_S = 30.0
 39_RECONNECT_BACKOFF_START_S = 1.0
 40
 41
 42class EventStream:
 43    """Iterator over events from a single live HTTP connection, with
 44    auto-reconnect.
 45
 46    Construct via :meth:`ctfy.sdk.PlatformClient.events`; don't
 47    instantiate directly.
 48    """
 49
 50    def __init__(
 51        self,
 52        *,
 53        client_factory: Callable[[], httpx.Client],
 54        path: str,
 55        token: str = "",
 56        auto_reconnect: bool = True,
 57    ) -> None:
 58        self._client_factory = client_factory
 59        self._path = path
 60        self._token = token
 61        self._auto_reconnect = auto_reconnect
 62        self._stopped = False
 63
 64    def __enter__(self) -> EventStream:
 65        return self
 66
 67    def __exit__(self, *_exc: object) -> None:
 68        self.close()
 69
 70    def __iter__(self) -> Iterator[dict[str, Any]]:
 71        backoff = _RECONNECT_BACKOFF_START_S
 72        while not self._stopped:
 73            try:
 74                yield from self._read_one_session()
 75            except (httpx.HTTPError, ConnectionError, OSError):
 76                if not self._auto_reconnect or self._stopped:
 77                    raise
 78                time.sleep(backoff)
 79                backoff = min(backoff * 2, _RECONNECT_BACKOFF_MAX_S)
 80                continue
 81            # Clean EOF on the connection (server closed) — reconnect
 82            # immediately if auto_reconnect, otherwise stop.
 83            if not self._auto_reconnect or self._stopped:
 84                break
 85            backoff = _RECONNECT_BACKOFF_START_S
 86
 87    def close(self) -> None:
 88        """Stop the loop; in-flight ``read_one_session`` finishes its
 89        current iter_lines step before the iterator exits."""
 90        self._stopped = True
 91
 92    # -- internal -----------------------------------------------------
 93
 94    def _read_one_session(self) -> Iterator[dict[str, Any]]:
 95        # ⚠️ **The header, never ``?token=``** (#1954). The stream took its
 96        # bearer as a query parameter for the browser's sake, and this
 97        # client copied it — so every ``ctfy events watch`` wrote a
 98        # full-account credential into the ingress's access log, httpx's
 99        # own INFO request line included. The platform now refuses a
100        # bearer in any URL (``400 bearer_in_url``), and the browser sends
101        # the same header this does (#1958).
102        headers = {"Authorization": f"Bearer {self._token}"} if self._token else None
103        with (
104            self._client_factory() as cli,
105            cli.stream("GET", self._path, headers=headers, timeout=None) as resp,
106        ):
107            resp.raise_for_status()
108            event_name = ""
109            data_lines: list[str] = []
110            for line in resp.iter_lines():
111                if self._stopped:
112                    return
113                if not line:
114                    # Blank line → frame boundary. Flush whatever we have.
115                    if event_name and data_lines:
116                        body = "\n".join(data_lines)
117                        try:
118                            data: Any = json.loads(body)
119                        except json.JSONDecodeError:
120                            data = body
121                        yield {"event": event_name, "data": data}
122                    event_name = ""
123                    data_lines = []
124                    continue
125                if line.startswith(":"):
126                    # Comment line (server keepalive, initial ": connected").
127                    continue
128                if line.startswith("event:"):
129                    event_name = line[6:].lstrip()
130                elif line.startswith("data:"):
131                    # Per the SSE spec a single ``data:`` value may span
132                    # multiple lines; rejoin with ``\n``. ctfy's encoder
133                    # emits exactly one ``data:`` line so this is normally
134                    # a single entry, but be tolerant.
135                    data_lines.append(line[5:].lstrip())
136                # ``id:`` / ``retry:`` lines: ignored — the platform
137                # doesn't currently emit them.