warpedpinball.transports.http

  1"""HTTP transport for Vector boards (plain HTTP on port 80)."""
  2
  3from __future__ import annotations
  4
  5import time
  6from typing import Any, Iterator, Optional
  7from urllib.parse import urlsplit
  8
  9import requests
 10
 11from .. import auth
 12from ..exceptions import (
 13    AuthenticationError,
 14    AuthenticationRequiredError,
 15    DeviceTimeoutError,
 16    DeviceUnreachableError,
 17    RateLimitedError,
 18    TransportError,
 19)
 20from . import Transport, parse_body, raise_for_status, serialize_body
 21
 22DEFAULT_TIMEOUT = 10.0
 23
 24
 25def _is_lan_address(host: str) -> bool:
 26    """True for private/link-local/loopback IPs (boards always live on these)."""
 27    import ipaddress
 28
 29    try:
 30        ip = ipaddress.ip_address(host)
 31    except ValueError:
 32        return False  # a hostname; leave proxy behavior alone
 33    return ip.is_private or ip.is_link_local or ip.is_loopback
 34#: Retries when the device says "429 Too many challenges" (expired challenges
 35#: are purged on each challenge request, so a short sleep usually clears it).
 36CHALLENGE_RETRIES = 3
 37CHALLENGE_RETRY_SLEEP = 1.0
 38
 39
 40class HttpTransport(Transport):
 41    """Talks to a Vector at ``http://<host>`` and handles HMAC auth signing."""
 42
 43    requires_password = True
 44
 45    def __init__(
 46        self,
 47        host: str,
 48        password: Optional[str] = None,
 49        timeout: float = DEFAULT_TIMEOUT,
 50        session: Optional[requests.Session] = None,
 51    ):
 52        host = host.rstrip("/")
 53        if not host.startswith("http://") and not host.startswith("https://"):
 54            host = "http://" + host
 55        self.base_url = host
 56        #: Bare host[:port] for error messages (no scheme/path noise).
 57        self._host_label = urlsplit(self.base_url).netloc or self.base_url
 58        self.password = password
 59        self.timeout = timeout
 60        if session is None:
 61            session = requests.Session()
 62            if _is_lan_address(urlsplit(self.base_url).hostname or ""):
 63                # Never route LAN traffic through a system/environment proxy.
 64                # Windows machines often have one configured (corporate, VPN,
 65                # WPAD auto-detect), and requests would otherwise send this
 66                # board's traffic to a proxy that cannot reach it.
 67                session.trust_env = False
 68        self._session = session
 69
 70    @property
 71    def description(self) -> str:
 72        return self.base_url
 73
 74    def close(self) -> None:
 75        self._session.close()
 76
 77    # -- internals ---------------------------------------------------------
 78
 79    def _wrap_request_exc(
 80        self, exc: requests.RequestException, path: str
 81    ) -> TransportError:
 82        """Turn a raw requests/urllib3 error into a clean, typed TransportError.
 83
 84        Timeouts and connection failures are the common, user-facing cases and
 85        get their own friendly subclasses; anything else keeps a compact
 86        message. The original exception is preserved on ``.cause`` (and these
 87        are raised ``from None`` so an uncaught error prints one short
 88        traceback, not the full urllib3/requests stack)."""
 89        if isinstance(exc, requests.exceptions.Timeout):
 90            return DeviceTimeoutError(self._host_label, timeout=self.timeout, cause=exc)
 91        if isinstance(exc, requests.exceptions.ConnectionError):
 92            return DeviceUnreachableError(self._host_label, cause=exc)
 93        return TransportError(f"Request to {self._host_label}{path} failed: {exc}")
 94
 95    def _fetch_challenge(self) -> str:
 96        """Fetch a fresh single-use challenge (never cached)."""
 97        for attempt in range(CHALLENGE_RETRIES + 1):
 98            try:
 99                resp = self._session.get(
100                    self.base_url + auth.CHALLENGE_PATH, timeout=self.timeout
101                )
102            except requests.RequestException as exc:
103                raise self._wrap_request_exc(exc, auth.CHALLENGE_PATH) from None
104            if resp.status_code == 429:
105                if attempt < CHALLENGE_RETRIES:
106                    time.sleep(CHALLENGE_RETRY_SLEEP)
107                    continue
108                raise RateLimitedError(
109                    "Device has too many outstanding auth challenges; "
110                    "retry after a short wait"
111                )
112            raise_for_status(resp.status_code, resp.text, auth.CHALLENGE_PATH)
113            data = resp.json()
114            return data["challenge"]
115        # Defensive: the loop above always returns or raises within the retry
116        # budget, so this is unreachable in practice.
117        raise RateLimitedError(  # pragma: no cover
118            "Device has too many outstanding auth challenges"
119        )
120
121    def _send(
122        self,
123        path: str,
124        body_str: Optional[str],
125        authenticated: bool,
126        stream: bool = False,
127    ) -> requests.Response:
128        headers = {}
129        if body_str is not None:
130            headers["Content-Type"] = "application/json"
131        if authenticated:
132            # An empty string is a valid (empty) password and is signed as
133            # such; only ``None`` means no password has been set.
134            if self.password is None:
135                raise AuthenticationRequiredError(
136                    f"Route {path!r} requires authentication but no password is set"
137                )
138            headers.update(
139                auth.auth_headers(self.password, self._fetch_challenge(), path, body_str or "")
140            )
141
142        method = "GET" if body_str is None else "POST"
143        url = self.base_url + path
144        retries = 1 if (method == "GET" and not authenticated) else 0
145        while True:
146            try:
147                return self._session.request(
148                    method,
149                    url,
150                    data=body_str.encode("utf-8") if body_str is not None else None,
151                    headers=headers,
152                    timeout=self.timeout,
153                    stream=stream,
154                )
155            except requests.RequestException as exc:
156                if retries > 0:
157                    retries -= 1
158                    continue
159                raise self._wrap_request_exc(exc, path) from None
160
161    def _request_with_auth_retry(
162        self, path: str, body_str: Optional[str], authenticated: bool, stream: bool
163    ) -> requests.Response:
164        resp = self._send(path, body_str, authenticated, stream=stream)
165        if authenticated and resp.status_code == 401:
166            # Challenges are single-use and expire after 60 s; a stale/consumed
167            # challenge earns exactly one retry. "Bad Credentials" never does.
168            try:
169                raise_for_status(resp.status_code, resp.text, path)
170            except AuthenticationError as exc:
171                if not auth.is_retryable_auth_failure(exc.reason):
172                    raise
173            resp = self._send(path, body_str, authenticated, stream=stream)
174        return resp
175
176    # -- Transport interface -------------------------------------------------
177
178    def request(self, path: str, body: Any = None, authenticated: bool = False) -> Any:
179        body_str = serialize_body(body)
180        resp = self._request_with_auth_retry(path, body_str, authenticated, stream=False)
181        raise_for_status(resp.status_code, resp.text, path)
182        return parse_body(resp.text)
183
184    def stream(
185        self, path: str, body: Any = None, authenticated: bool = False
186    ) -> Iterator[bytes]:
187        body_str = serialize_body(body)
188        resp = self._request_with_auth_retry(path, body_str, authenticated, stream=True)
189        if resp.status_code >= 300:
190            text = resp.text  # small error body; safe to read
191            raise_for_status(resp.status_code, text, path)
192
193        def _iter() -> Iterator[bytes]:
194            try:
195                for chunk in resp.iter_content(chunk_size=4096):
196                    if chunk:
197                        yield chunk
198            except requests.RequestException as exc:
199                raise self._wrap_request_exc(exc, path) from None
200            finally:
201                resp.close()
202
203        return _iter()