connected_accounts/eventsub.py
Rama main · réplica de lectura del repositorio original.
"""Host-owned Twitch EventSub WebSockets and bounded, token-free events.
The welcome message, rather than a plugin or renderer, supplies the session ID
used for subscriptions. A redirected/reconnecting socket is accepted only on
Twitch's fixed EventSub host. Native plugins still run as the local OS user;
this is an API permission boundary, not an operating-system sandbox.
"""
from __future__ import annotations
import json
import re
import secrets
import threading
import time
from collections import deque
from dataclasses import dataclass, field
from urllib.parse import urlsplit
from connected_accounts.models import AccountError
SOCKET_URL = "wss://eventsub.wss.twitch.tv/ws"
EVENT_TYPES = frozenset({"stream.online", "stream.offline", "channel.update", "channel.chat.message"})
MAX_EVENTS = 500
_NUMBER = re.compile(r"^[0-9]{1,30}$")
def _text(value, limit=500) -> str:
return value[:limit] if isinstance(value, str) else ""
def _reconnect_url(value) -> str:
if not isinstance(value, str) or len(value) > 4096 or any(ord(c) < 33 for c in value):
raise ValueError("Invalid EventSub reconnect destination")
parsed = urlsplit(value)
if (parsed.scheme != "wss" or parsed.hostname != "eventsub.wss.twitch.tv"
or parsed.username is not None or parsed.password is not None or parsed.port is not None
or parsed.fragment or parsed.path != "/ws"):
raise ValueError("Invalid EventSub reconnect destination")
return value
def _connect(url):
from websockets.sync.client import connect
# No implicit redirects, unbounded incoming messages or compression state.
return connect(url, open_timeout=10, close_timeout=2, max_size=1_048_576,
max_queue=16, compression=None)
@dataclass
class _Stream:
account_id: str
broadcaster_id: str
types: tuple[str, ...]
stopped: threading.Event = field(default_factory=threading.Event)
lock: threading.RLock = field(default_factory=threading.RLock)
sockets: list = field(default_factory=list)
buffer: deque = field(default_factory=lambda: deque(maxlen=MAX_EVENTS))
seen: deque = field(default_factory=lambda: deque(maxlen=1000))
cursor: int = 0
status: str = "connecting"
error: str = ""
thread: threading.Thread | None = None
class _Revoked(Exception):
pass
class EventSubManager:
"""One listener per connected account; all subscriptions are host-owned.
``connector`` and ``clock`` permit deterministic offline tests. Each
``start`` names its consumer (the desktop, stream-workflow monitoring, one
plugin run); the listener subscribes to the union of their event types.
``release`` drops one consumer: the last one out stops the listener and
clears its buffer, and a narrower union resubscribes without the types
nobody needs (chat stops being read when no consumer wants it). ``stop``
and ``close`` must be called by the host on account disconnection and
application exit.
After every fresh subscription, sinks get a ``session.subscribed`` event
(never buffered): EventSub does not replay ``stream.online`` for a stream
that started while the socket was reconnecting.
"""
def __init__(self, service, *, connector=None, clock=time.monotonic):
self.service = service
self._connect = connector or _connect
self._clock = clock
self._lock = threading.RLock()
self._streams: dict[str, _Stream] = {}
self._sequences: dict[str, int] = {}
self._sequence_lock = threading.Lock()
self._closed = False
self._sinks = []
self._consumers: dict[str, dict[str, tuple[str, ...]]] = {}
def add_sink(self, callback):
"""Receive sanitized host events; OAuth credentials never reach a sink."""
if not callable(callback):
raise TypeError("An event sink must be callable")
with self._lock:
if callback not in self._sinks:
self._sinks.append(callback)
def remove_sink(self, callback):
with self._lock:
self._sinks = [sink for sink in self._sinks if sink != callback]
def start(self, account_id: str, broadcaster_id: str, event_types, *, consumer: str = "desktop") -> dict:
if not isinstance(broadcaster_id, str) or not _NUMBER.fullmatch(broadcaster_id):
raise AccountError("invalid_request", "Choose a valid Twitch channel.")
if (not isinstance(event_types, (list, tuple)) or not event_types or len(event_types) > 4
or any(not isinstance(t, str) or t not in EVENT_TYPES for t in event_types)):
raise AccountError("invalid_request", "Choose supported Twitch event types.")
types = tuple(sorted(set(event_types)))
account = next((a for a in self.service.accounts() if a.get("id") == account_id), None)
if account is None or account.get("provider") != "twitch":
raise AccountError("account_not_found", "Connect a Twitch account first.", status=404)
if account.get("status") != "connected":
raise AccountError("reconnect_required", "Reconnect Twitch to receive events.", status=401)
if not isinstance(account.get("user_id"), str) or not _NUMBER.fullmatch(account["user_id"]):
raise AccountError("reconnect_required", "Reconnect Twitch before receiving events.", status=401)
if "channel.chat.message" in types and "user:read:chat" not in account.get("scopes", []):
raise AccountError("missing_scope", "Reconnect Twitch with permission to read chat.", status=403)
with self._lock:
if self._closed:
raise AccountError("service_closed", "Connected Accounts is stopping.", status=503)
consumers = self._consumers.setdefault(account_id, {})
existing = self._streams.get(account_id)
if existing is not None and existing.broadcaster_id != broadcaster_id:
if any(name != consumer for name in consumers):
raise AccountError("channel_in_use", "Another feature is monitoring a different channel. "
"Stop it before changing the channel.", status=409)
# Switching channels is a new selection, not permission to
# carry the previous channel's requested chat types across.
consumers.clear()
consumers[consumer] = tuple(sorted(set(types) | set(consumers.get(consumer, ()))))
types = self._union(consumers)
if "user:read:chat" not in account.get("scopes", []):
types = tuple(t for t in types if t != "channel.chat.message")
if (existing is not None and existing.broadcaster_id == broadcaster_id
and existing.types == types and existing.thread is not None
and existing.thread.is_alive()):
return self._status(existing)
stream, replaced = self._launch(account_id, broadcaster_id, types)
self._join(replaced)
return self._status(stream)
@staticmethod
def _union(consumers: dict) -> tuple[str, ...]:
return tuple(sorted(set().union(*consumers.values()))) if consumers else ()
def _launch(self, account_id: str, broadcaster_id: str,
types: tuple[str, ...]) -> tuple[_Stream, _Stream | None]:
"""Replace the account's listener. Call with the manager lock held,
then ``_join`` the replaced one after releasing it: its thread may be
delivering an event, which takes the manager lock."""
existing = self._streams.get(account_id)
# Closing the previous socket first avoids duplicate Twitch
# subscriptions racing the replacement welcome/subscription call.
if existing is not None:
self._stop(existing, join=False)
stream = _Stream(account_id, broadcaster_id, types)
stream.thread = threading.Thread(target=self._run, args=(stream,), daemon=True,
name="twitch-eventsub")
self._streams[account_id] = stream
stream.thread.start()
return stream, existing
def release(self, account_id: str, consumer: str) -> None:
"""One consumer is done. Stop or narrow the listener to what is left."""
stopped = None
with self._lock:
consumers = self._consumers.get(account_id)
if not consumers or consumers.pop(consumer, None) is None:
return
stream = self._streams.get(account_id)
types = self._union(consumers)
if not consumers or stream is None or not types:
self._consumers.pop(account_id, None)
stopped = self._streams.pop(account_id, None)
elif not self._closed and set(stream.types) - set(types):
_, stopped = self._launch(account_id, stream.broadcaster_id,
tuple(t for t in stream.types if t in types))
if stopped is not None:
self._stop(stopped)
@staticmethod
def _status(stream: _Stream) -> dict:
with stream.lock:
result = {"status": stream.status, "broadcaster_id": stream.broadcaster_id,
"event_types": list(stream.types)}
if stream.error:
result["error"] = stream.error
return result
def events(self, account_id: str, *, broadcaster_id: str | None = None,
after: int = 0, limit: int = 100) -> dict:
if not isinstance(after, int) or isinstance(after, bool) or after < 0:
raise AccountError("invalid_request", "The event cursor must be a nonnegative integer.")
if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 100:
raise AccountError("invalid_request", "Request between one and 100 events.")
account = next((a for a in self.service.accounts() if a.get("id") == account_id), None)
if account is None or account.get("status") != "connected":
self.stop(account_id)
return {"status": "reconnect_required", "events": [], "cursor": after}
with self._lock:
stream = self._streams.get(account_id)
if stream is None or (broadcaster_id is not None and broadcaster_id != stream.broadcaster_id):
return {"status": "stopped", "events": [], "cursor": after}
with stream.lock:
if "user:read:chat" not in account.get("scopes", []):
stream.buffer = deque((e for e in stream.buffer if e["type"] != "channel.chat.message"),
maxlen=MAX_EVENTS)
available = [dict(e, data=dict(e["data"])) for e in stream.buffer if e["sequence"] > after]
result = self._status(stream)
result["events"] = available[:limit]
result["cursor"] = result["events"][-1]["sequence"] if result["events"] else max(after, stream.cursor)
result["truncated"] = bool(stream.buffer and after < stream.buffer[0]["sequence"] - 1)
return result
def stop(self, account_id: str) -> None:
with self._lock:
self._consumers.pop(account_id, None)
stream = self._streams.pop(account_id, None)
if stream is not None:
self._stop(stream)
def close(self) -> None:
with self._lock:
self._closed = True
streams = list(self._streams.values())
self._streams.clear()
self._consumers.clear()
self._sinks.clear()
for stream in streams:
self._stop(stream)
@staticmethod
def _close_socket(socket) -> None:
try:
socket.close()
except Exception:
# Closing is best effort; a socket error must not mask the reason
# the listener is stopping or reconnecting.
pass
def _stop(self, stream: _Stream, *, join: bool = True) -> None:
stream.stopped.set()
with stream.lock:
sockets = list(stream.sockets)
stream.buffer.clear()
stream.status = "stopped"
for socket in sockets:
self._close_socket(socket)
if join:
self._join(stream)
@staticmethod
def _join(stream: _Stream | None) -> None:
if stream is not None and stream.thread is not None and stream.thread is not threading.current_thread():
stream.thread.join(timeout=3)
@staticmethod
def _packet(socket, timeout: float) -> dict:
raw = socket.recv(timeout=max(0.001, timeout))
if not isinstance(raw, (str, bytes)) or len(raw) > 1_048_576:
raise ValueError("Invalid EventSub message")
packet = json.loads(raw)
if not isinstance(packet, dict) or not isinstance(packet.get("metadata"), dict):
raise ValueError("Invalid EventSub message")
if not isinstance(packet.get("payload"), dict):
raise ValueError("Invalid EventSub message")
return packet
def _welcome(self, socket) -> tuple[str, float]:
packet = self._packet(socket, 10)
session = packet["payload"].get("session")
if packet["metadata"].get("message_type") != "session_welcome" or not isinstance(session, dict):
raise ValueError("EventSub did not welcome the session")
session_id = session.get("id")
timeout = session.get("keepalive_timeout_seconds")
if (not isinstance(session_id, str) or not 1 <= len(session_id) <= 512
or not isinstance(timeout, (int, float)) or isinstance(timeout, bool)
or not 1 <= timeout <= 600):
raise ValueError("Invalid EventSub session")
return session_id, float(timeout)
def _run(self, stream: _Stream) -> None:
socket = None
pending_socket = None
url, transferring = SOCKET_URL, False
failures = 0
try:
while not stream.stopped.is_set():
try:
pending_socket = self._connect(url)
with stream.lock:
stream.sockets.append(pending_socket)
if stream.stopped.is_set():
break
session_id, timeout = self._welcome(pending_socket)
subscribed = not transferring
if subscribed:
self.service.perform(stream.account_id, "subscribe_events",
broadcaster_id=stream.broadcaster_id,
event_types=list(stream.types), session_id=session_id)
# A Twitch-directed transfer keeps the old socket alive until
# the replacement welcome; Twitch transfers subscriptions.
if socket is not None:
self._close_socket(socket)
with stream.lock:
stream.sockets.remove(socket)
socket, pending_socket = pending_socket, None
with stream.lock:
stream.status, stream.error = "connected", ""
failures = 0
if subscribed:
self._announce(stream)
url = self._listen(stream, socket, timeout)
transferring = True
with stream.lock:
stream.status = "reconnecting"
except _Revoked:
with stream.lock:
stream.status = "reconnect_required"
stream.error = "Twitch revoked this subscription. Reconnect your account."
break
except AccountError as error:
if error.code in {"reconnect_required", "invalid_token", "account_not_found",
"missing_scope", "permission_denied", "provider_not_configured"}:
with stream.lock:
stream.status = "reconnect_required"
stream.error = "Reconnect Twitch with permission for these events."
break
self._discard_sockets(stream)
socket = pending_socket = None
url, transferring = SOCKET_URL, False
with stream.lock:
stream.status, stream.error = "reconnecting", "Twitch events are temporarily unavailable."
failures += 1
stream.stopped.wait(min(30, 2 ** min(failures - 1, 5)))
except Exception:
# Network exceptions may include sensitive upstream text.
self._discard_sockets(stream)
socket = pending_socket = None
url, transferring = SOCKET_URL, False
with stream.lock:
stream.status, stream.error = "reconnecting", "Twitch events are temporarily unavailable."
failures += 1
stream.stopped.wait(min(30, 2 ** min(failures - 1, 5)))
finally:
self._discard_sockets(stream)
def _discard_sockets(self, stream: _Stream) -> None:
with stream.lock:
sockets = stream.sockets[:]
stream.sockets.clear()
for socket in sockets:
self._close_socket(socket)
def _listen(self, stream: _Stream, socket, timeout: float) -> str:
deadline = self._clock() + timeout
while not stream.stopped.is_set():
remaining = deadline - self._clock()
if remaining <= 0:
raise TimeoutError("EventSub keepalive expired")
packet = self._packet(socket, remaining)
deadline = self._clock() + timeout
kind = packet["metadata"].get("message_type")
if kind == "session_reconnect":
session = packet["payload"].get("session") or {}
return _reconnect_url(session.get("reconnect_url"))
if kind == "revocation":
raise _Revoked()
if kind == "notification":
self._notification(stream, packet)
elif kind != "session_keepalive":
raise ValueError("Unexpected EventSub message")
raise ConnectionError("EventSub stopped")
def _announce(self, stream: _Stream) -> None:
"""Tell sinks a fresh subscription began; events in the gap are lost."""
event = {"id": "session-" + secrets.token_hex(16), "type": "session.subscribed",
"broadcaster_id": stream.broadcaster_id,
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "data": {}}
self._deliver(stream, event)
def _deliver(self, stream: _Stream, event: dict) -> None:
with self._lock:
sinks = list(self._sinks)
for sink in sinks:
try:
sink(stream.account_id, dict(event, data=dict(event["data"])))
except Exception:
# A failed optional recorder must not tear down subscriptions
# or put chat/provider data in the engine's exception log.
continue
def _notification(self, stream: _Stream, packet: dict) -> None:
subscription = packet["payload"].get("subscription")
event = packet["payload"].get("event")
if not isinstance(subscription, dict) or not isinstance(event, dict):
raise ValueError("Invalid EventSub notification")
kind = subscription.get("type")
condition = subscription.get("condition") or {}
if (kind not in stream.types or not isinstance(condition, dict)
or str(condition.get("broadcaster_user_id")) != stream.broadcaster_id
or str(event.get("broadcaster_user_id")) != stream.broadcaster_id):
return
message_id = _text(packet["metadata"].get("message_id"), 512)
if not message_id:
return
fields = ("broadcaster_user_id", "broadcaster_user_login", "broadcaster_user_name")
data = {key: _text(event.get(key)) for key in fields}
if kind == "stream.online":
for key in ("id", "type", "started_at"):
data[key] = _text(event.get(key))
elif kind == "channel.update":
for key in ("title", "language", "category_id", "category_name"):
data[key] = _text(event.get(key))
elif kind == "channel.chat.message":
for key in ("chatter_user_id", "chatter_user_login", "chatter_user_name", "message_id"):
data[key] = _text(event.get(key))
message = event.get("message")
data["text"] = _text(message.get("text") if isinstance(message, dict) else "", 2000)
with stream.lock:
if stream.stopped.is_set() or message_id in stream.seen:
return
stream.seen.append(message_id)
# A restarted subscription must not reuse a cursor retained by a
# plugin. A separate lock keeps channel cursors independent and
# avoids taking the manager lock while the stream is locked.
with self._sequence_lock:
stream.cursor = self._sequences.get(stream.account_id, 0) + 1
self._sequences[stream.account_id] = stream.cursor
stream.buffer.append({"sequence": stream.cursor, "id": message_id, "type": kind,
"broadcaster_id": stream.broadcaster_id,
"timestamp": _text(packet["metadata"].get("message_timestamp"), 64),
"data": data})
event = dict(stream.buffer[-1], data=dict(data))
self._deliver(stream, event)