From 152c3ae891e6278ca43aaf1fc8c1770dd96b7af8 Mon Sep 17 00:00:00 2001 From: Raul Bardaji Date: Sun, 30 Aug 2026 21:07:36 +0200 Subject: [PATCH 1/2] feat(pelican): stream file events over SSE at GET /pelican/subscribe The event server notifies subscribers when an object appears in a namespace, but that stream existed only in the client library, so an Endpoint could not offer subscriptions to its own callers. The Endpoint holds one upstream STOMP subscription per event source and fans it out to every SSE listener. The event server requires a unique client-id per subscriber, so a subscription per caller would split the stream between them or leave an orphaned id registered upstream on every connection. The upstream opens on the first listener and closes on the last, so an idle Endpoint holds no connection. The wire format is implemented here rather than taken from ndp-ep, to avoid the API depending on its own client SDK and on pelicanfs. Adds websockets as a dependency. Closes #262. --- CHANGELOG.md | 15 + api/routes/pelican_routes.py | 108 +++- .../pelican_services/event_protocol.py | 220 +++++++ .../pelican_services/event_subscription.py | 465 +++++++++++++++ docs/configuration.md | 26 + example.env | 22 + requirements.txt | 1 + tests/test_pelican_events.py | 561 ++++++++++++++++++ 8 files changed, 1417 insertions(+), 1 deletion(-) create mode 100644 api/services/pelican_services/event_protocol.py create mode 100644 api/services/pelican_services/event_subscription.py create mode 100644 tests/test_pelican_events.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 11d9e64..5f473b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- **`GET /pelican/subscribe` streams Pelican file events as Server-Sent Events.** The event server notifies subscribers whenever an object appears in a Pelican namespace, but that stream existed only in the `ndp-ep` client library; the API had no way to expose it, so an Endpoint could not offer subscriptions to its own callers. The new route returns a `text/event-stream` where each notification arrives as an `event: file` message carrying the object's `name`, `url`, `size` and `mod_time`. The `url` goes straight to `/pelican/read` for the contents or `/pelican/download` for the file, which is the pipeline the route exists to enable: subscribe, then read what arrived. A `: keepalive` comment is emitted during quiet periods so a proxy does not mistake an idle stream for a dead one, and `X-Accel-Buffering: no` stops nginx holding events back until its buffer fills. +- **`GET /pelican/subscriptions` reports the upstream subscriptions the Endpoint holds**, with each one's connection state, listener count and how many events were dropped for slow listeners. + +### Changed +- The event server speaks STOMP 1.2 over a WebSocket, so `websockets` is now a dependency. + +### Notes on the design +- **One upstream subscription per event source, fanned out to every listener.** The event server requires a unique `client-id` per subscriber: two connections sharing one compete for the same events rather than both receiving them. Opening a subscription per SSE caller would therefore either split the stream between callers or leave an orphaned client id registered upstream on every connection. Instead the Endpoint subscribes once per event source under its own stable id and distributes each event to all its listeners. The upstream connection opens when the first listener arrives and closes when the last one leaves, so an idle Endpoint holds no connection at all. +- **The STOMP protocol is implemented here rather than taken from the client library.** Depending on `ndp-ep` would make the API depend on its own client SDK and pull in `pelicanfs`, which pins Python 3.11. The cost is that the wire format now lives in two repositories and has to be kept in step by hand. +- **Per-listener buffers are bounded**, unlike the client library's. A caller that stops reading must not be able to grow the Endpoint's memory without limit, so a full buffer drops its oldest event and counts it in `/pelican/subscriptions`. + +### Backwards compatibility +- Purely additive. Both routes sit behind the authorization added in #261 and are only mounted when `PELICAN_ENABLED` is set. Subscriptions need `PELICAN_EVENT_CLIENT_ID` — or an `AFFINITIES_EP_UUID` to derive it from — and `/pelican/subscribe` answers 503 with the reason when the event server is not configured, so an Endpoint that never sets it is unaffected. The other new settings (`PELICAN_EVENT_SERVER_URL`, `PELICAN_EVENT_USERNAME`, `PELICAN_EVENT_PASSWORD`, `PELICAN_EVENT_VIRTUAL_HOST`, `PELICAN_EVENT_HEARTBEAT_MS`) are optional and documented in `example.env` and `docs/configuration.md`. + ## [0.34.23] - 2026-08-30 ### Added diff --git a/api/routes/pelican_routes.py b/api/routes/pelican_routes.py index fbbed21..56f4796 100644 --- a/api/routes/pelican_routes.py +++ b/api/routes/pelican_routes.py @@ -5,7 +5,14 @@ These endpoints allow browsing and downloading from external Pelican federations. """ -from fastapi import APIRouter, Depends, HTTPException, Query, Response +from fastapi import ( + APIRouter, + Depends, + HTTPException, + Query, + Request, + Response, +) from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import Optional, Dict, Any @@ -16,11 +23,17 @@ ) from api.services.pelican_services.download_file import download_file, stream_file from api.services.pelican_services.read_file import read_object +from api.services.pelican_services.event_subscription import ( + EventSubscriptionUnavailable, + broker, + load_config, +) from api.services.pelican_services.import_metadata import import_file_as_resource from api.services.auth_services import ( get_user_for_read_operation, get_user_for_write_operation, ) +import json import logging import os @@ -339,6 +352,99 @@ async def read_file_contents( raise HTTPException(status_code=500, detail=f"Error reading object: {str(e)}") +def _sse(event: str, payload: dict) -> str: + """Format one Server-Sent Event message.""" + return f"event: {event}\ndata: {json.dumps(payload)}\n\n" + + +@router.get("/subscribe") +async def subscribe_to_events( + request: Request, + event_source: str = Query( + ..., description="Namespace to watch, e.g. osdf/vdc/public/data" + ), +): + """ + Stream Pelican file events as Server-Sent Events. + + The Endpoint holds one upstream subscription per event source and + fans it out, so every listener on a source receives every event + rather than competing for them. + + Each event arrives as an ``event: file`` message whose data carries + the object's ``name``, ``url``, ``size`` and ``mod_time``. The + ``url`` can be handed straight to ``/pelican/read`` to get the + contents, or to ``/pelican/download`` to fetch it as a file. + + A ``: keepalive`` comment is sent during quiet periods, so a proxy + does not mistake an idle stream for a dead one. + + Parameters + ---------- + request : Request + Used to notice that the caller has gone away. + event_source : str + Namespace to watch. + + Returns + ------- + StreamingResponse + A ``text/event-stream``. + + Raises + ------ + HTTPException + 503 if the event server is not configured on this Endpoint. + """ + try: + config = load_config() + except EventSubscriptionUnavailable as exc: + raise HTTPException(status_code=503, detail=str(exc)) + + async def event_stream(): + # Sent immediately, so the caller can tell the stream is open + # even while the namespace is quiet. + yield ": subscribed\n\n" + try: + async for event in broker.listen(event_source, config): + if await request.is_disconnected(): + break + if event is None: + yield ": keepalive\n\n" + continue + yield _sse("file", event) + except EventSubscriptionUnavailable as exc: + yield _sse("error", {"detail": str(exc)}) + except Exception as exc: # pragma: no cover - defensive + logger.error(f"Pelican event stream failed: {exc}") + yield _sse("error", {"detail": f"{type(exc).__name__}: {exc}"}) + + return StreamingResponse( + event_stream(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + # Without this nginx buffers the stream and holds events + # back until the buffer fills, defeating the point. + "X-Accel-Buffering": "no", + }, + ) + + +@router.get("/subscriptions") +async def list_subscriptions(): + """ + Report the upstream subscriptions this Endpoint currently holds. + + Returns + ------- + dict + One entry per event source with its connection state, listener + count, and how many events were dropped for slow listeners. + """ + return {"success": True, "subscriptions": broker.status()} + + @router.post("/import-metadata") async def import_metadata( request: ImportMetadataRequest, diff --git a/api/services/pelican_services/event_protocol.py b/api/services/pelican_services/event_protocol.py new file mode 100644 index 0000000..0e0c5dc --- /dev/null +++ b/api/services/pelican_services/event_protocol.py @@ -0,0 +1,220 @@ +# api/services/pelican_services/event_protocol.py +""" +Wire format for Pelican file-event subscriptions. + +The event server speaks STOMP 1.2 over a WebSocket. This module holds +the parts that need no network — frame encoding, frame parsing and the +file event carried in a MESSAGE body — so they stay importable and +testable without a connection. + +STOMP itself does not define the body schema. The event server sends a +JSON object with ``name``, ``url``, ``size`` and ``mod_time``, sometimes +wrapped in a JSON string; both spellings are accepted. + +The same protocol is implemented in the ``ndp-ep`` client library. It is +reimplemented here rather than depended on, so the API does not take a +dependency on its own client SDK; the two have to be kept in step by +hand. +""" + +import json +import logging +from dataclasses import dataclass, field +from typing import Any, Dict, Optional +from urllib.parse import urlparse, urlunparse + +logger = logging.getLogger(__name__) + +# STOMP 1.2 header escaping. Order matters on the way out: the backslash +# substitution has to happen first or it would re-escape the others. +_ESCAPES = ( + ("\\", "\\\\"), + ("\r", "\\r"), + ("\n", "\\n"), + (":", "\\c"), +) +_UNESCAPES = {"\\": "\\", "r": "\r", "n": "\n", "c": ":"} + +_WS_SCHEMES = {"http": "ws", "https": "wss", "ws": "ws", "wss": "wss"} + + +def escape_header(value: str) -> str: + """Escape a header name or value for transmission.""" + for plain, escaped in _ESCAPES: + value = value.replace(plain, escaped) + return value + + +def unescape_header(value: str) -> str: + """ + Reverse :func:`escape_header`. + + Scans left to right rather than substituting repeatedly, so an + escaped backslash followed by a letter yields a backslash and that + letter, instead of being read a second time as a newline. + """ + out = [] + index = 0 + while index < len(value): + char = value[index] + if char == "\\" and index + 1 < len(value): + out.append(_UNESCAPES.get(value[index + 1], value[index + 1])) + index += 2 + else: + out.append(char) + index += 1 + return "".join(out) + + +@dataclass +class Frame: + """A STOMP frame.""" + + command: str + headers: Dict[str, str] = field(default_factory=dict) + body: str = "" + + def encode(self) -> bytes: + """Serialise the frame, terminated by the NUL octet.""" + lines = [self.command] + lines.extend( + f"{escape_header(key)}:{escape_header(value)}" + for key, value in self.headers.items() + ) + return ("\n".join(lines) + "\n\n" + self.body + "\x00").encode() + + +def parse_frame(raw: bytes) -> Frame: + """ + Parse a STOMP frame. + + Repeated headers keep the first value, as STOMP 1.2 requires. + """ + text = raw.rstrip(b"\x00").decode(errors="replace") + head, _, body = text.partition("\n\n") + lines = head.split("\n") + + headers: Dict[str, str] = {} + for line in lines[1:]: + if ":" in line: + key, _, value = line.partition(":") + headers.setdefault(unescape_header(key), unescape_header(value)) + + return Frame(lines[0], headers, body) + + +def is_heartbeat(raw: bytes) -> bool: + """Report whether a received payload is a heartbeat, not a frame.""" + return raw in (b"\n", b"\r\n") + + +def websocket_url(value: str) -> str: + """ + Convert an http(s) or ws(s) URL to its WebSocket form. + + Raises + ------ + ValueError + If the URL has no host or an unusable scheme. + """ + parsed = urlparse(value) + if parsed.scheme not in _WS_SCHEMES or not parsed.netloc: + raise ValueError( + f"Event server URL must be an http(s) or ws(s) URL, got '{value}'." + ) + return urlunparse(parsed._replace(scheme=_WS_SCHEMES[parsed.scheme])) + + +def parse_file_event( + body: str, event_id: str = "", destination: str = "" +) -> Dict[str, Any]: + """ + Build a file event from a MESSAGE body. + + Parameters + ---------- + body : str + The raw MESSAGE body. + event_id : str + Identity to attach, if already computed. + destination : str + Destination the event arrived on. + + Returns + ------- + dict + ``name``, ``url``, ``size``, ``mod_time``, ``event_id`` and + ``destination``, ready to be serialised into an SSE payload. + + Raises + ------ + ValueError + If the body is not JSON, or a required field is missing or of + the wrong type. + """ + payload = decode_body(body) + + name = payload.get("name") + url = payload.get("url") + size = payload.get("size") + mod_time = payload.get("mod_time") + + if not isinstance(name, str) or not name: + raise ValueError("event 'name' must be a non-empty string") + if not isinstance(url, str) or not url: + raise ValueError("event 'url' must be a non-empty string") + # bool is a subclass of int, so it has to be excluded explicitly. + if isinstance(size, bool) or not isinstance(size, int) or size < 0: + raise ValueError("event 'size' must be a non-negative integer") + if not isinstance(mod_time, str) or not mod_time: + raise ValueError("event 'mod_time' must be a non-empty string") + + return { + "name": name, + "url": url, + "size": size, + "mod_time": mod_time, + "event_id": event_id, + "destination": destination, + } + + +def event_identity(frame: Frame) -> str: + """ + Work out what makes this event the same event on redelivery. + + The publisher stamps each event with a ``uuid``, which survives a + redelivery. The STOMP ``message-id`` does not always, so it is only + a fallback. + """ + try: + payload = decode_body(frame.body) + except ValueError: + payload = {} + + identifier = payload.get("uuid") + if identifier: + return str(identifier) + return frame.headers.get("message-id", "unknown-message") + + +def decode_body(body: str) -> Dict[str, Any]: + """Decode a MESSAGE body, unwrapping a JSON-string envelope.""" + try: + payload: Optional[Any] = json.loads(body) + except (TypeError, json.JSONDecodeError) as exc: + raise ValueError("event body is not valid JSON") from exc + + # The publisher currently serialises the event as a JSON string + # inside the STOMP body; accept that as well as a plain object. + if isinstance(payload, str): + try: + payload = json.loads(payload) + except json.JSONDecodeError as exc: + raise ValueError( + "event body is a JSON string that does not contain JSON" + ) from exc + + if not isinstance(payload, dict): + raise ValueError("event body must be a JSON object") + return payload diff --git a/api/services/pelican_services/event_subscription.py b/api/services/pelican_services/event_subscription.py new file mode 100644 index 0000000..1390bf1 --- /dev/null +++ b/api/services/pelican_services/event_subscription.py @@ -0,0 +1,465 @@ +# api/services/pelican_services/event_subscription.py +""" +Fan-out of Pelican file events to Server-Sent Events listeners. + +The event server hands every subscription a ``client-id`` that must be +unique: two connections sharing one compete for the same events instead +of both receiving them. A subscription per SSE caller would therefore +either need a fresh identity each time — leaving orphaned client ids +registered upstream — or silently split the stream between callers. + +So the Endpoint holds **one** upstream subscription per event source, +under its own stable client id, and fans each event out to every SSE +listener attached to that source. The upstream connection is opened when +the first listener arrives and closed when the last one leaves. +""" + +import asyncio +import base64 +import inspect +import logging +import os +from collections import OrderedDict +from dataclasses import dataclass +from typing import AsyncIterator, Dict, Optional, Set + +from api.services.pelican_services.event_protocol import ( + Frame, + event_identity, + is_heartbeat, + parse_file_event, + parse_frame, + websocket_url, +) + +try: # pragma: no cover - absence is covered by load_config's guard + import websockets +except ImportError: # pragma: no cover + websockets = None + +logger = logging.getLogger(__name__) + +DEFAULT_EVENT_SERVER = "https://stomp-server.chtcdev.chtc.io/ws" +DEFAULT_VIRTUAL_HOST = "playground" +DEFAULT_HEARTBEAT_MS = 10000 +MAX_BACKOFF_SECONDS = 30 + +# Per-listener buffer. Bounded, unlike the client library's: a browser +# tab that stops reading must not be able to grow the Endpoint's memory +# without limit. When it fills, the oldest event is dropped and counted. +LISTENER_BUFFER = 256 + +# How many event identities to remember for redelivery suppression. The +# server may redeliver after a reconnect; without this every listener +# would see the same event twice. +SEEN_EVENTS = 2048 + +# Silence after which a listener is handed a keepalive. Without one, an +# idle stream looks dead to proxies and they close it. +IDLE_TIMEOUT_SECONDS = 15.0 + + +class EventSubscriptionUnavailable(RuntimeError): + """The subscription cannot be served with the current configuration.""" + + +@dataclass(frozen=True) +class EventServerConfig: + """Connection settings for the event server.""" + + url: str + client_id: str + username: str + password: str + virtual_host: str + heartbeat_ms: int + + +def load_config() -> EventServerConfig: + """ + Read the event server settings from the environment. + + Returns + ------- + EventServerConfig + Validated settings. + + Raises + ------ + EventSubscriptionUnavailable + If a required setting is missing or unusable. Settings are + declared with ``extra: "allow"``, so a typo would otherwise + surface as a failed connection instead of a clear error. + """ + if websockets is None: + raise EventSubscriptionUnavailable( + "The 'websockets' package is required for Pelican event " + "subscriptions and is not installed." + ) + + raw_url = os.getenv("PELICAN_EVENT_SERVER_URL") or DEFAULT_EVENT_SERVER + try: + url = websocket_url(raw_url) + except ValueError as exc: + raise EventSubscriptionUnavailable(str(exc)) from exc + + # A stable identity per Endpoint. Falling back to the Endpoint UUID + # keeps it unique across deployments without another thing to set. + client_id = ( + os.getenv("PELICAN_EVENT_CLIENT_ID") or os.getenv("AFFINITIES_EP_UUID") or "" + ).strip() + if not client_id: + raise EventSubscriptionUnavailable( + "PELICAN_EVENT_CLIENT_ID is not set and no Endpoint UUID is " + "available to derive it from. The event server requires a " + "unique client id per subscriber." + ) + if "/" in client_id: + raise EventSubscriptionUnavailable( + "PELICAN_EVENT_CLIENT_ID must not contain '/': it is one " + "segment of the STOMP destination." + ) + + username = os.getenv("PELICAN_EVENT_USERNAME", "") + password = os.getenv("PELICAN_EVENT_PASSWORD", "") + if bool(username) != bool(password): + raise EventSubscriptionUnavailable( + "PELICAN_EVENT_USERNAME and PELICAN_EVENT_PASSWORD must be " "set together." + ) + + return EventServerConfig( + url=url, + client_id=client_id, + username=username, + password=password, + virtual_host=os.getenv("PELICAN_EVENT_VIRTUAL_HOST", DEFAULT_VIRTUAL_HOST), + heartbeat_ms=_heartbeat_ms(), + ) + + +def _heartbeat_ms() -> int: + """Resolve the liveness interval, falling back on an unusable value.""" + raw = os.getenv("PELICAN_EVENT_HEARTBEAT_MS", "") + if not raw: + return DEFAULT_HEARTBEAT_MS + try: + value = int(raw) + except ValueError: + logger.warning( + f"PELICAN_EVENT_HEARTBEAT_MS is not an integer ({raw!r}); " + f"using {DEFAULT_HEARTBEAT_MS}." + ) + return DEFAULT_HEARTBEAT_MS + if value < 1000: + logger.warning( + f"PELICAN_EVENT_HEARTBEAT_MS must be at least 1000 (got " + f"{value}); using {DEFAULT_HEARTBEAT_MS}." + ) + return DEFAULT_HEARTBEAT_MS + return value + + +def _basic_auth(username: str, password: str) -> str: + """Build the Basic credential for the WebSocket handshake.""" + raw = f"{username}:{password}".encode() + return f"Basic {base64.b64encode(raw).decode()}" + + +def _header_kwargs(headers: Optional[Dict[str, str]]) -> Dict[str, object]: + """Name the handshake-header argument the way this websockets wants it.""" + if not headers: + return {} + parameters = inspect.signature(websockets.connect).parameters + if "additional_headers" in parameters: + return {"additional_headers": headers} + return {"extra_headers": headers} + + +class _Upstream: + """One STOMP subscription, shared by every listener on an event source.""" + + def __init__(self, event_source: str, config: EventServerConfig): + self.event_source = event_source.strip().strip("/") + self.config = config + self.listeners: Set[asyncio.Queue] = set() + self.state = "starting" + self.last_error = "" + self.dropped = 0 + self._seen: "OrderedDict[str, None]" = OrderedDict() + self._task: Optional[asyncio.Task] = None + self._stop = asyncio.Event() + + @property + def destination(self) -> str: + """The STOMP destination this subscription listens on.""" + return f"{self.config.client_id}/{self.event_source}" + + def start(self) -> None: + """Begin connecting, without blocking the caller.""" + if self._task is None: + self._task = asyncio.ensure_future(self._run()) + + async def stop(self) -> None: + """Close the upstream connection and wait for the task to end.""" + self._stop.set() + task = self._task + self._task = None + if task is not None: + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception as exc: # pragma: no cover - defensive + logger.debug(f"Upstream task ended with {exc}") + + def add_listener(self) -> asyncio.Queue: + """Attach a listener and hand back the queue it should read.""" + queue: asyncio.Queue = asyncio.Queue(maxsize=LISTENER_BUFFER) + self.listeners.add(queue) + return queue + + def remove_listener(self, queue: asyncio.Queue) -> None: + """Detach a listener that has gone away.""" + self.listeners.discard(queue) + + def _already_seen(self, identity: str) -> bool: + """Suppress a redelivery, remembering a bounded number of ids.""" + if identity in self._seen: + return True + self._seen[identity] = None + while len(self._seen) > SEEN_EVENTS: + self._seen.popitem(last=False) + return False + + def _publish(self, payload: dict) -> None: + """Hand an event to every listener, dropping the oldest if full.""" + for queue in list(self.listeners): + if queue.full(): + try: + queue.get_nowait() + self.dropped += 1 + except asyncio.QueueEmpty: # pragma: no cover - race only + pass + try: + queue.put_nowait(payload) + except asyncio.QueueFull: # pragma: no cover - race only + self.dropped += 1 + + async def _run(self) -> None: + """Keep the upstream connected, backing off after a failure.""" + delay = 1.0 + while not self._stop.is_set(): + try: + await self._session() + delay = 1.0 + except asyncio.CancelledError: + raise + except Exception as exc: + self.state = "disconnected" + self.last_error = f"{type(exc).__name__}: {exc}" + logger.warning( + f"Pelican event subscription to {self.destination} " + f"failed ({self.last_error}); retrying in {delay:.0f}s." + ) + try: + await asyncio.wait_for(self._stop.wait(), timeout=delay) + except asyncio.TimeoutError: + pass + delay = min(delay * 2, MAX_BACKOFF_SECONDS) + + async def _session(self) -> None: + """Run one connection, from handshake until it drops.""" + headers = None + if self.config.username: + headers = { + "Authorization": _basic_auth(self.config.username, self.config.password) + } + + self.state = "connecting" + # Our own heartbeat below is the liveness signal, so the + # websockets keepalive would only duplicate it. + connect = websockets.connect( + self.config.url, ping_interval=None, **_header_kwargs(headers) + ) + async with connect as connection: + await connection.send(self._connect_frame().encode()) + + greeting = await self._receive(connection) + if greeting.command != "CONNECTED": + raise RuntimeError( + f"Expected CONNECTED from the event server, got " + f"{greeting.command or 'nothing'}." + ) + + await connection.send(self._subscribe_frame().encode()) + self.state = "connected" + self.last_error = "" + logger.info(f"Subscribed to {self.destination} as {self.config.client_id}") + + heartbeat = asyncio.ensure_future(self._heartbeat(connection)) + try: + while not self._stop.is_set(): + frame = await self._receive(connection) + if frame.command == "MESSAGE": + await self._on_message(connection, frame) + elif frame.command == "ERROR": + message = frame.body or frame.headers.get( + "message", "server error" + ) + self.last_error = f"Server error: {message}" + logger.error(f"Event server reported an error: {message}") + finally: + heartbeat.cancel() + self.state = "disconnected" + + def _connect_frame(self) -> Frame: + return Frame( + "CONNECT", + { + "accept-version": "1.2", + "host": self.config.virtual_host, + "client-id": self.config.client_id, + "heart-beat": ( + f"{self.config.heartbeat_ms},{self.config.heartbeat_ms}" + ), + }, + ) + + def _subscribe_frame(self) -> Frame: + return Frame( + "SUBSCRIBE", + { + "id": self.config.client_id, + "subscription": self.config.client_id, + "destination": self.destination, + "ack": "client-individual", + }, + ) + + async def _heartbeat(self, connection) -> None: + """ + Send the heartbeats promised in CONNECT. + + At half the negotiated interval, so an ordinary scheduling delay + is not read by the server as a dead client. + """ + interval = self.config.heartbeat_ms / 2000 + while True: + await asyncio.sleep(interval) + await connection.send(b"\n") + + async def _receive(self, connection) -> Frame: + """Read the next frame, skipping heartbeats.""" + while True: + data = await connection.recv() + if isinstance(data, str): + data = data.encode() + if is_heartbeat(data): + continue + return parse_frame(data) + + async def _on_message(self, connection, frame: Frame) -> None: + """Publish a MESSAGE to the listeners and acknowledge it.""" + identity = event_identity(frame) + destination = frame.headers.get("destination", "") + + if not self._already_seen(identity): + try: + payload = parse_file_event(frame.body, identity, destination) + except ValueError as exc: + logger.warning(f"Ignoring unreadable event: {exc}") + else: + self._publish(payload) + + # Acknowledged even when unreadable or a duplicate: leaving it + # unacknowledged would have the server redeliver it forever. + ack_id = frame.headers.get("ack") + if ack_id: + await connection.send(Frame("ACK", {"id": ack_id}).encode()) + + +class PelicanEventBroker: + """Holds one :class:`_Upstream` per event source, reference counted.""" + + def __init__(self) -> None: + self._upstreams: Dict[str, _Upstream] = {} + self._lock = asyncio.Lock() + + def status(self) -> Dict[str, dict]: + """Report what is currently subscribed, for diagnostics.""" + return { + source: { + "state": upstream.state, + "destination": upstream.destination, + "listeners": len(upstream.listeners), + "dropped_events": upstream.dropped, + "last_error": upstream.last_error, + } + for source, upstream in self._upstreams.items() + } + + async def listen( + self, + event_source: str, + config: EventServerConfig, + idle_timeout: float = IDLE_TIMEOUT_SECONDS, + ) -> AsyncIterator[Optional[dict]]: + """ + Yield events for ``event_source`` until the caller goes away. + + Opens the upstream subscription on the first listener and closes + it when the last one leaves, so an idle Endpoint holds no + connection to the event server. + + Yields ``None`` once ``idle_timeout`` passes with no event, so + the caller can emit a keepalive. The wait is timed out *here* + rather than around the iterator, because cancelling an + ``__anext__`` would unwind this generator and tear the + subscription down on every quiet interval. + + Parameters + ---------- + event_source : str + Namespace to watch, e.g. ``osdf/vdc/public/data``. + config : EventServerConfig + Settings for the upstream connection. + idle_timeout : float + Seconds of silence before yielding ``None``. + + Raises + ------ + EventSubscriptionUnavailable + If ``event_source`` is empty. + """ + source = event_source.strip().strip("/") + if not source: + raise EventSubscriptionUnavailable( + "event_source must be a non-empty namespace path." + ) + + async with self._lock: + upstream = self._upstreams.get(source) + if upstream is None: + upstream = _Upstream(source, config) + self._upstreams[source] = upstream + upstream.start() + queue = upstream.add_listener() + + try: + while True: + try: + yield await asyncio.wait_for(queue.get(), timeout=idle_timeout) + except asyncio.TimeoutError: + yield None + finally: + async with self._lock: + upstream.remove_listener(queue) + if not upstream.listeners: + self._upstreams.pop(source, None) + await upstream.stop() + + +#: Process-wide broker. One per worker, which is what keeps the client +#: id unique: two workers would otherwise share it and split the stream. +broker = PelicanEventBroker() diff --git a/docs/configuration.md b/docs/configuration.md index 5246b36..afb656a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -349,6 +349,32 @@ the API's memory; a larger object is refused with 413 and must be fetched with default. **Where:** raise it only if your callers genuinely read larger objects inline. +#### `PELICAN_EVENT_SERVER_URL` +*Optional · default: the CHTC event server.* +Event server backing `GET /pelican/subscribe`, as an `http(s)` or `ws(s)` URL. + +#### `PELICAN_EVENT_CLIENT_ID` +*Optional · default: `AFFINITIES_EP_UUID`.* +Identifies this Endpoint to the event server. **Must be unique**: two +subscribers sharing an id compete for the same events instead of both receiving +them. Must not contain `/`, since it is one segment of the STOMP destination. +`GET /pelican/subscribe` returns 503 when neither this nor the Endpoint UUID is +set. **Where:** leave empty unless one host runs several Endpoints. + +#### `PELICAN_EVENT_USERNAME` / `PELICAN_EVENT_PASSWORD` +*Optional · default: empty. Set both or neither.* +Credentials the event server checks against its own store — unrelated to the +Endpoint token. Temporary: they are due to be replaced by an access token +issued for NDP. + +#### `PELICAN_EVENT_VIRTUAL_HOST` +*Optional · default: `playground`.* +STOMP virtual host sent in the CONNECT frame. + +#### `PELICAN_EVENT_HEARTBEAT_MS` +*Optional · default: `10000`. Minimum 1000.* +Liveness interval. A non-numeric or too-small value falls back to the default. + --- ## Remote execution (Rexec) diff --git a/example.env b/example.env index f4227dd..b649425 100644 --- a/example.env +++ b/example.env @@ -325,6 +325,28 @@ PELICAN_DIRECT_READS=False # Default: 10485760 (10 MiB) PELICAN_MAX_READ_BYTES=10485760 +# Pelican file-event subscriptions (GET /pelican/subscribe) +# The event server notifies subscribers when an object appears in a namespace +# Leave the URL empty to use the default event server +PELICAN_EVENT_SERVER_URL= + +# Identifies this Endpoint to the event server. MUST be unique: two subscribers +# sharing an id compete for the same events instead of both receiving them +# Must not contain "/". Falls back to AFFINITIES_EP_UUID when left empty +PELICAN_EVENT_CLIENT_ID= + +# Event server credentials, checked against its own store +# Unrelated to the Endpoint token; set both or neither +# Temporary: they will be replaced by an access token issued for NDP +PELICAN_EVENT_USERNAME= +PELICAN_EVENT_PASSWORD= + +# STOMP virtual host on the event server +PELICAN_EVENT_VIRTUAL_HOST=playground + +# Liveness interval in milliseconds, minimum 1000 +PELICAN_EVENT_HEARTBEAT_MS=10000 + # ============================================== # Rexec Deployment API Configuration # ============================================== diff --git a/requirements.txt b/requirements.txt index 3eecb50..1e468fc 100644 --- a/requirements.txt +++ b/requirements.txt @@ -33,3 +33,4 @@ opentelemetry-instrumentation-fastapi>=0.41b0 opentelemetry-instrumentation-httpx>=0.41b0 opentelemetry-instrumentation-requests>=0.41b0 opentelemetry-exporter-otlp>=1.20.0 +websockets>=11.0 diff --git a/tests/test_pelican_events.py b/tests/test_pelican_events.py new file mode 100644 index 0000000..164ca9a --- /dev/null +++ b/tests/test_pelican_events.py @@ -0,0 +1,561 @@ +"""Tests for Pelican file-event subscriptions (issue #262).""" + +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from api.services.pelican_services.event_protocol import ( + Frame, + escape_header, + event_identity, + is_heartbeat, + parse_file_event, + parse_frame, + unescape_header, + websocket_url, +) +from api.services.pelican_services.event_subscription import ( + DEFAULT_HEARTBEAT_MS, + EventServerConfig, + EventSubscriptionUnavailable, + LISTENER_BUFFER, + PelicanEventBroker, + _Upstream, + load_config, +) + +CONFIG = EventServerConfig( + url="wss://events.example/ws", + client_id="ep-test", + username="user", + password="secret", + virtual_host="playground", + heartbeat_ms=10000, +) + + +def _event_body(**overrides): + payload = { + "name": "data.csv", + "url": "pelican://osg-htc.org/public/data.csv", + "size": 42, + "mod_time": "2026-08-30T12:00:00Z", + } + payload.update(overrides) + return json.dumps(payload) + + +class TestHeaderEscaping: + """STOMP 1.2 header escaping.""" + + def test_round_trip_covers_every_escape(self): + """ + Every escaped character must survive a round trip. The four + substitutions are easy to corrupt one at a time and the damage + is invisible until the server rejects a frame. + """ + raw = "a:b\nc\rd\\e" + + assert unescape_header(escape_header(raw)) == raw + + def test_escapes_use_the_spelling_the_protocol_requires(self): + assert escape_header("a:b") == "a\\cb" + assert escape_header("a\nb") == "a\\nb" + assert escape_header("a\rb") == "a\\rb" + assert escape_header("a\\b") == "a\\\\b" + + def test_backslash_is_escaped_first(self): + """Otherwise the backslash pass would re-escape the others.""" + assert escape_header("\\n") == "\\\\n" + assert unescape_header("\\\\n") == "\\n" + + +class TestFrames: + """Frame encoding and parsing.""" + + def test_encode_terminates_with_nul_after_a_blank_line(self): + frame = Frame("CONNECT", {"accept-version": "1.2"}, "") + + assert frame.encode() == b"CONNECT\naccept-version:1.2\n\n\x00" + + def test_encode_parse_round_trip(self): + frame = Frame("SEND", {"destination": "ep/osdf/pub"}, "body") + + parsed = parse_frame(frame.encode()) + + assert parsed.command == "SEND" + assert parsed.headers == {"destination": "ep/osdf/pub"} + assert parsed.body == "body" + + def test_repeated_headers_keep_the_first_value(self): + raw = b"MESSAGE\nid:first\nid:second\n\nbody\x00" + + assert parse_frame(raw).headers["id"] == "first" + + def test_parse_unescapes_headers(self): + raw = b"MESSAGE\nkey:a\\cb\n\n\x00" + + assert parse_frame(raw).headers["key"] == "a:b" + + def test_heartbeat_recognition(self): + assert is_heartbeat(b"\n") is True + assert is_heartbeat(b"\r\n") is True + assert is_heartbeat(b"MESSAGE\n\n\x00") is False + + +class TestWebsocketUrl: + """URL normalisation.""" + + @pytest.mark.parametrize( + "given,expected", + [ + ("https://a.example/ws", "wss://a.example/ws"), + ("http://a.example/ws", "ws://a.example/ws"), + ("wss://a.example/ws", "wss://a.example/ws"), + ("ws://a.example/ws", "ws://a.example/ws"), + ], + ) + def test_scheme_is_mapped(self, given, expected): + assert websocket_url(given) == expected + + @pytest.mark.parametrize("given", ["ftp://a.example", "not-a-url", ""]) + def test_unusable_url_is_rejected(self, given): + with pytest.raises(ValueError): + websocket_url(given) + + +class TestParseFileEvent: + """The application-level event inside a MESSAGE body.""" + + def test_valid_event(self): + event = parse_file_event(_event_body(), "id-1", "ep/osdf/pub") + + assert event["name"] == "data.csv" + assert event["size"] == 42 + assert event["event_id"] == "id-1" + assert event["destination"] == "ep/osdf/pub" + + def test_json_string_envelope_is_unwrapped(self): + """The publisher currently wraps the object in a JSON string.""" + event = parse_file_event(json.dumps(_event_body())) + + assert event["name"] == "data.csv" + + @pytest.mark.parametrize( + "overrides", + [ + {"name": ""}, + {"name": 5}, + {"url": ""}, + {"size": -1}, + {"size": "42"}, + {"size": True}, + {"mod_time": ""}, + ], + ) + def test_malformed_event_is_rejected(self, overrides): + with pytest.raises(ValueError): + parse_file_event(_event_body(**overrides)) + + def test_non_json_body_is_rejected(self): + with pytest.raises(ValueError): + parse_file_event("not json") + + def test_json_that_is_not_an_object_is_rejected(self): + with pytest.raises(ValueError): + parse_file_event("[1, 2, 3]") + + +class TestEventIdentity: + """Redelivery identity.""" + + def test_publisher_uuid_wins(self): + frame = Frame( + "MESSAGE", + {"message-id": "m-1"}, + _event_body(uuid="u-1"), + ) + + assert event_identity(frame) == "u-1" + + def test_message_id_is_the_fallback(self): + frame = Frame("MESSAGE", {"message-id": "m-1"}, _event_body()) + + assert event_identity(frame) == "m-1" + + def test_unreadable_body_still_yields_an_identity(self): + frame = Frame("MESSAGE", {}, "not json") + + assert event_identity(frame) == "unknown-message" + + +class TestLoadConfig: + """Environment resolution.""" + + @staticmethod + def _env(**overrides): + env = { + "PELICAN_EVENT_CLIENT_ID": "ep-test", + "PELICAN_EVENT_USERNAME": "user", + "PELICAN_EVENT_PASSWORD": "secret", + } + env.update(overrides) + return env + + def test_defaults(self): + with patch.dict("os.environ", self._env(), clear=True): + config = load_config() + + assert config.url.startswith("wss://") + assert config.client_id == "ep-test" + assert config.virtual_host == "playground" + assert config.heartbeat_ms == DEFAULT_HEARTBEAT_MS + + def test_endpoint_uuid_is_the_fallback_client_id(self): + """One less thing to configure, and unique per deployment.""" + env = {"AFFINITIES_EP_UUID": "uuid-1"} + with patch.dict("os.environ", env, clear=True): + assert load_config().client_id == "uuid-1" + + def test_missing_client_id_is_refused(self): + with patch.dict("os.environ", {}, clear=True): + with pytest.raises(EventSubscriptionUnavailable) as exc: + load_config() + + assert "unique client id" in str(exc.value) + + def test_client_id_with_a_slash_is_refused(self): + """It is one segment of the destination, so a slash reroutes it.""" + env = self._env(PELICAN_EVENT_CLIENT_ID="a/b") + with patch.dict("os.environ", env, clear=True): + with pytest.raises(EventSubscriptionUnavailable): + load_config() + + def test_half_a_credential_is_refused(self): + env = self._env(PELICAN_EVENT_PASSWORD="") + with patch.dict("os.environ", env, clear=True): + with pytest.raises(EventSubscriptionUnavailable): + load_config() + + def test_unusable_server_url_is_refused(self): + env = self._env(PELICAN_EVENT_SERVER_URL="ftp://nope") + with patch.dict("os.environ", env, clear=True): + with pytest.raises(EventSubscriptionUnavailable): + load_config() + + @pytest.mark.parametrize("value", ["not-a-number", "0", "500"]) + def test_unusable_heartbeat_falls_back(self, value): + """Settings allow extra keys, so a typo has to be caught here.""" + env = self._env(PELICAN_EVENT_HEARTBEAT_MS=value) + with patch.dict("os.environ", env, clear=True): + assert load_config().heartbeat_ms == DEFAULT_HEARTBEAT_MS + + def test_explicit_heartbeat_is_used(self): + env = self._env(PELICAN_EVENT_HEARTBEAT_MS="30000") + with patch.dict("os.environ", env, clear=True): + assert load_config().heartbeat_ms == 30000 + + +class TestUpstreamFrames: + """The frames the Endpoint sends to the event server.""" + + def test_destination_is_client_id_over_event_source(self): + upstream = _Upstream("/osdf/vdc/public/", CONFIG) + + assert upstream.destination == "ep-test/osdf/vdc/public" + + def test_connect_frame(self): + headers = _Upstream("osdf/pub", CONFIG)._connect_frame().headers + + assert headers["accept-version"] == "1.2" + assert headers["host"] == "playground" + assert headers["client-id"] == "ep-test" + assert headers["heart-beat"] == "10000,10000" + + def test_subscribe_frame_acknowledges_individually(self): + headers = _Upstream("osdf/pub", CONFIG)._subscribe_frame().headers + + assert headers["destination"] == "ep-test/osdf/pub" + assert headers["ack"] == "client-individual" + + +class TestUpstreamFanOut: + """Delivery to the listeners attached to one upstream.""" + + @pytest.mark.asyncio + async def test_every_listener_receives_every_event(self): + """ + The whole point of sharing one upstream: listeners must not + compete for events the way two STOMP clients on one id would. + """ + upstream = _Upstream("osdf/pub", CONFIG) + first = upstream.add_listener() + second = upstream.add_listener() + + upstream._publish({"name": "a.csv"}) + + assert first.get_nowait() == {"name": "a.csv"} + assert second.get_nowait() == {"name": "a.csv"} + + @pytest.mark.asyncio + async def test_slow_listener_drops_its_oldest_event(self): + """A listener that stops reading must not grow without limit.""" + upstream = _Upstream("osdf/pub", CONFIG) + queue = upstream.add_listener() + for index in range(LISTENER_BUFFER): + upstream._publish({"n": index}) + + upstream._publish({"n": "newest"}) + + assert queue.qsize() == LISTENER_BUFFER + assert upstream.dropped == 1 + assert queue.get_nowait() == {"n": 1} + + @pytest.mark.asyncio + async def test_removed_listener_stops_receiving(self): + upstream = _Upstream("osdf/pub", CONFIG) + queue = upstream.add_listener() + upstream.remove_listener(queue) + + upstream._publish({"name": "a.csv"}) + + assert queue.empty() + + def test_redelivery_is_suppressed(self): + upstream = _Upstream("osdf/pub", CONFIG) + + assert upstream._already_seen("u-1") is False + assert upstream._already_seen("u-1") is True + + def test_seen_identities_are_bounded(self): + from api.services.pelican_services.event_subscription import ( + SEEN_EVENTS, + ) + + upstream = _Upstream("osdf/pub", CONFIG) + for index in range(SEEN_EVENTS + 10): + upstream._already_seen(f"u-{index}") + + assert len(upstream._seen) <= SEEN_EVENTS + + +class TestUpstreamMessages: + """MESSAGE handling and acknowledgement.""" + + @pytest.mark.asyncio + async def test_event_is_published_and_acknowledged(self): + upstream = _Upstream("osdf/pub", CONFIG) + queue = upstream.add_listener() + connection = AsyncMock() + frame = Frame( + "MESSAGE", + {"ack": "a-1", "destination": "ep-test/osdf/pub"}, + _event_body(uuid="u-1"), + ) + + await upstream._on_message(connection, frame) + + assert queue.get_nowait()["name"] == "data.csv" + connection.send.assert_awaited_once() + assert b"ACK" in connection.send.await_args.args[0] + + @pytest.mark.asyncio + async def test_duplicate_is_suppressed_but_still_acknowledged(self): + """ + An unacknowledged message is redelivered forever, so a duplicate + has to be acknowledged even though it is not published again. + """ + upstream = _Upstream("osdf/pub", CONFIG) + queue = upstream.add_listener() + connection = AsyncMock() + frame = Frame("MESSAGE", {"ack": "a-1"}, _event_body(uuid="u-1")) + + await upstream._on_message(connection, frame) + await upstream._on_message(connection, frame) + + assert queue.qsize() == 1 + assert connection.send.await_count == 2 + + @pytest.mark.asyncio + async def test_unreadable_event_is_dropped_but_acknowledged(self): + upstream = _Upstream("osdf/pub", CONFIG) + queue = upstream.add_listener() + connection = AsyncMock() + frame = Frame("MESSAGE", {"ack": "a-1"}, "not json") + + await upstream._on_message(connection, frame) + + assert queue.empty() + connection.send.assert_awaited_once() + + +class TestBroker: + """One upstream per event source, reference counted.""" + + @pytest.mark.asyncio + async def test_two_listeners_share_one_upstream(self): + broker = PelicanEventBroker() + with patch.object(_Upstream, "start"): + first = broker.listen("osdf/pub", CONFIG) + second = broker.listen("osdf/pub", CONFIG) + task_a = asyncio.ensure_future(first.__anext__()) + task_b = asyncio.ensure_future(second.__anext__()) + await asyncio.sleep(0) + + assert len(broker._upstreams) == 1 + upstream = broker._upstreams["osdf/pub"] + assert len(upstream.listeners) == 2 + + upstream._publish({"name": "a.csv"}) + assert await task_a == {"name": "a.csv"} + assert await task_b == {"name": "a.csv"} + + await first.aclose() + await second.aclose() + + @pytest.mark.asyncio + async def test_upstream_is_dropped_when_the_last_listener_leaves(self): + """An idle Endpoint should hold no connection to the server.""" + broker = PelicanEventBroker() + with ( + patch.object(_Upstream, "start"), + patch.object(_Upstream, "stop", new=AsyncMock()), + ): + stream = broker.listen("osdf/pub", CONFIG, idle_timeout=0.01) + # One keepalive tick is enough to get the upstream registered. + assert await stream.__anext__() is None + assert len(broker._upstreams) == 1 + + await stream.aclose() + + assert broker._upstreams == {} + + @pytest.mark.asyncio + async def test_idle_stream_yields_none_for_a_keepalive(self): + """ + The wait has to time out inside the generator: cancelling an + __anext__ from outside would unwind it and tear the + subscription down on every quiet interval. + """ + broker = PelicanEventBroker() + with patch.object(_Upstream, "start"): + stream = broker.listen("osdf/pub", CONFIG, idle_timeout=0.01) + + assert await stream.__anext__() is None + + upstream = broker._upstreams["osdf/pub"] + upstream._publish({"name": "a.csv"}) + assert await stream.__anext__() == {"name": "a.csv"} + + await stream.aclose() + + @pytest.mark.asyncio + async def test_empty_event_source_is_refused(self): + broker = PelicanEventBroker() + with pytest.raises(EventSubscriptionUnavailable): + await broker.listen(" / ", CONFIG).__anext__() + + def test_status_reports_each_upstream(self): + broker = PelicanEventBroker() + broker._upstreams["osdf/pub"] = _Upstream("osdf/pub", CONFIG) + + status = broker.status() + + assert status["osdf/pub"]["destination"] == "ep-test/osdf/pub" + assert status["osdf/pub"]["listeners"] == 0 + + +class TestSubscribeRoute: + """GET /pelican/subscribe and /pelican/subscriptions.""" + + @staticmethod + def _client(): + from fastapi import FastAPI + from fastapi.testclient import TestClient + from api.routes.pelican_routes import router + + app = FastAPI() + app.include_router(router) + return app, TestClient(app) + + @staticmethod + def _as(roles): + return lambda: { + "roles": roles, + "groups": [], + "sub": "test_user", + "username": "Test User", + } + + def test_requires_authentication(self): + """The stream inherits the router gate added for issue #261.""" + _app, client = self._client() + + response = client.get("/pelican/subscribe", params={"event_source": "osdf/pub"}) + + assert response.status_code == 401 + + @patch("api.routes.pelican_routes.load_config") + def test_unconfigured_endpoint_reports_unavailable(self, mock_config): + mock_config.side_effect = EventSubscriptionUnavailable("no client id") + from api.services.auth_services import get_current_user + + app, client = self._client() + app.dependency_overrides[get_current_user] = self._as(["ndp_viewer"]) + try: + response = client.get( + "/pelican/subscribe", params={"event_source": "osdf/pub"} + ) + finally: + app.dependency_overrides.clear() + + assert response.status_code == 503 + assert response.json()["detail"] == "no client id" + + @patch("api.routes.pelican_routes.broker") + @patch("api.routes.pelican_routes.load_config") + def test_stream_emits_events_and_keepalives(self, mock_config, mock_broker): + mock_config.return_value = CONFIG + + async def fake_listen(event_source, config, *args, **kwargs): + yield None + yield {"name": "a.csv", "url": "pelican://x/a.csv"} + + mock_broker.listen = fake_listen + from api.services.auth_services import get_current_user + + app, client = self._client() + app.dependency_overrides[get_current_user] = self._as(["ndp_viewer"]) + try: + response = client.get( + "/pelican/subscribe", params={"event_source": "osdf/pub"} + ) + finally: + app.dependency_overrides.clear() + + body = response.text + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + assert ": subscribed" in body + assert ": keepalive" in body + assert "event: file" in body + assert '"name": "a.csv"' in body + + @patch("api.routes.pelican_routes.broker") + def test_subscriptions_lists_upstream_state(self, mock_broker): + mock_broker.status = MagicMock( + return_value={"osdf/pub": {"state": "connected", "listeners": 2}} + ) + from api.services.auth_services import get_current_user + + app, client = self._client() + app.dependency_overrides[get_current_user] = self._as(["ndp_viewer"]) + try: + response = client.get("/pelican/subscriptions") + finally: + app.dependency_overrides.clear() + + assert response.status_code == 200 + assert response.json()["subscriptions"]["osdf/pub"]["listeners"] == 2 From bfefca0b8bd365c632424584b4dcfd24f28d9a7b Mon Sep 17 00:00:00 2001 From: Raul Bardaji Date: Mon, 31 Aug 2026 11:27:08 +0200 Subject: [PATCH 2/2] feat(pelican): let a subscriber bring its own event server credentials The event server identity and credentials were the Endpoint's alone, so every caller subscribed as the Endpoint. They are now accepted per request as client_id, username and password, falling back to the Endpoint's configuration for callers without an account of their own. They are taken from headers as well as query parameters, and the headers win: uvicorn and nginx both write the query string to their access logs, which would put a password on disk in plain text. Credentials are taken as a pair, so supplying only a username cannot borrow the Endpoint's password and sign the caller in as the Endpoint under another name. The upstream is now keyed by client id as well as event source. Callers sharing an identity still share one connection and each receive every event; a caller with its own gets its own, since two identities cannot be served over one authenticated session. Part of #262. --- CHANGELOG.md | 10 +- api/routes/pelican_routes.py | 55 ++++- .../pelican_services/event_subscription.py | 119 +++++++--- docs/configuration.md | 26 ++- tests/test_pelican_events.py | 219 +++++++++++++++++- 5 files changed, 373 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f473b7..392aa08 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,18 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - **`GET /pelican/subscribe` streams Pelican file events as Server-Sent Events.** The event server notifies subscribers whenever an object appears in a Pelican namespace, but that stream existed only in the `ndp-ep` client library; the API had no way to expose it, so an Endpoint could not offer subscriptions to its own callers. The new route returns a `text/event-stream` where each notification arrives as an `event: file` message carrying the object's `name`, `url`, `size` and `mod_time`. The `url` goes straight to `/pelican/read` for the contents or `/pelican/download` for the file, which is the pipeline the route exists to enable: subscribe, then read what arrived. A `: keepalive` comment is emitted during quiet periods so a proxy does not mistake an idle stream for a dead one, and `X-Accel-Buffering: no` stops nginx holding events back until its buffer fills. -- **`GET /pelican/subscriptions` reports the upstream subscriptions the Endpoint holds**, with each one's connection state, listener count and how many events were dropped for slow listeners. +- **`GET /pelican/subscriptions` reports the upstream subscriptions the Endpoint holds**, with each one's connection state, event source, client id, listener count and how many events were dropped for slow listeners. No credential appears in it — the username is reported only as an `authenticated` flag — since any viewer on the Endpoint can read the route. +- **A caller may bring its own event server identity and credentials.** `client_id`, `username` and `password` are accepted as query parameters, and as the `X-Pelican-Event-Client-Id`, `X-Pelican-Event-Username` and `X-Pelican-Event-Password` headers, which take precedence; anything omitted falls back to the Endpoint's own configuration, so callers with an account of their own and callers without one are served by the same route. The headers exist because a query string is written to the access logs of both uvicorn and nginx, which would put a password on disk in plain text. Credentials are taken as a pair: supplying only a username is refused rather than silently borrowing the Endpoint's password, which would sign the caller in as the Endpoint under a name of their choosing. ### Changed - The event server speaks STOMP 1.2 over a WebSocket, so `websockets` is now a dependency. ### Notes on the design -- **One upstream subscription per event source, fanned out to every listener.** The event server requires a unique `client-id` per subscriber: two connections sharing one compete for the same events rather than both receiving them. Opening a subscription per SSE caller would therefore either split the stream between callers or leave an orphaned client id registered upstream on every connection. Instead the Endpoint subscribes once per event source under its own stable id and distributes each event to all its listeners. The upstream connection opens when the first listener arrives and closes when the last one leaves, so an idle Endpoint holds no connection at all. +- **One upstream subscription per destination, fanned out to every listener.** The event server requires a unique `client-id` per subscriber: two connections sharing one are served by splitting the events between them, so each sees only a fraction. Opening a subscription per SSE caller would therefore either split the stream that way or leave an orphaned client id registered upstream on every connection. Instead the upstream is keyed by client id *and* event source: subscribers presenting the same identity share one connection and each receive every event on it, while a caller bringing its own credentials gets its own, since two identities cannot be served over a single authenticated session. The connection opens when the first listener arrives and closes when the last one leaves, so an idle Endpoint holds none at all. - **The STOMP protocol is implemented here rather than taken from the client library.** Depending on `ndp-ep` would make the API depend on its own client SDK and pull in `pelicanfs`, which pins Python 3.11. The cost is that the wire format now lives in two repositories and has to be kept in step by hand. - **Per-listener buffers are bounded**, unlike the client library's. A caller that stops reading must not be able to grow the Endpoint's memory without limit, so a full buffer drops its oldest event and counts it in `/pelican/subscriptions`. +### Known limitation +- **The broker is per worker process, not per Endpoint.** `Dockerfile.allinone` runs uvicorn with four workers, and each holds its own broker. Two callers relying on the Endpoint's configured client id can therefore land on different workers and open two upstream connections under one identity, which the event server serves by splitting the events between them — so each would see only part of the stream. Callers that supply their own client id are unaffected, and so is any deployment running a single worker. Fixing it properly needs the subscription to live outside the worker processes. + ### Backwards compatibility -- Purely additive. Both routes sit behind the authorization added in #261 and are only mounted when `PELICAN_ENABLED` is set. Subscriptions need `PELICAN_EVENT_CLIENT_ID` — or an `AFFINITIES_EP_UUID` to derive it from — and `/pelican/subscribe` answers 503 with the reason when the event server is not configured, so an Endpoint that never sets it is unaffected. The other new settings (`PELICAN_EVENT_SERVER_URL`, `PELICAN_EVENT_USERNAME`, `PELICAN_EVENT_PASSWORD`, `PELICAN_EVENT_VIRTUAL_HOST`, `PELICAN_EVENT_HEARTBEAT_MS`) are optional and documented in `example.env` and `docs/configuration.md`. +- Purely additive. Both routes sit behind the authorization added in #261 and are only mounted when `PELICAN_ENABLED` is set. Subscriptions need a client id from somewhere — the request, `PELICAN_EVENT_CLIENT_ID`, or an `AFFINITIES_EP_UUID` to derive it from — and `/pelican/subscribe` answers 503 with the reason when none is available, so an Endpoint that never configures the event server is unaffected. The other new settings (`PELICAN_EVENT_SERVER_URL`, `PELICAN_EVENT_USERNAME`, `PELICAN_EVENT_PASSWORD`, `PELICAN_EVENT_VIRTUAL_HOST`, `PELICAN_EVENT_HEARTBEAT_MS`) are optional and documented in `example.env` and `docs/configuration.md`. ## [0.34.23] - 2026-08-30 diff --git a/api/routes/pelican_routes.py b/api/routes/pelican_routes.py index 56f4796..ef3b27e 100644 --- a/api/routes/pelican_routes.py +++ b/api/routes/pelican_routes.py @@ -8,6 +8,7 @@ from fastapi import ( APIRouter, Depends, + Header, HTTPException, Query, Request, @@ -363,14 +364,35 @@ async def subscribe_to_events( event_source: str = Query( ..., description="Namespace to watch, e.g. osdf/vdc/public/data" ), + client_id: Optional[str] = Query( + None, + description=( + "Identity to present to the event server. Must be unique: " + "two subscribers sharing one are served by splitting the " + "events between them. Defaults to the Endpoint's own." + ), + ), + username: Optional[str] = Query( + None, + description=( + "Event server username. Prefer the X-Pelican-Event-Username " + "header. Defaults to the Endpoint's own credentials." + ), + ), + password: Optional[str] = Query( + None, + description=( + "Event server password. Prefer the X-Pelican-Event-Password " + "header: a query string is written to the access log." + ), + ), + header_client_id: Optional[str] = Header(None, alias="X-Pelican-Event-Client-Id"), + header_username: Optional[str] = Header(None, alias="X-Pelican-Event-Username"), + header_password: Optional[str] = Header(None, alias="X-Pelican-Event-Password"), ): """ Stream Pelican file events as Server-Sent Events. - The Endpoint holds one upstream subscription per event source and - fans it out, so every listener on a source receives every event - rather than competing for them. - Each event arrives as an ``event: file`` message whose data carries the object's ``name``, ``url``, ``size`` and ``mod_time``. The ``url`` can be handed straight to ``/pelican/read`` to get the @@ -379,12 +401,29 @@ async def subscribe_to_events( A ``: keepalive`` comment is sent during quiet periods, so a proxy does not mistake an idle stream for a dead one. + The caller may bring its own event server identity and credentials; + anything it omits falls back to the Endpoint's configuration. They + are accepted both as query parameters and as headers, and the + headers win. **Prefer the headers**: a query string is recorded in + the access log of both uvicorn and nginx, so a password passed that + way is written to disk in plain text. + + Subscribers presenting the same client id share one upstream + connection and each receive every event on it. A caller with its own + id gets its own connection, since the two cannot be served over a + single authenticated session. + Parameters ---------- request : Request Used to notice that the caller has gone away. event_source : str Namespace to watch. + client_id, username, password : str, optional + Event server identity and credentials, overriding the + Endpoint's. + header_client_id, header_username, header_password : str, optional + The same three, taken from headers, which take precedence. Returns ------- @@ -394,10 +433,14 @@ async def subscribe_to_events( Raises ------ HTTPException - 503 if the event server is not configured on this Endpoint. + 503 if no usable event server configuration results. """ try: - config = load_config() + config = load_config( + client_id=header_client_id or client_id, + username=header_username or username, + password=header_password or password, + ) except EventSubscriptionUnavailable as exc: raise HTTPException(status_code=503, detail=str(exc)) diff --git a/api/services/pelican_services/event_subscription.py b/api/services/pelican_services/event_subscription.py index 1390bf1..595ed3d 100644 --- a/api/services/pelican_services/event_subscription.py +++ b/api/services/pelican_services/event_subscription.py @@ -75,9 +75,28 @@ class EventServerConfig: heartbeat_ms: int -def load_config() -> EventServerConfig: +def load_config( + client_id: Optional[str] = None, + username: Optional[str] = None, + password: Optional[str] = None, +) -> EventServerConfig: """ - Read the event server settings from the environment. + Resolve the event server settings for one subscription. + + The caller may bring its own identity and credentials; whatever it + does not supply falls back to the Endpoint's own configuration. That + is what lets an Endpoint serve callers who have an account on the + event server and callers who do not, from the same route. + + Parameters + ---------- + client_id : str, optional + Identity to present to the event server, overriding + ``PELICAN_EVENT_CLIENT_ID``. + username : str, optional + Event server username, overriding ``PELICAN_EVENT_USERNAME``. + password : str, optional + Event server password, overriding ``PELICAN_EVENT_PASSWORD``. Returns ------- @@ -103,35 +122,47 @@ def load_config() -> EventServerConfig: except ValueError as exc: raise EventSubscriptionUnavailable(str(exc)) from exc - # A stable identity per Endpoint. Falling back to the Endpoint UUID - # keeps it unique across deployments without another thing to set. - client_id = ( - os.getenv("PELICAN_EVENT_CLIENT_ID") or os.getenv("AFFINITIES_EP_UUID") or "" + # A caller's own identity wins; otherwise the Endpoint's, falling + # back to its UUID so a deployment has a unique one without another + # thing to set. + resolved_id = ( + client_id + or os.getenv("PELICAN_EVENT_CLIENT_ID") + or os.getenv("AFFINITIES_EP_UUID") + or "" ).strip() - if not client_id: + if not resolved_id: raise EventSubscriptionUnavailable( - "PELICAN_EVENT_CLIENT_ID is not set and no Endpoint UUID is " - "available to derive it from. The event server requires a " - "unique client id per subscriber." + "No client id: pass one on the request, or set " + "PELICAN_EVENT_CLIENT_ID on the Endpoint. The event server " + "requires a unique client id per subscriber." ) - if "/" in client_id: + if "/" in resolved_id: raise EventSubscriptionUnavailable( - "PELICAN_EVENT_CLIENT_ID must not contain '/': it is one " - "segment of the STOMP destination." + "The client id must not contain '/': it is one segment of " + "the STOMP destination." ) - username = os.getenv("PELICAN_EVENT_USERNAME", "") - password = os.getenv("PELICAN_EVENT_PASSWORD", "") - if bool(username) != bool(password): - raise EventSubscriptionUnavailable( - "PELICAN_EVENT_USERNAME and PELICAN_EVENT_PASSWORD must be " "set together." - ) + # Credentials are taken as a pair. Supplying only a username on the + # request must not silently fall back to the Endpoint's password, + # which would sign the caller in as the Endpoint under another name. + if username is not None or password is not None: + resolved_user = username or "" + resolved_password = password or "" + source = "The username and password" + else: + resolved_user = os.getenv("PELICAN_EVENT_USERNAME", "") + resolved_password = os.getenv("PELICAN_EVENT_PASSWORD", "") + source = "PELICAN_EVENT_USERNAME and PELICAN_EVENT_PASSWORD" + + if bool(resolved_user) != bool(resolved_password): + raise EventSubscriptionUnavailable(f"{source} must be set together.") return EventServerConfig( url=url, - client_id=client_id, - username=username, - password=password, + client_id=resolved_id, + username=resolved_user, + password=resolved_password, virtual_host=os.getenv("PELICAN_EVENT_VIRTUAL_HOST", DEFAULT_VIRTUAL_HOST), heartbeat_ms=_heartbeat_ms(), ) @@ -380,23 +411,39 @@ async def _on_message(self, connection, frame: Frame) -> None: class PelicanEventBroker: - """Holds one :class:`_Upstream` per event source, reference counted.""" + """ + Holds one :class:`_Upstream` per destination, reference counted. + + The key is the STOMP destination — client id *and* event source — + not the event source alone. Callers presenting the same identity + share one connection and every event on it; a caller bringing its + own credentials gets its own, because the two cannot be served over + a single authenticated session. + """ def __init__(self) -> None: self._upstreams: Dict[str, _Upstream] = {} self._lock = asyncio.Lock() def status(self) -> Dict[str, dict]: - """Report what is currently subscribed, for diagnostics.""" + """ + Report what is currently subscribed, for diagnostics. + + Keyed by destination. No credential appears here — the username + is reported only as a flag, since this route is readable by any + viewer on the Endpoint. + """ return { - source: { + destination: { "state": upstream.state, - "destination": upstream.destination, + "event_source": upstream.event_source, + "client_id": upstream.config.client_id, + "authenticated": bool(upstream.config.username), "listeners": len(upstream.listeners), "dropped_events": upstream.dropped, "last_error": upstream.last_error, } - for source, upstream in self._upstreams.items() + for destination, upstream in self._upstreams.items() } async def listen( @@ -438,11 +485,13 @@ async def listen( "event_source must be a non-empty namespace path." ) + key = f"{config.client_id}/{source}" + async with self._lock: - upstream = self._upstreams.get(source) + upstream = self._upstreams.get(key) if upstream is None: upstream = _Upstream(source, config) - self._upstreams[source] = upstream + self._upstreams[key] = upstream upstream.start() queue = upstream.add_listener() @@ -456,10 +505,16 @@ async def listen( async with self._lock: upstream.remove_listener(queue) if not upstream.listeners: - self._upstreams.pop(source, None) + self._upstreams.pop(key, None) await upstream.stop() -#: Process-wide broker. One per worker, which is what keeps the client -#: id unique: two workers would otherwise share it and split the stream. +#: Broker for this worker process. +#: +#: Note that it is per *process*, not per Endpoint: the shipped image +#: runs uvicorn with several workers, so two callers presenting the same +#: client id can land on different workers and open two upstream +#: connections under one identity — which the event server serves by +#: splitting the events between them. Callers that bring their own +#: client id are unaffected. See the CHANGELOG for the open issue. broker = PelicanEventBroker() diff --git a/docs/configuration.md b/docs/configuration.md index afb656a..321cd1c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -355,17 +355,29 @@ Event server backing `GET /pelican/subscribe`, as an `http(s)` or `ws(s)` URL. #### `PELICAN_EVENT_CLIENT_ID` *Optional · default: `AFFINITIES_EP_UUID`.* -Identifies this Endpoint to the event server. **Must be unique**: two -subscribers sharing an id compete for the same events instead of both receiving -them. Must not contain `/`, since it is one segment of the STOMP destination. -`GET /pelican/subscribe` returns 503 when neither this nor the Endpoint UUID is -set. **Where:** leave empty unless one host runs several Endpoints. +Identity this Endpoint presents to the event server, used for callers that do +not bring their own. **Must be unique**: two subscribers sharing an id are +served by splitting the events between them, so each sees only a fraction. Must +not contain `/`, since it is one segment of the STOMP destination. When neither +this, the Endpoint UUID, nor a caller-supplied id is available, +`GET /pelican/subscribe` answers 503 saying so. **Where:** leave empty unless +one host runs several Endpoints. #### `PELICAN_EVENT_USERNAME` / `PELICAN_EVENT_PASSWORD` *Optional · default: empty. Set both or neither.* Credentials the event server checks against its own store — unrelated to the -Endpoint token. Temporary: they are due to be replaced by an access token -issued for NDP. +Endpoint token. Used for callers that do not supply their own. Temporary: they +are due to be replaced by an access token issued for NDP. + +> **Callers may override all three.** `GET /pelican/subscribe` accepts +> `client_id`, `username` and `password` as query parameters, and the same +> three as the `X-Pelican-Event-Client-Id`, `X-Pelican-Event-Username` and +> `X-Pelican-Event-Password` headers, which win. Prefer the headers: a query +> string is written to the access logs of both uvicorn and nginx, so a password +> passed that way lands on disk in plain text. Credentials are taken as a pair — +> supplying only a username is refused rather than borrowing the Endpoint's +> password. Subscribers presenting the same client id share one upstream +> connection and each receive every event on it. #### `PELICAN_EVENT_VIRTUAL_HOST` *Optional · default: `playground`.* diff --git a/tests/test_pelican_events.py b/tests/test_pelican_events.py index 164ca9a..9b4bbdd 100644 --- a/tests/test_pelican_events.py +++ b/tests/test_pelican_events.py @@ -2,6 +2,7 @@ import asyncio import json +from dataclasses import replace from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -191,6 +192,70 @@ def test_unreadable_body_still_yields_an_identity(self): assert event_identity(frame) == "unknown-message" +class TestCallerSuppliedCredentials: + """A caller may bring its own identity and credentials.""" + + @staticmethod + def _env(**overrides): + env = { + "PELICAN_EVENT_CLIENT_ID": "ep-test", + "PELICAN_EVENT_USERNAME": "ep-user", + "PELICAN_EVENT_PASSWORD": "ep-secret", + } + env.update(overrides) + return env + + def test_caller_values_win_over_the_endpoint(self): + with patch.dict("os.environ", self._env(), clear=True): + config = load_config( + client_id="caller-1", username="mine", password="also-mine" + ) + + assert config.client_id == "caller-1" + assert config.username == "mine" + assert config.password == "also-mine" + + def test_omitted_values_fall_back_to_the_endpoint(self): + """Callers without an account of their own still work.""" + with patch.dict("os.environ", self._env(), clear=True): + config = load_config() + + assert config.client_id == "ep-test" + assert config.username == "ep-user" + + def test_a_caller_id_alone_still_uses_endpoint_credentials(self): + with patch.dict("os.environ", self._env(), clear=True): + config = load_config(client_id="caller-1") + + assert config.client_id == "caller-1" + assert config.username == "ep-user" + + def test_half_a_caller_credential_does_not_borrow_the_other_half(self): + """ + Falling back per field would sign the caller in as the Endpoint + under a name of their choosing, so the pair is taken together. + """ + with patch.dict("os.environ", self._env(), clear=True): + with pytest.raises(EventSubscriptionUnavailable) as exc: + load_config(username="mine") + + assert "must be set together" in str(exc.value) + + def test_a_caller_id_with_a_slash_is_refused(self): + with patch.dict("os.environ", self._env(), clear=True): + with pytest.raises(EventSubscriptionUnavailable): + load_config(client_id="a/b") + + def test_a_caller_id_lets_an_unconfigured_endpoint_serve(self): + """An Endpoint with no event settings of its own still works.""" + with patch.dict("os.environ", {}, clear=True): + config = load_config( + client_id="caller-1", username="mine", password="also-mine" + ) + + assert config.client_id == "caller-1" + + class TestLoadConfig: """Environment resolution.""" @@ -392,10 +457,10 @@ async def test_unreadable_event_is_dropped_but_acknowledged(self): class TestBroker: - """One upstream per event source, reference counted.""" + """One upstream per destination, reference counted.""" @pytest.mark.asyncio - async def test_two_listeners_share_one_upstream(self): + async def test_two_listeners_on_one_identity_share_an_upstream(self): broker = PelicanEventBroker() with patch.object(_Upstream, "start"): first = broker.listen("osdf/pub", CONFIG) @@ -405,7 +470,7 @@ async def test_two_listeners_share_one_upstream(self): await asyncio.sleep(0) assert len(broker._upstreams) == 1 - upstream = broker._upstreams["osdf/pub"] + upstream = broker._upstreams["ep-test/osdf/pub"] assert len(upstream.listeners) == 2 upstream._publish({"name": "a.csv"}) @@ -415,6 +480,34 @@ async def test_two_listeners_share_one_upstream(self): await first.aclose() await second.aclose() + @pytest.mark.asyncio + async def test_a_caller_with_its_own_identity_gets_its_own_upstream(self): + """ + Two identities cannot share one authenticated session, so the + upstream is keyed by client id as well as event source. + """ + other = replace(CONFIG, client_id="caller-1", username="other") + broker = PelicanEventBroker() + with patch.object(_Upstream, "start"): + first = broker.listen("osdf/pub", CONFIG, idle_timeout=0.01) + second = broker.listen("osdf/pub", other, idle_timeout=0.01) + # A keepalive tick apiece registers both upstreams. + assert await first.__anext__() is None + assert await second.__anext__() is None + + assert set(broker._upstreams) == { + "ep-test/osdf/pub", + "caller-1/osdf/pub", + } + + # An event on one identity must not reach the other. + broker._upstreams["caller-1/osdf/pub"]._publish({"name": "a.csv"}) + assert await second.__anext__() == {"name": "a.csv"} + assert await first.__anext__() is None + + await first.aclose() + await second.aclose() + @pytest.mark.asyncio async def test_upstream_is_dropped_when_the_last_listener_leaves(self): """An idle Endpoint should hold no connection to the server.""" @@ -445,7 +538,7 @@ async def test_idle_stream_yields_none_for_a_keepalive(self): assert await stream.__anext__() is None - upstream = broker._upstreams["osdf/pub"] + upstream = broker._upstreams["ep-test/osdf/pub"] upstream._publish({"name": "a.csv"}) assert await stream.__anext__() == {"name": "a.csv"} @@ -459,12 +552,27 @@ async def test_empty_event_source_is_refused(self): def test_status_reports_each_upstream(self): broker = PelicanEventBroker() - broker._upstreams["osdf/pub"] = _Upstream("osdf/pub", CONFIG) + broker._upstreams["ep-test/osdf/pub"] = _Upstream("osdf/pub", CONFIG) - status = broker.status() + status = broker.status()["ep-test/osdf/pub"] - assert status["osdf/pub"]["destination"] == "ep-test/osdf/pub" - assert status["osdf/pub"]["listeners"] == 0 + assert status["event_source"] == "osdf/pub" + assert status["client_id"] == "ep-test" + assert status["listeners"] == 0 + + def test_status_never_reports_a_credential(self): + """ + Any viewer on the Endpoint can read this route, so it must not + echo back a password another caller supplied. + """ + broker = PelicanEventBroker() + broker._upstreams["ep-test/osdf/pub"] = _Upstream("osdf/pub", CONFIG) + + rendered = json.dumps(broker.status()) + + assert CONFIG.password not in rendered + assert CONFIG.username not in rendered + assert broker.status()["ep-test/osdf/pub"]["authenticated"] is True class TestSubscribeRoute: @@ -497,6 +605,101 @@ def test_requires_authentication(self): assert response.status_code == 401 + def _subscribe(self, params=None, headers=None): + """Call the route as a viewer, returning the response.""" + from api.services.auth_services import get_current_user + + app, client = self._client() + app.dependency_overrides[get_current_user] = self._as(["ndp_viewer"]) + try: + return client.get( + "/pelican/subscribe", + params={"event_source": "osdf/pub", **(params or {})}, + headers=headers or {}, + ) + finally: + app.dependency_overrides.clear() + + @patch("api.routes.pelican_routes.broker") + @patch("api.routes.pelican_routes.load_config") + def test_credentials_are_accepted_as_query_parameters( + self, mock_config, mock_broker + ): + mock_config.return_value = CONFIG + + async def fake_listen(event_source, config, *args, **kwargs): + yield None + + mock_broker.listen = fake_listen + + self._subscribe( + { + "client_id": "caller-1", + "username": "mine", + "password": "also-mine", + } + ) + + mock_config.assert_called_once_with( + client_id="caller-1", username="mine", password="also-mine" + ) + + @patch("api.routes.pelican_routes.broker") + @patch("api.routes.pelican_routes.load_config") + def test_headers_win_over_query_parameters(self, mock_config, mock_broker): + """ + A query string is written to the access log, so the header is + the safer way to pass a password and has to take precedence. + """ + mock_config.return_value = CONFIG + + async def fake_listen(event_source, config, *args, **kwargs): + yield None + + mock_broker.listen = fake_listen + + self._subscribe( + {"client_id": "from-query", "username": "q", "password": "q-pass"}, + { + "X-Pelican-Event-Client-Id": "from-header", + "X-Pelican-Event-Username": "h", + "X-Pelican-Event-Password": "h-pass", + }, + ) + + mock_config.assert_called_once_with( + client_id="from-header", username="h", password="h-pass" + ) + + @patch("api.routes.pelican_routes.broker") + @patch("api.routes.pelican_routes.load_config") + def test_nothing_supplied_falls_back_to_the_endpoint( + self, mock_config, mock_broker + ): + mock_config.return_value = CONFIG + + async def fake_listen(event_source, config, *args, **kwargs): + yield None + + mock_broker.listen = fake_listen + + self._subscribe() + + mock_config.assert_called_once_with( + client_id=None, username=None, password=None + ) + + @patch("api.routes.pelican_routes.load_config") + def test_a_rejected_credential_is_reported_not_swallowed(self, mock_config): + mock_config.side_effect = EventSubscriptionUnavailable( + "The username and password must be set together." + ) + + response = self._subscribe({"username": "mine"}) + + assert response.status_code == 503 + assert "set together" in response.json()["detail"] + @patch("api.routes.pelican_routes.load_config") def test_unconfigured_endpoint_reports_unavailable(self, mock_config): mock_config.side_effect = EventSubscriptionUnavailable("no client id")