src/genro_asgi/channel/frame.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 """Versioned channel frames with bounded JSON routing information and opaque bytes.15 16 The frame layer never interprets payload bytes. Socket and in-process streams17 share :class:`FrameCodec`, including strict version, length and JSON checks.18 """19 20 from __future__ import annotations21 import asyncio22 import copy23 import json24 import math25 import logging26 import time27 import struct28 import uuid29 from typing import Any30 31 from ..transport_limits import (32 DEFAULT_MAX_FRAME_SIZE, DEFAULT_WARN_FRAME_SIZE, FrameTooLarge,33 frame_max_size, integer_setting,34 )35 36 CHANNEL_MAGIC = b"GNRF"37 CHANNEL_VERSION = 138 HEADER = struct.Struct("!4sBII")39 HEADER_SIZE = HEADER.size40 MAX_FRAME_SIZE = DEFAULT_MAX_FRAME_SIZE41 _logger = logging.getLogger(__name__)42 REGISTER_METHOD = "REGISTER"43 REGISTER_PATH = "/register"44 RESERVED_INFO_KEYS = frozenset({"id", "method", "path"})45 MAX_ROUTING_STRING = 409646 ALLOWED_METHODS = frozenset({"REGISTER", "POST", "CALL", "REPLY", "EVENT"})47 __all__ = [48 "CHANNEL_MAGIC",49 "CHANNEL_VERSION",50 "HEADER_SIZE",51 "MAX_FRAME_SIZE",52 "FrameTooLarge",53 "REGISTER_METHOD",54 "REGISTER_PATH",55 "Frame",56 "FrameCodec",57 "FrameStream",58 ]59 60 61 class Frame:62 """An immutable routing record and opaque byte payload."""63 64 __slots__ = ("_id", "_method", "_path", "_info", "_payload")65 66 def __init__(67 self,68 *,69 id: str | None = None,70 method: str = "POST",71 path: str = "/",72 info: dict[str, Any] | None = None,73 payload: bytes = b"",74 ) -> None:75 codec = FrameCodec()76 self._id = codec.validate_routing_string("id", str(uuid.uuid4()) if id is None else id)77 self._method = codec.validate_routing_string("method", method)78 if self._method not in ALLOWED_METHODS:79 raise ValueError(f"unsupported frame method {self._method!r}")80 self._path = codec.validate_routing_string("path", path)81 if info is None:82 info = {}83 if not isinstance(info, dict):84 raise TypeError("frame info must be a JSON object")85 collision = RESERVED_INFO_KEYS.intersection(info)86 if collision:87 raise ValueError(f"frame info contains reserved keys: {', '.join(sorted(collision))}")88 try:89 codec.validate_json(info)90 except RecursionError as exc:91 raise ValueError("frame info nesting exceeds decoder capacity") from exc92 if not isinstance(payload, bytes):93 raise TypeError("frame payload must be bytes")94 self._info = copy.deepcopy(info)95 self._payload = payload96 97 @property98 def id(self) -> str:99 return self._id100 101 @property102 def method(self) -> str:103 return self._method104 105 @property106 def path(self) -> str:107 return self._path108 109 @property110 def info(self) -> dict[str, Any]:111 return copy.deepcopy(self._info)112 113 @property114 def payload(self) -> bytes:115 return self._payload116 117 def encode(self) -> bytes:118 return FrameCodec().encode(self)119 120 def __repr__(self) -> str:121 return f"<Frame {self.method} {self.path} id={self.id}>"122 123 124 class FrameCodec:125 """Encode, decode and validate one version of the frame protocol."""126 127 def __init__(128 self, *, max_size: int | None = None, warn_size: int | None = None,129 warning_interval: int | None = None,130 ) -> None:131 self.max_size = frame_max_size() if max_size is None else max_size132 self.warn_size = (integer_setting("GNR_ASGI_FRAME_WARN_BYTES", DEFAULT_WARN_FRAME_SIZE)133 if warn_size is None else warn_size)134 self.warning_interval = (integer_setting("GNR_ASGI_FRAME_WARN_INTERVAL_SECONDS", 60)135 if warning_interval is None else warning_interval)136 for name, value, minimum in (("max_size", self.max_size, 1),137 ("warn_size", self.warn_size, 0),138 ("warning_interval", self.warning_interval, 0)):139 if isinstance(value, bool) or not isinstance(value, int) or value < minimum:140 raise ValueError(f"{name} must be an integer >= {minimum}")141 if self.max_size > 2**32 - 1:142 raise ValueError("max_size must fit an unsigned 32-bit integer")143 self._last_warning: float | None = None144 145 def warn_large_frame(self, frame: Frame, size: int, direction: str) -> None:146 if not self.warn_size or size <= self.warn_size:147 return148 now = time.monotonic()149 if self._last_warning is not None and now - self._last_warning < self.warning_interval:150 return151 self._last_warning = now152 snapshot = frame.info.get("worker_snapshot")153 worker = snapshot.get("name") if isinstance(snapshot, dict) else None154 _logger.warning(155 "Large transport frame: bytes=%s threshold=%s direction=%s method=%s path=%s worker=%s",156 size, self.warn_size, direction, frame.method, frame.path, worker,157 )158 159 def reject_constant(self, value: str) -> None:160 raise ValueError(f"non-finite JSON number {value!r}")161 162 def object_pairs(self, pairs: list[tuple[str, Any]]) -> dict[str, Any]:163 result: dict[str, Any] = {}164 for key, value in pairs:165 if key in result:166 raise ValueError(f"duplicate JSON key {key!r}")167 result[key] = value168 return result169 170 def validate_json(self, value: Any) -> None:171 if value is None or isinstance(value, (str, bool, int)):172 return173 if isinstance(value, float):174 if not math.isfinite(value):175 raise ValueError("non-finite numbers are not valid frame info")176 return177 if isinstance(value, list):178 for item in value:179 self.validate_json(item)180 return181 if isinstance(value, dict):182 for key, item in value.items():183 if not isinstance(key, str):184 raise TypeError("frame info keys must be strings")185 self.validate_json(item)186 return187 raise TypeError(f"frame info value {type(value).__name__} is not JSON-compatible")188 189 def validate_routing_string(self, name: str, value: Any) -> str:190 if not isinstance(value, str) or not value or len(value) > MAX_ROUTING_STRING:191 raise ValueError(192 f"frame {name} must be a nonempty string of at most {MAX_ROUTING_STRING} characters"193 )194 return value195 196 def get_header_lengths(self, header: bytes) -> tuple[int, int]:197 if len(header) != HEADER_SIZE:198 raise ValueError("truncated frame header")199 magic, version, info_length, payload_length = HEADER.unpack(header)200 if magic != CHANNEL_MAGIC:201 raise ValueError("invalid frame magic")202 if version != CHANNEL_VERSION:203 raise ValueError(f"unsupported frame version {version}")204 if info_length + payload_length > self.max_size:205 raise FrameTooLarge(info_length + payload_length, self.max_size)206 return info_length, payload_length207 208 def encode(self, frame: Frame) -> bytes:209 record = {"id": frame.id, "method": frame.method, "path": frame.path, **frame.info}210 self.validate_json(record)211 try:212 info = json.dumps(record, allow_nan=False, separators=(",", ":")).encode()213 except (TypeError, ValueError, RecursionError) as exc:214 raise ValueError(f"invalid frame info: {exc}") from exc215 size = len(info) + len(frame.payload)216 if size > self.max_size:217 raise FrameTooLarge(size, self.max_size)218 header = HEADER.pack(CHANNEL_MAGIC, CHANNEL_VERSION, len(info), len(frame.payload))219 self.warn_large_frame(frame, size, "send")220 return header + info + frame.payload221 222 def get_frame(self, wire: bytes) -> Frame:223 if len(wire) < HEADER_SIZE:224 raise ValueError("truncated frame header")225 ilength, plength = self.get_header_lengths(wire[:HEADER_SIZE])226 expected = HEADER_SIZE + ilength + plength227 if len(wire) != expected:228 raise ValueError(229 f"truncated or overlong frame: expected {expected} bytes, got {len(wire)}"230 )231 try:232 record = json.loads(233 wire[HEADER_SIZE : HEADER_SIZE + ilength],234 object_pairs_hook=self.object_pairs,235 parse_constant=self.reject_constant,236 )237 except (UnicodeDecodeError, json.JSONDecodeError, ValueError, RecursionError) as exc:238 raise ValueError(f"invalid frame info JSON: {exc}") from exc239 if not isinstance(record, dict):240 raise ValueError("frame info must be a JSON object")241 try:242 frame_id = record.pop("id")243 method = record.pop("method")244 path = record.pop("path")245 except KeyError as exc:246 raise ValueError(f"frame info missing {exc.args[0]!r}") from exc247 frame = Frame(248 id=frame_id,249 method=method,250 path=path,251 info=record,252 payload=wire[HEADER_SIZE + ilength :],253 )254 self.warn_large_frame(frame, ilength + plength, "receive")255 return frame256 257 258 class FrameStream:259 def __init__(260 self,261 reader: asyncio.StreamReader,262 writer: asyncio.StreamWriter,263 *,264 max_size: int | None = None,265 ) -> None:266 self.reader = reader267 self.writer = writer268 self.codec = FrameCodec(max_size=max_size)269 self.max_size = self.codec.max_size270 self._write_lock = asyncio.Lock()271 272 async def read(self) -> Frame | None:273 try:274 header = await self.reader.readexactly(HEADER_SIZE)275 except ConnectionResetError:276 return None277 except asyncio.IncompleteReadError as exc:278 if not exc.partial:279 return None280 raise ValueError("truncated frame header") from exc281 ilength, plength = self.codec.get_header_lengths(header)282 try:283 body = await self.reader.readexactly(ilength + plength)284 except (asyncio.IncompleteReadError, ConnectionResetError) as exc:285 raise ValueError("truncated frame body") from exc286 return self.codec.get_frame(header + body)287 288 async def write(self, frame: Frame) -> None:289 wire = self.codec.encode(frame)290 async with self._write_lock:291 self.writer.write(wire)292 await self.writer.drain()293 294 async def close(self) -> None:295 self.writer.close()296 try:297 await self.writer.wait_closed()298 except (BrokenPipeError, ConnectionResetError):299 pass