src/genro_asgi/wsx.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The WSX envelope: one message on a websocket, in either direction.16 17 A WSX message is the text ``WSX://`` followed by JSON. ``WsxEnvelope`` is that18 message as an object, and it is built two ways — from the text a socket19 delivered, or from the fields somebody is about to send::20 21 envelope = WsxEnvelope(text) # read one22 reply = WsxEnvelope(id=envelope.id, status=200, data=…) # write one23 await socket.send_text(reply.encode())24 25 The prefix is what tells a WSX message from any other text on the socket, and26 the routing fields parallel the internal channel's info, so a27 message copies into a CALL one field at a time.28 29 **A request carries ``method`` and ``path``; an answer carries ``status``.**30 Both carry ``data`` and, when they belong to a page, ``page_id``. ``id`` is31 what correlates an answer with the message it answers, and its ABSENCE is32 meaningful twice: a message with no id is an event nobody answers, and a33 message the server sends by itself never has one. ``reply_path`` is where a34 page asks to be called back when the work is done. A field nobody set does not35 reach the wire — a null there would read as a value.36 37 **The application data stays serialized while routing.** The outer JSON38 contains a TYTX string. WsxEnvelope parses only that JSON and keeps an explicit39 SerializedWsxPayload. Its data property is an opt-in consumer decoder; routing40 uses serialized_data. Value constructors and public send_message still serialize41 ordinary Python values. Application endpoints adapt XML/msgpack responses to42 browser JSON, while forwarding responses travel without application codecs.43 44 **A text that is not a WSX message raises.** So does a body that is not JSON,45 and one that is not an object. All three are the same thing to a reader — this46 text is not a message of ours — and the read loop logs and moves on.47 48 ``WsxConnection`` is one live connection speaking that protocol: ``serve()`` is49 its whole life. It gates the handshake, accepts, reads messages until the50 client leaves, and waits for what is still in flight. Every message with an51 ``id`` becomes a synthetic HTTP request with the method ``WSK``, handed to the52 application its ``path`` names through the server's own demux — so an53 application learns no new method and a websocket message travels the road a54 request travels. A message with no ``id`` is an event: served, answered by55 nothing.56 57 **The gate answers in one shape: accept, then close with a readable code.** A58 handshake on a path no application serves, or one missing the cookie its home59 application demands, closes 1008. The single exception is a hostile Origin,60 refused BEFORE the accept — there is nobody to tell, because nobody was61 admitted. The state of the server is judged higher up, in ``on_websocket``,62 above the demux and for every websocket alike: this gate never sees a63 connection the machine had already refused.64 """65 66 from __future__ import annotations67 68 import asyncio69 import json70 import logging71 from typing import Any72 73 from genro_tytx import to_tytx74 75 from .application import BaseApplication76 from .exceptions import HTTPException, WebSocketDisconnect77 from .wsx_payload import SerializedWsxPayload, WsxResponseEncoder78 from .middleware.session import SessionMiddleware79 from .types import Receive, Scope, Send80 from .websocket import WebSocket81 82 __all__ = ["WsxConnection", "WsxEnvelope"]83 84 WSX_PREFIX = "WSX://"85 86 #: The reserved first segment: what the server answers itself, never an87 #: application. ``/_wsx/ping`` is the control ping, served inline.88 WSX_ROOT = "_wsx"89 PING_PATH = f"/{WSX_ROOT}/ping"90 91 #: The command a page sends before any message of its own: the application92 #: decides — it validates the page and writes the channel on its row — and the93 #: CONNECTION binds, because it is the one holding the socket (owner, N30).94 #: Recognised on the path the demux leaves once the mount is taken off, never95 #: on the whole path: the mount is the application's, the segment is the core's.96 OPENCHANNEL_PATH = f"/{WSX_ROOT}/openchannel"97 98 #: How long the closing connection waits for the messages still in flight.99 DRAIN_TIMEOUT_SECONDS = 5.0100 101 102 class WsxEnvelope:103 """One WSX message: read from its text, or built to be sent."""104 105 def __init__(106 self,107 text: str | None = None,108 *,109 id: str | None = None,110 method: str | None = None,111 path: str | None = None,112 data: Any = None,113 page_id: str | None = None,114 reply_path: str | None = None,115 status: int | None = None,116 serialized_data: SerializedWsxPayload | None = None,117 ) -> None:118 """Initialize this instance.119 120 Args:121 text: a WSX message to read; the keywords are ignored when it is given.122 id: what correlates an answer with its message; ``None`` for an event.123 method: the request's method — ``WSK`` for a page's rpc.124 path: the request's path, which names the application and the route.125 data: the payload, as a Python value.126 page_id: the page the message belongs to, when it belongs to one.127 reply_path: where the page asks to be called back.128 status: the answer's status; ``None`` on a request.129 130 Raises:131 ValueError: ``text`` is not a WSX message, its body is not JSON, or132 that body is not an object.133 """134 if text is not None:135 fields = self.read_text(text)136 id = fields.get("id")137 method = fields.get("method")138 path = fields.get("path")139 page_id = fields.get("page_id")140 reply_path = fields.get("reply_path")141 status = fields.get("status")142 serialized_data = SerializedWsxPayload(fields.get("data"))143 for name, value in (("id", id), ("method", method), ("path", path),144 ("page_id", page_id), ("reply_path", reply_path)):145 if value is not None and (not isinstance(value, str) or len(value) > 4096):146 raise ValueError(f"invalid WSX {name}")147 self.id = id148 self.method = method149 self.path = path150 self.serialized_data = serialized_data or SerializedWsxPayload(151 to_tytx(data, "json") if data is not None else None152 )153 self.page_id = page_id154 self.reply_path = reply_path155 self.status = status156 157 @property158 def data(self) -> Any:159 """Read a value at an explicit consumer; routers use serialized_data."""160 return self.serialized_data.decode()161 162 def read_text(self, text: str) -> dict[str, Any]:163 """The JSON body of one WSX message.164 165 Args:166 text: the message as the socket delivered it.167 168 Returns:169 The body, its ``data`` still the TYTX string.170 171 Raises:172 ValueError: no ``WSX://`` prefix, a body that is not JSON, or a173 body that is not an object.174 """175 if not text.startswith(WSX_PREFIX):176 raise ValueError("this text is not a WSX:// message")177 try:178 body = json.loads(text[len(WSX_PREFIX) :])179 except ValueError as exc:180 raise ValueError(f"this WSX message carries no JSON: {exc}") from None181 if not isinstance(body, dict):182 raise ValueError(f"a WSX body must be an object, got {type(body).__name__}")183 return body184 185 def encode(self) -> str:186 """The wire text: the prefix and the JSON body.187 188 Returns:189 ``WSX://`` followed by the JSON of the fields that were set —190 ``data`` as its TYTX string, and nothing for a field left out.191 """192 body: dict[str, Any] = {}193 for name in ("id", "method", "path", "page_id", "reply_path", "status"):194 value = getattr(self, name)195 if value is not None:196 body[name] = value197 if self.serialized_data.text is not None:198 body["data"] = self.serialized_data.text199 return WSX_PREFIX + json.dumps(body)200 201 def __repr__(self) -> str:202 told = f"{self.method} {self.path}" if self.status is None else f"status {self.status}"203 return f"<WsxEnvelope {told} id={self.id}>"204 205 206 class WsxConnection:207 """One websocket connection speaking WSX, from the handshake to the end."""208 209 def __init__(self, server: Any, scope: Scope, receive: Receive, send: Send) -> None:210 """Initialize this instance.211 212 Args:213 server: the server this connection belongs to — it owns the demux, the214 identity, the request registry and the websocket registry.215 scope: the ASGI websocket scope of the handshake.216 receive: the ASGI receive callable.217 send: the ASGI send callable.218 """219 self.server = server220 self.socket = WebSocket(scope, receive, send)221 self.avatar: Any = None222 self.session: Any = None223 self.home: BaseApplication | None = None224 self._tasks: set[asyncio.Task[None]] = set()225 self._slots = asyncio.Semaphore(server.websocket_max_concurrent)226 self._logger = logging.getLogger(__name__)227 228 async def serve(self) -> None:229 """Live this connection: gate, accept, read, drain.230 231 Returns when the socket is over — the client left, or the gate turned232 it away. Registers the socket for the span of the connection and takes233 it out in the ``finally``, whatever happened.234 """235 if not await self._open_gate():236 return237 self.server.websockets.register(self.socket)238 try:239 await self._read_messages()240 finally:241 await self._drain()242 self.server.websockets.unregister(self.socket)243 244 async def _open_gate(self) -> bool:245 """Judge the handshake and accept it; ``False`` when it was turned away.246 247 Sets ``avatar``, ``session`` and ``home``.248 """249 refusal = self._origin_refusal()250 if refusal is not None:251 await self.socket.refuse(1008, refusal)252 return False253 await self.socket.accept()254 app, _ = self.server.demux(self.socket.scope)255 if not isinstance(app, BaseApplication):256 await self.socket.close(1008, "no application at this path")257 return False258 self.home = app259 cookie = app.handshake_cookie260 if cookie is not None and cookie not in self.socket.cookies:261 await self.socket.close(1008, f"connection cookie required: {cookie}")262 return False263 try:264 self.avatar = self.server.authenticate(self.socket.scope)265 except HTTPException as refused:266 await self.socket.close(1008, refused.detail or "unauthorized")267 return False268 self.session = self._read_session()269 return True270 271 def _origin_refusal(self) -> str | None:272 """Why this Origin is not admitted, or ``None`` when it is.273 274 No ``Origin`` at all passes: it is not a browser, and the gate exists275 against a page on another site, not against a client of its own. With a276 declared list the header must be in it (``*`` admits everyone);277 without one, the Origin must be the host the handshake came to.278 """279 origin = self.socket.headers.get("origin")280 if not origin:281 return None282 allowed = self.server.websocket_origins283 if allowed:284 if "*" in allowed or origin in allowed:285 return None286 return f"origin not allowed: {origin}"287 host = self.socket.headers.get("host") or ""288 if str(origin).partition("://")[2] == host:289 return None290 return f"origin not allowed: {origin}"291 292 def _read_session(self) -> Any:293 """The session of the handshake, read from the layer that owns it.294 295 The middleware chain never sees a websocket scope, so the session is296 asked of ``SessionMiddleware`` itself — a pure reading, which creates297 nothing. ``None`` when that middleware is off or no cookie arrived.298 """299 layer = self.server.get_middleware(SessionMiddleware)300 return layer.get_session(self.socket.scope) if layer is not None else None301 302 async def _read_messages(self) -> None:303 """Read until the client leaves, serving each message on a task of its own.304 305 What is not a WSX message is logged and dropped — a text of another306 shape, and a binary frame, which this protocol has no use for: the307 socket may carry other traffic, and one message nobody understands308 never ends a connection. The control ping is answered inline, outside309 the ceiling, so a connection whose slots are all busy still answers310 "are you there".311 """312 try:313 while True:314 message = await self.socket.read_message()315 text = message.get("text")316 if text is None:317 self._logger.warning("Websocket: a binary frame carries no WSX message")318 continue319 try:320 envelope = WsxEnvelope(text)321 except ValueError as broken:322 self._logger.warning("Websocket: message dropped, %s", broken)323 continue324 if envelope.path == PING_PATH:325 await self._answer(envelope, 200, "pong")326 continue327 task = asyncio.create_task(self._serve_message(envelope))328 self._tasks.add(task)329 task.add_done_callback(self._tasks.discard)330 except WebSocketDisconnect:331 return332 333 async def _drain(self) -> None:334 """Wait for the messages still in flight, then cut what is left."""335 if not self._tasks:336 return337 _, pending = await asyncio.wait(set(self._tasks), timeout=DRAIN_TIMEOUT_SECONDS)338 for task in pending:339 task.cancel()340 341 async def _serve_message(self, envelope: WsxEnvelope) -> None:342 """Serve one message as a request, and answer it when it asked to be.343 344 A message with an ``id`` is registered in the server's request registry345 — the shutdown waits for it, and it shows in the picture — and answered346 with its status. A message without one is an event: served the same347 way, answered by nothing, its failure logged.348 """349 scope = self._request_scope(envelope)350 item = self.server.requests.register(scope) if envelope.id else None351 try:352 async with self._slots:353 status, data = await self._call_application(envelope, scope)354 finally:355 if item is not None:356 item.run_cleanups()357 self.server.requests.unregister(item)358 if envelope.id is not None:359 await self._answer(envelope, status, data)360 361 async def _call_application(self, envelope: WsxEnvelope, scope: Scope) -> tuple[int, Any]:362 """Hand one message to the application its path names.363 364 Args:365 envelope: the message being served.366 scope: its synthetic http scope, built once by the caller.367 368 Returns:369 The status and the data of the answer. An ``HTTPException`` becomes370 its own status, anything else a 500 — the socket survives either.371 372 The one message this connection looks at twice is ``openchannel``: the373 application decides whether that page may speak here, and only if it374 answered 200 is the page bound to this socket.375 """376 collected: list[Any] = []377 try:378 app, target = self.server.demux(scope)379 await app(target, self._request_body(envelope), self._collector(collected))380 except HTTPException as refused:381 return refused.status, refused.detail382 except Exception as failure:383 self._logger.exception("Websocket: message %s failed", envelope.path)384 return 500, f"{type(failure).__name__}: {failure}"385 try:386 status, data = self._answer_of(387 collected, endpoint=not getattr(app, "forwards_payloads", False)388 )389 except (TypeError, ValueError) as failure:390 self._logger.exception("Websocket: invalid application reply")391 return 500, f"invalid application reply: {failure}"392 if status == 200 and target.get("path") == OPENCHANNEL_PATH and envelope.page_id:393 # The application said yes: now the page speaks on THIS socket.394 # Bound only after the answer, so a page that was refused — not395 # this connection's, or never born — is never bound at all.396 self.server.websockets.bind_page(envelope.page_id, self.socket)397 return status, data398 399 def _request_scope(self, envelope: WsxEnvelope) -> Scope:400 """The synthetic HTTP scope of one message.401 402 The method is ``WSK``, the convention an application accepts by serving403 rpc over a websocket; the path is the message's own. The handshake's404 headers, identity and session travel with EVERY message: they were405 judged once, and they hold for the connection. ``genro.page_id`` and406 ``genro.reply_path`` are there only when the message carried them.407 """408 scope: Scope = {409 **self.socket.scope,410 "type": "http",411 "method": "WSK",412 "path": envelope.path or "/",413 "headers": [(n, v) for n, v in self.socket.scope.get("headers", [])414 if n.lower() not in (b"content-type", b"content-length", b"x-tytx-transport")]415 + [(b"content-type", b"application/json"), (b"x-tytx-transport", b"json")],416 "auth": self.avatar,417 "session": self.session,418 }419 if envelope.page_id is not None:420 scope["genro.page_id"] = envelope.page_id421 if envelope.reply_path is not None:422 scope["genro.reply_path"] = envelope.reply_path423 return scope424 425 def _request_body(self, envelope: WsxEnvelope) -> Receive:426 """A ``receive`` that hands the message's data over as the request body."""427 body = (envelope.serialized_data.text or "").encode("utf-8")428 429 async def receive() -> Any:430 return {"type": "http.request", "body": body, "more_body": False}431 432 return receive433 434 def _collector(self, collected: list[Any]) -> Send:435 """A ``send`` that keeps the answer instead of writing it to a socket."""436 437 async def send(message: Any) -> None:438 if message["type"] == "http.response.body" and message.get("more_body"):439 raise RuntimeError("a streaming answer cannot travel on a websocket message")440 collected.append(message)441 442 return send443 444 def _answer_of(self, collected: list[Any], *, endpoint: bool = True) -> tuple[int, Any]:445 """The status and the data an application's answer carried.446 447 Returns:448 The status, and the body read by its content-type: one of the three449 TYTX transports comes back hydrated, anything else as the text it450 decodes to, or as the bytes it is.451 452 This is the envelope's OWN reading of an answer, the mirror of what453 ``Request.decode_body`` does with a request: the three transports are454 the same three, and nothing else is understood here — a websocket455 answer carries no form and no upload.456 """457 status = 200458 content_type = ""459 for message in collected:460 if message["type"] == "http.response.start":461 status = message["status"]462 content_type = dict(message.get("headers") or {}).get(b"content-type", b"").decode()463 body = b"".join(m.get("body", b"") for m in collected if m["type"] == "http.response.body")464 if endpoint:465 return status, WsxResponseEncoder().encode(body, content_type)466 if not body:467 return status, SerializedWsxPayload(None)468 if "json" not in content_type:469 raise ValueError("forwarded WSK reply was not adapted at its application endpoint")470 return status, SerializedWsxPayload(body.decode("utf-8"))471 472 async def _answer(self, envelope: WsxEnvelope, status: int, data: Any) -> None:473 """Write the answer to one message back onto the socket."""474 if not self.socket.connected:475 return476 await self.socket.send_text(477 WsxEnvelope(id=envelope.id, status=status,478 **({"serialized_data": data} if isinstance(data, SerializedWsxPayload)479 else {"data": data})).encode()480 )