pyrtan / src / pyrtan / transport.py
10,621 bytes · last changed 2026-10-26 in Watches and growing from a gap: trees that run when something speaks, and the branch a gap asks for drafted by rules; release 0.9.0
| 1 | """How a tree reaches the outside: the HTTP form of I/O. |
| 2 | |
| 3 | A :class:`Transport` turns a request into a :class:`Response`. Further forms |
| 4 | of I/O (files, shells, serial lines, queues) arrive as further transports |
| 5 | behind the same seam, each with a recorded and a recording wrapper so trees |
| 6 | over them are testable offline the same way. The real one is |
| 7 | :class:`UrllibTransport`, on the standard library. :class:`RecordedTransport` |
| 8 | replays responses from a fixture, so a tree can be tested end to end with no |
| 9 | network, and :class:`RecordingTransport` wraps a real transport to write such |
| 10 | a fixture. |
| 11 | |
| 12 | The runtime, not the transport, enforces a tree's host allow-list; a |
| 13 | transport only moves bytes. The one policy a transport does hold is |
| 14 | *egress*: an operator who hosts trees for other people can tell the real |
| 15 | transport to reach only public addresses, so a tree with ``hosts = ["*"]`` |
| 16 | cannot be pointed at the host's own network. |
| 17 | """ |
| 18 | |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | import ipaddress |
| 22 | import json |
| 23 | import socket |
| 24 | import time |
| 25 | import urllib.error |
| 26 | import urllib.parse |
| 27 | import urllib.request |
| 28 | from dataclasses import dataclass, field |
| 29 | from pathlib import Path |
| 30 | from typing import Any, Protocol |
| 31 | |
| 32 | EGRESS_POLICIES = ("any", "public") |
| 33 | |
| 34 | |
| 35 | @dataclass |
| 36 | class Response: |
| 37 | status: int |
| 38 | headers: dict[str, str] |
| 39 | body: str |
| 40 | url: str |
| 41 | elapsed_ms: int = 0 |
| 42 | |
| 43 | @property |
| 44 | def json(self) -> Any: |
| 45 | try: |
| 46 | return json.loads(self.body) |
| 47 | except (json.JSONDecodeError, TypeError): |
| 48 | return None |
| 49 | |
| 50 | def to_state(self) -> dict[str, Any]: |
| 51 | """The shape a tree sees after an ``http`` observation.""" |
| 52 | return { |
| 53 | "status": self.status, |
| 54 | "ok": 200 <= self.status < 300, |
| 55 | "headers": self.headers, |
| 56 | "body": self.body, |
| 57 | "json": self.json, |
| 58 | "url": self.url, |
| 59 | "elapsed_ms": self.elapsed_ms, |
| 60 | } |
| 61 | |
| 62 | |
| 63 | class Transport(Protocol): |
| 64 | def request(self, method: str, url: str, headers: dict[str, str] | None = None, body: str | None = None, timeout: float = 10.0) -> Response: ... |
| 65 | |
| 66 | |
| 67 | class TransportError(RuntimeError): |
| 68 | pass |
| 69 | |
| 70 | |
| 71 | class EgressRefused(TransportError): |
| 72 | """The egress policy refused a URL before any connection was made.""" |
| 73 | |
| 74 | |
| 75 | def check_egress(url: str, policy: str = "any", *, resolve=socket.getaddrinfo) -> None: |
| 76 | """Raise :class:`EgressRefused` if ``url`` is not allowed under ``policy``. |
| 77 | |
| 78 | ``any`` allows every http(s) URL. ``public`` also refuses a host that |
| 79 | resolves to a loopback, private, link-local, multicast, reserved or |
| 80 | unspecified address: the operator's LAN, the container network, and the |
| 81 | cloud metadata endpoint. The check resolves the name here and the client |
| 82 | resolves it again to connect, so a name that changes answers between the |
| 83 | two can slip past; a host that must be sealed should also firewall. |
| 84 | """ |
| 85 | if policy not in EGRESS_POLICIES: |
| 86 | raise ValueError(f"egress policy must be one of {EGRESS_POLICIES}, got {policy!r}") |
| 87 | parts = urllib.parse.urlsplit(url) |
| 88 | if parts.scheme not in ("http", "https"): |
| 89 | raise EgressRefused(f"{url}: only http and https are allowed") |
| 90 | host = parts.hostname |
| 91 | if not host: |
| 92 | raise EgressRefused(f"{url}: no host") |
| 93 | if policy == "any": |
| 94 | return |
| 95 | if host.lower() in ("localhost",) or host.lower().endswith(".localhost"): |
| 96 | raise EgressRefused(f"{url}: {host} is not a public address") |
| 97 | try: |
| 98 | addresses = {info[4][0] for info in resolve(host, parts.port or (443 if parts.scheme == "https" else 80), proto=socket.IPPROTO_TCP)} |
| 99 | except (socket.gaierror, OSError) as e: |
| 100 | raise EgressRefused(f"{url}: cannot resolve {host}: {e}") from e |
| 101 | for text in addresses: |
| 102 | try: |
| 103 | address = ipaddress.ip_address(text.split("%", 1)[0]) |
| 104 | except ValueError: |
| 105 | raise EgressRefused(f"{url}: {host} resolved to {text!r}, which is not an address") |
| 106 | if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped is not None: |
| 107 | address = address.ipv4_mapped |
| 108 | if not address.is_global or address.is_multicast: |
| 109 | raise EgressRefused(f"{url}: {host} resolves to {address}, which is not a public address") |
| 110 | |
| 111 | |
| 112 | class _EgressRedirectHandler(urllib.request.HTTPRedirectHandler): |
| 113 | """Apply the egress policy to every redirect target as well as the first URL.""" |
| 114 | |
| 115 | def __init__(self, policy: str) -> None: |
| 116 | super().__init__() |
| 117 | self.policy = policy |
| 118 | |
| 119 | def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[override] |
| 120 | check_egress(newurl, self.policy) |
| 121 | return super().redirect_request(req, fp, code, msg, headers, newurl) |
| 122 | |
| 123 | |
| 124 | class UrllibTransport: |
| 125 | """The standard library's HTTP client, with a fixed user agent and no |
| 126 | cookie jar: every request is stateless unless a tree sends state itself. |
| 127 | |
| 128 | ``egress`` is ``any`` (the default: every host a tree names) or |
| 129 | ``public`` (refuse private, loopback, link-local and metadata addresses, |
| 130 | on the first URL and on every redirect).""" |
| 131 | |
| 132 | def __init__(self, user_agent: str = "pyrtan/0.9 (+https://github.com/b-i-t-8/pyrtan)", egress: str = "any") -> None: |
| 133 | if egress not in EGRESS_POLICIES: |
| 134 | raise ValueError(f"egress policy must be one of {EGRESS_POLICIES}, got {egress!r}") |
| 135 | self.user_agent = user_agent |
| 136 | self.egress = egress |
| 137 | self._opener = urllib.request.build_opener(_EgressRedirectHandler(egress)) if egress != "any" else None |
| 138 | |
| 139 | def request(self, method: str, url: str, headers: dict[str, str] | None = None, body: str | None = None, timeout: float = 10.0) -> Response: |
| 140 | check_egress(url, self.egress) |
| 141 | hdrs = {"User-Agent": self.user_agent, "Accept": "application/json, text/html;q=0.9, */*;q=0.8"} |
| 142 | hdrs.update(headers or {}) |
| 143 | data = body.encode("utf-8") if body is not None else None |
| 144 | req = urllib.request.Request(url, data=data, headers=hdrs, method=method.upper()) |
| 145 | started = time.monotonic() |
| 146 | opener = self._opener.open if self._opener is not None else urllib.request.urlopen |
| 147 | try: |
| 148 | with opener(req, timeout=timeout) as resp: |
| 149 | raw = resp.read() |
| 150 | return Response(resp.status, {k.lower(): v for k, v in resp.headers.items()}, _decode(raw, resp.headers.get_content_charset()), resp.geturl(), int((time.monotonic() - started) * 1000)) |
| 151 | except urllib.error.HTTPError as e: |
| 152 | raw = e.read() if hasattr(e, "read") else b"" |
| 153 | return Response(e.code, {k.lower(): v for k, v in (e.headers or {}).items()}, _decode(raw, None), url, int((time.monotonic() - started) * 1000)) |
| 154 | except (urllib.error.URLError, TimeoutError, OSError) as e: |
| 155 | raise TransportError(f"{method} {url}: {e}") from e |
| 156 | |
| 157 | |
| 158 | def _decode(raw: bytes, charset: str | None) -> str: |
| 159 | for enc in (charset, "utf-8", "latin-1"): |
| 160 | if not enc: |
| 161 | continue |
| 162 | try: |
| 163 | return raw.decode(enc) |
| 164 | except (UnicodeDecodeError, LookupError): |
| 165 | continue |
| 166 | return raw.decode("utf-8", errors="replace") |
| 167 | |
| 168 | |
| 169 | def _key(method: str, url: str) -> str: |
| 170 | return f"{method.upper()} {url}" |
| 171 | |
| 172 | |
| 173 | @dataclass |
| 174 | class RecordedTransport: |
| 175 | """Replay responses from a fixture: ``{"GET https://...": {"status": 200, |
| 176 | "body": "..."}}``. A ``body`` may be a JSON value instead of a string, in |
| 177 | which case it is serialised. Unknown requests raise, so a test fails |
| 178 | loudly when a tree reaches somewhere its fixture did not foresee.""" |
| 179 | |
| 180 | responses: dict[str, Any] = field(default_factory=dict) |
| 181 | calls: list[tuple[str, str]] = field(default_factory=list) |
| 182 | |
| 183 | @classmethod |
| 184 | def from_file(cls, path: str | Path) -> "RecordedTransport": |
| 185 | return cls(json.loads(Path(path).read_text(encoding="utf-8"))) |
| 186 | |
| 187 | def add(self, method: str, url: str, status: int = 200, body: Any = "", headers: dict[str, str] | None = None) -> None: |
| 188 | self.responses[_key(method, url)] = {"status": status, "body": body, "headers": headers or {}} |
| 189 | |
| 190 | def request(self, method: str, url: str, headers: dict[str, str] | None = None, body: str | None = None, timeout: float = 10.0) -> Response: |
| 191 | self.calls.append((method.upper(), url)) |
| 192 | rec = self._take(_key(method, url), "response") |
| 193 | text = rec.get("body", "") |
| 194 | if not isinstance(text, str): |
| 195 | text = json.dumps(text) |
| 196 | return Response(int(rec.get("status", 200)), {k.lower(): v for k, v in rec.get("headers", {}).items()}, text, url, 0) |
| 197 | |
| 198 | def perform(self, source: str, spec: dict[str, Any], timeout: float = 10.0, allowed: Any = None) -> dict[str, Any]: |
| 199 | """The other sources, from the same fixture: a record under |
| 200 | :func:`pyrtan.adapters.adapter_key` is what the node observes.""" |
| 201 | from .adapters import adapter_key |
| 202 | |
| 203 | key = adapter_key(source, spec) |
| 204 | self.calls.append((source, key)) |
| 205 | return json.loads(json.dumps(self._take(key, "result"))) |
| 206 | |
| 207 | def _take(self, key: str, noun: str) -> dict[str, Any]: |
| 208 | rec = self.responses.get(key) |
| 209 | if rec is None: |
| 210 | raise TransportError(f"no recorded {noun} for {key}") |
| 211 | if isinstance(rec, list): |
| 212 | # A sequence: consume in order, repeating the last. |
| 213 | rec = rec.pop(0) if len(rec) > 1 else rec[0] |
| 214 | if "error" in rec: |
| 215 | raise TransportError(str(rec["error"])) |
| 216 | return rec |
| 217 | |
| 218 | |
| 219 | @dataclass |
| 220 | class RecordingTransport: |
| 221 | """Pass through to ``inner`` and keep what came back, to be saved as a |
| 222 | fixture for :class:`RecordedTransport`.""" |
| 223 | |
| 224 | inner: Transport |
| 225 | adapters: Any = None |
| 226 | recorded: dict[str, Any] = field(default_factory=dict) |
| 227 | |
| 228 | def request(self, method: str, url: str, headers: dict[str, str] | None = None, body: str | None = None, timeout: float = 10.0) -> Response: |
| 229 | resp = self.inner.request(method, url, headers, body, timeout) |
| 230 | self.recorded[_key(method, url)] = {"status": resp.status, "headers": resp.headers, "body": resp.body} |
| 231 | return resp |
| 232 | |
| 233 | def perform(self, source: str, spec: dict[str, Any], timeout: float = 10.0, allowed: Any = None) -> dict[str, Any]: |
| 234 | from .adapters import LocalAdapters, adapter_key |
| 235 | |
| 236 | if self.adapters is None: |
| 237 | self.adapters = LocalAdapters() |
| 238 | observed = self.adapters.perform(source, spec, timeout, allowed) |
| 239 | self.recorded[adapter_key(source, spec)] = json.loads(json.dumps(observed)) |
| 240 | return observed |
| 241 | |
| 242 | def save(self, path: str | Path) -> None: |
| 243 | Path(path).write_text(json.dumps(self.recorded, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") |