From 9fdca204f868a9bc80f95372d53dadbe6e9c959a Mon Sep 17 00:00:00 2001 From: Edward Viaene Date: Sat, 14 Nov 2026 14:48:55 -0600 Subject: [PATCH 1/2] Security: credentials stay on their origin across redirects (SDK); loopback Host/Origin checks and --auth-token-file (MCP) - 2.13.18 Co-Authored-By: Claude Opus 5.5 --- packages/embeddedci-mcp/CHANGELOG.md | 8 ++++ packages/embeddedci-mcp/README.md | 3 +- packages/embeddedci-mcp/pyproject.toml | 2 +- packages/embeddedci-mcp/server.json | 4 +- .../src/embeddedci_mcp/__main__.py | 22 ++++++++-- packages/embeddedci/CHANGELOG.md | 6 +++ packages/embeddedci/pyproject.toml | 2 +- packages/embeddedci/src/embeddedci/_http.py | 42 +++++++++++++++++++ .../src/embeddedci/aivision/client.py | 3 +- .../src/embeddedci/aivision/cloud.py | 3 +- .../embeddedci/src/embeddedci/benchpod/ci.py | 3 +- .../src/embeddedci/benchpod/cli_login.py | 3 +- .../src/embeddedci/benchpod/cloud_auth.py | 7 ++-- .../src/embeddedci/benchpod/lease.py | 3 +- .../src/embeddedci/benchpod/server_api.py | 3 +- .../embeddedci/benchpod/transport/cloud.py | 3 +- packages/embeddedci/tests/test_cloud.py | 11 ++--- .../tests/test_http_redirect_auth.py | 32 ++++++++++++++ packages/embeddedci/tests/test_lease.py | 3 +- packages/embeddedci/tests/test_one_source.py | 3 +- .../embeddedci/tests/test_refusal_errors.py | 4 +- 21 files changed, 143 insertions(+), 27 deletions(-) create mode 100644 packages/embeddedci/src/embeddedci/_http.py create mode 100644 packages/embeddedci/tests/test_http_redirect_auth.py diff --git a/packages/embeddedci-mcp/CHANGELOG.md b/packages/embeddedci-mcp/CHANGELOG.md index 5faf4de..5f4f218 100644 --- a/packages/embeddedci-mcp/CHANGELOG.md +++ b/packages/embeddedci-mcp/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog — `embeddedci-mcp` +## 2.13.18 + +- HTTP transport: a loopback bind always checks Host and Origin (only 127.0.0.1, localhost and + [::1] on its port), whatever the `mcp` version's default, so a web page in the user's browser + cannot reach it through DNS rebinding. `--auth-token-file` reads the bearer token from a file; + a token given with `--auth-token` is visible to other users in `ps`, so prefer the file or + `EMBEDDEDCI_MCP_TOKEN`. + ## 2.13.17 - Needs `embeddedci` 2.13.17: `flash` can no longer pass Tcl to OpenOCD through the file name or diff --git a/packages/embeddedci-mcp/README.md b/packages/embeddedci-mcp/README.md index 2961c4b..a989b33 100644 --- a/packages/embeddedci-mcp/README.md +++ b/packages/embeddedci-mcp/README.md @@ -143,7 +143,8 @@ chat never blocks CI on that pod. | `--timeout` | — | `30` | per-command device timeout | | `--transport` | — | `stdio` | `stdio` or `http` | | `--host` / `--port` | — | `127.0.0.1` / `8000` | HTTP bind address | -| `--auth-token` | `EMBEDDEDCI_MCP_TOKEN` | — | require `Authorization: Bearer ` (mandatory off loopback) | +| `--auth-token` | `EMBEDDEDCI_MCP_TOKEN` | — | require `Authorization: Bearer ` (mandatory off loopback); prefer the variable or `--auth-token-file`, since a command-line token shows in `ps` | +| `--auth-token-file` | — | — | read the bearer token from this file's first line | | `--allowed-host` | — | — | Host header(s) to accept on a network bind (DNS-rebinding protection) | | `--allow-unauthenticated` | — | off | serve a network address without a token (isolated networks only) | diff --git a/packages/embeddedci-mcp/pyproject.toml b/packages/embeddedci-mcp/pyproject.toml index 4e2d662..eaa9233 100644 --- a/packages/embeddedci-mcp/pyproject.toml +++ b/packages/embeddedci-mcp/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "embeddedci-mcp" -version = "2.13.17" +version = "2.13.18" description = "MCP server that lets AI agents drive an EmbeddedCI BenchPod hardware-in-the-loop bench" readme = "README.md" requires-python = ">=3.10" diff --git a/packages/embeddedci-mcp/server.json b/packages/embeddedci-mcp/server.json index f77660a..3ccdc84 100644 --- a/packages/embeddedci-mcp/server.json +++ b/packages/embeddedci-mcp/server.json @@ -3,7 +3,7 @@ "name": "io.github.embeddedci-com/embeddedci-mcp", "title": "EmbeddedCI BenchPod", "description": "Drive a BenchPod hardware-in-the-loop bench: power, flash, UART, I2C, CAN, analog and logic.", - "version": "2.13.17", + "version": "2.13.18", "websiteUrl": "https://www.embeddedci.com/docs/benchpod-mcp", "repository": { "url": "https://github.com/embeddedci-com/embeddedci-python", @@ -15,7 +15,7 @@ "registryType": "pypi", "registryBaseUrl": "https://pypi.org", "identifier": "embeddedci-mcp", - "version": "2.13.17", + "version": "2.13.18", "runtimeHint": "uvx", "transport": { "type": "stdio" diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/__main__.py b/packages/embeddedci-mcp/src/embeddedci_mcp/__main__.py index cc3de05..88f2b2d 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/__main__.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/__main__.py @@ -52,8 +52,11 @@ def build_parser() -> argparse.ArgumentParser: http.add_argument("--host", default="127.0.0.1", help="Bind address (default 127.0.0.1).") http.add_argument("--port", type=int, default=8000, help="Bind port (default 8000).") http.add_argument("--auth-token", default=os.environ.get(TOKEN_ENV), - help=f"Require 'Authorization: Bearer ' on every request (or set {TOKEN_ENV}). " - "Mandatory when binding a non-loopback address.") + help=f"Require 'Authorization: Bearer ' on every request. Prefer {TOKEN_ENV} " + "or --auth-token-file: a token on the command line is visible to other users " + "of this computer (ps). Mandatory when binding a non-loopback address.") + http.add_argument("--auth-token-file", default=None, metavar="PATH", + help="Read the bearer token from this file (its first line).") http.add_argument("--allowed-host", action="append", default=[], metavar="HOST[:PORT]", help="Host header value to accept (repeatable), for DNS-rebinding protection on a " "non-loopback bind. Without it, Host checking is off and the token protects.") @@ -76,6 +79,12 @@ def main(argv: Optional[Sequence[str]] = None) -> None: mcp.run(transport="stdio") return + if args.auth_token_file: + with open(args.auth_token_file, encoding="utf-8") as f: + args.auth_token = (f.readline() or "").strip() or None + if not args.auth_token: + parser.error(f"--auth-token-file {args.auth_token_file} holds no token") + loopback = _is_loopback(args.host) if not loopback and not args.auth_token and not args.allow_unauthenticated: parser.error(f"--host {args.host} exposes the bench to the network: pass --auth-token (or set " @@ -85,7 +94,14 @@ def main(argv: Optional[Sequence[str]] = None) -> None: from mcp.server.transport_security import TransportSecuritySettings mcp.settings.host, mcp.settings.port = args.host, args.port - if not loopback: + if loopback: + # A loopback bind still serves any web page in the user's browser: accept only loopback Host + # and Origin headers (no DNS rebinding), set here rather than left to the mcp version's default. + hosts = [f"127.0.0.1:{args.port}", f"localhost:{args.port}", f"[::1]:{args.port}"] + mcp.settings.transport_security = TransportSecuritySettings( + enable_dns_rebinding_protection=True, allowed_hosts=hosts, + allowed_origins=[f"http://{h}" for h in hosts]) + else: # FastMCP only accepts loopback Host headers by default, which rejects every real client # of a network bind. Accept the operator's hosts, or rely on the bearer token. allowed: List[str] = list(args.allowed_host) diff --git a/packages/embeddedci/CHANGELOG.md b/packages/embeddedci/CHANGELOG.md index 604c8c9..65dc118 100644 --- a/packages/embeddedci/CHANGELOG.md +++ b/packages/embeddedci/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog — `embeddedci` +## 2.13.18 + +- Security: a redirect to another host (or from https to http) no longer carries the API key, + session token, lease or cookies along. Every SDK call to embeddedci.com and AI Vision goes + through one opener that drops them; nothing redirects today, so this guards the future. + ## 2.13.17 - Security: `flash()` puts the firmware path in braces in OpenOCD's `program` command and refuses a diff --git a/packages/embeddedci/pyproject.toml b/packages/embeddedci/pyproject.toml index 130cb8c..aac65b7 100644 --- a/packages/embeddedci/pyproject.toml +++ b/packages/embeddedci/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "embeddedci" -version = "2.13.17" +version = "2.13.18" description = "BenchPod SDK and pytest plugin for EmbeddedCI hardware-in-the-loop testing" readme = "README.md" requires-python = ">=3.10" diff --git a/packages/embeddedci/src/embeddedci/_http.py b/packages/embeddedci/src/embeddedci/_http.py new file mode 100644 index 0000000..3b60f16 --- /dev/null +++ b/packages/embeddedci/src/embeddedci/_http.py @@ -0,0 +1,42 @@ +"""HTTP for calls that carry credentials. + +``urllib``'s redirect handler copies every header to the new URL, ``Authorization`` included, +even when the redirect leaves the host: an endpoint that redirects to object storage (or a +``BENCHPOD_API_BASE`` that redirects) would hand it the API key or session token. :func:`urlopen` +drops the credentials on a redirect to another host or scheme. +""" + +from __future__ import annotations + +import urllib.parse +import urllib.request +from typing import Any, Optional + +#: Headers that hold credentials, dropped on a redirect that leaves the origin. +_SECRET_HEADERS = ("authorization", "x-benchpod-lease", "cookie") + + +class _DropAuthOnRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore[override] + new = super().redirect_request(req, fp, code, msg, headers, newurl) + if new is None: + return None + old, nxt = urllib.parse.urlsplit(req.full_url), urllib.parse.urlsplit(new.full_url) + if (old.scheme, old.hostname, old.port) != (nxt.scheme, nxt.hostname, nxt.port): + for name in list(new.headers): + if name.lower() in _SECRET_HEADERS: + del new.headers[name] + for name in list(new.unredirected_hdrs): + if name.lower() in _SECRET_HEADERS: + del new.unredirected_hdrs[name] + return new + + +_opener = urllib.request.build_opener(_DropAuthOnRedirect) + + +def urlopen(req: Any, data: Optional[bytes] = None, timeout: Optional[float] = None, **kw: Any): + """:func:`urllib.request.urlopen` that keeps credentials on their own origin across redirects.""" + if timeout is None: + return _opener.open(req, data, **kw) + return _opener.open(req, data, timeout=timeout, **kw) diff --git a/packages/embeddedci/src/embeddedci/aivision/client.py b/packages/embeddedci/src/embeddedci/aivision/client.py index f87ecc3..d5f3f30 100644 --- a/packages/embeddedci/src/embeddedci/aivision/client.py +++ b/packages/embeddedci/src/embeddedci/aivision/client.py @@ -23,6 +23,7 @@ import urllib.error import urllib.parse import urllib.request +from embeddedci import _http import uuid from contextlib import contextmanager from pathlib import Path @@ -248,7 +249,7 @@ def _open_once(self, method: str, path: str, *, body: Optional[bytes] = None, co if self._lease is not None: req.add_header("X-Lease-Id", self._lease.id) try: - return urllib.request.urlopen(req, timeout=timeout or self.timeout) + return _http.urlopen(req, timeout=timeout or self.timeout) except urllib.error.HTTPError as e: raw = e.read() try: diff --git a/packages/embeddedci/src/embeddedci/aivision/cloud.py b/packages/embeddedci/src/embeddedci/aivision/cloud.py index ed5b932..1ca0a86 100644 --- a/packages/embeddedci/src/embeddedci/aivision/cloud.py +++ b/packages/embeddedci/src/embeddedci/aivision/cloud.py @@ -21,6 +21,7 @@ from urllib.parse import quote, urlsplit from ..benchpod.cloud_auth import DEFAULT_API_BASE, DEFAULT_AUDIENCE, mint_session_token +from embeddedci import _http #: Prefix of a cloud connection string: ``embeddedci:``. CLOUD_PREFIX = "embeddedci:" @@ -133,7 +134,7 @@ def register_device(base: str, name: str, public_key: str, *, api_key: Optional[ headers={"Authorization": auth, "Content-Type": "application/json", "Accept": "application/json", "User-Agent": USER_AGENT}) try: - with urllib.request.urlopen(req, timeout=timeout) as resp: + with _http.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read() or b"{}") except urllib.error.HTTPError as e: raw = e.read() diff --git a/packages/embeddedci/src/embeddedci/benchpod/ci.py b/packages/embeddedci/src/embeddedci/benchpod/ci.py index 8b07ae5..5ab9ce4 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/ci.py +++ b/packages/embeddedci/src/embeddedci/benchpod/ci.py @@ -20,6 +20,7 @@ import urllib.error import urllib.parse import urllib.request +from embeddedci import _http import warnings from typing import Any, Dict, Iterable, List, Optional @@ -229,7 +230,7 @@ def _request( }, ) try: - with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: + with _http.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: raw = resp.read().decode("utf-8") return json.loads(raw) if raw.strip() else {} except urllib.error.HTTPError as exc: diff --git a/packages/embeddedci/src/embeddedci/benchpod/cli_login.py b/packages/embeddedci/src/embeddedci/benchpod/cli_login.py index 47af9e4..8c0534f 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/cli_login.py +++ b/packages/embeddedci/src/embeddedci/benchpod/cli_login.py @@ -26,6 +26,7 @@ from .cli_config import config_dir from .cloud_auth import USER_AGENT from .errors import CloudAuthError +from embeddedci import _http #: Same margin as the CLI: a token this close to expiry counts as expired. EXPIRY_SKEW = timedelta(seconds=60) @@ -108,7 +109,7 @@ def _refresh(api_base: str, refresh_token: str, timeout: float = _HTTP_TIMEOUT) "User-Agent": USER_AGENT}, ) try: - with urllib.request.urlopen(request, timeout=timeout) as resp: + with _http.urlopen(request, timeout=timeout) as resp: body = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: msg = exc.read().decode("utf-8", "replace") diff --git a/packages/embeddedci/src/embeddedci/benchpod/cloud_auth.py b/packages/embeddedci/src/embeddedci/benchpod/cloud_auth.py index db3f098..9ca7178 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/cloud_auth.py +++ b/packages/embeddedci/src/embeddedci/benchpod/cloud_auth.py @@ -24,6 +24,7 @@ import urllib.error import urllib.parse import urllib.request +from embeddedci import _http from datetime import datetime from typing import Callable, Tuple @@ -85,7 +86,7 @@ def mint_oidc_token(audience: str = DEFAULT_AUDIENCE) -> str: }, ) try: - with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: + with _http.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: body = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: # reason 3 detail = exc.read().decode("utf-8", "replace")[:300] @@ -112,7 +113,7 @@ def exchange_token(api_base: str, oidc_token: str) -> dict: }, ) try: - with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: + with _http.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: msg = exc.read().decode("utf-8", "replace") @@ -141,7 +142,7 @@ def _exchange_for_cloud_session(api_base: str, authorization: str, what: str, }, ) try: - with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: + with _http.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: return json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: msg = exc.read().decode("utf-8", "replace") diff --git a/packages/embeddedci/src/embeddedci/benchpod/lease.py b/packages/embeddedci/src/embeddedci/benchpod/lease.py index fc56344..2e9f054 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/lease.py +++ b/packages/embeddedci/src/embeddedci/benchpod/lease.py @@ -15,6 +15,7 @@ import time import urllib.error import urllib.request +from embeddedci import _http import uuid import warnings from typing import Callable, Optional, Tuple @@ -172,7 +173,7 @@ def _post(self, path: str, *, renewed: bool = False) -> Tuple[Optional[str], Opt request.add_header("Accept", "application/json") request.add_header("User-Agent", USER_AGENT) try: - with urllib.request.urlopen(request, timeout=_LEASE_HTTP_TIMEOUT) as resp: + with _http.urlopen(request, timeout=_LEASE_HTTP_TIMEOUT) as resp: return None, json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", "replace")[:300] diff --git a/packages/embeddedci/src/embeddedci/benchpod/server_api.py b/packages/embeddedci/src/embeddedci/benchpod/server_api.py index bb8aab7..fdb59a0 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/server_api.py +++ b/packages/embeddedci/src/embeddedci/benchpod/server_api.py @@ -19,6 +19,7 @@ import urllib.error import urllib.parse import urllib.request +from embeddedci import _http from typing import Any, Callable, Dict, List, Optional, Tuple from .cloud_auth import DEFAULT_API_BASE, USER_AGENT @@ -108,7 +109,7 @@ def request( headers["X-Benchpod-Lease"] = self.lease_id req = urllib.request.Request(url, data=body, method=method, headers=headers) try: - with urllib.request.urlopen(req, timeout=timeout or self.timeout) as resp: + with _http.urlopen(req, timeout=timeout or self.timeout) as resp: raw = resp.read() if not parse_json: return resp.status, raw diff --git a/packages/embeddedci/src/embeddedci/benchpod/transport/cloud.py b/packages/embeddedci/src/embeddedci/benchpod/transport/cloud.py index d3a404e..bb248bd 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/transport/cloud.py +++ b/packages/embeddedci/src/embeddedci/benchpod/transport/cloud.py @@ -20,6 +20,7 @@ from datetime import datetime, timezone import urllib.error import urllib.request +from embeddedci import _http from typing import Any, Callable, Optional from urllib.parse import quote @@ -426,7 +427,7 @@ def command(self, req: dict) -> Any: # type: ignore[override] if self.lease_id: request.add_header("X-Benchpod-Lease", self.lease_id) try: - with urllib.request.urlopen(request, timeout=timeout + 10) as resp: + with _http.urlopen(request, timeout=timeout + 10) as resp: raw = resp.read() break except urllib.error.HTTPError as exc: diff --git a/packages/embeddedci/tests/test_cloud.py b/packages/embeddedci/tests/test_cloud.py index a97ee7a..1f18e0d 100644 --- a/packages/embeddedci/tests/test_cloud.py +++ b/packages/embeddedci/tests/test_cloud.py @@ -11,6 +11,7 @@ from embeddedci.benchpod.errors import CloudAuthError, ConnectionConfigError from embeddedci.benchpod.transport import open_transport from embeddedci.benchpod.transport.cloud import CloudTransport, _WsTunnelSocket +from embeddedci import _http # -- connection parsing ----------------------------------------------------- @@ -171,7 +172,7 @@ def test_mint_oidc_request_failure(monkeypatch): def boom(*_a, **_k): raise OSError("connection refused") - monkeypatch.setattr(cloud_auth.urllib.request, "urlopen", boom) + monkeypatch.setattr(_http, "urlopen", boom) with pytest.raises(CloudAuthError, match="failed to request a GitHub OIDC token"): cloud_auth.mint_oidc_token() @@ -233,7 +234,7 @@ def fake_urlopen(request, timeout=None): captured["auth"] = request.get_header("Authorization") return _Resp(b'{"status":"ok","data":"pong"}') - monkeypatch.setattr(urllib.request, "urlopen", fake_urlopen) + monkeypatch.setattr(_http, "urlopen", fake_urlopen) assert t.command({"cmd": "ping"}) == "pong" assert "/api/cloud/devices/command" in captured["url"] @@ -259,7 +260,7 @@ def __enter__(self): def __exit__(self, *a): return False - monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **k: _Resp()) + monkeypatch.setattr(_http, "urlopen", lambda *a, **k: _Resp()) with pytest.raises(FirmwareError): t.command({"cmd": "dap_start"}) @@ -413,7 +414,7 @@ def fake(request, timeout=None): raise outcome return _Resp() - monkeypatch.setattr(urllib.request, "urlopen", fake) + monkeypatch.setattr(_http, "urlopen", fake) monkeypatch.setattr(cloud.time, "sleep", lambda s: None) return calls @@ -540,7 +541,7 @@ def fake(request, timeout=None): return _Resp(json.dumps({"token": "cloud", "expires_at": "2030-01-01T00:00:00Z"}).encode()) monkeypatch.delenv("BENCHPOD_API_KEY", raising=False) - monkeypatch.setattr(urllib.request, "urlopen", fake) + monkeypatch.setattr(_http, "urlopen", fake) token, expires_at = cloud_auth.mint_session_token("https://example.test", user_token=lambda: "user-jwt") assert token == "cloud" and seen == {"auth": "Bearer user-jwt", "url": "https://example.test/api/auth/token"} assert expires_at == pytest.approx(1893456000.0) diff --git a/packages/embeddedci/tests/test_http_redirect_auth.py b/packages/embeddedci/tests/test_http_redirect_auth.py new file mode 100644 index 0000000..a15922d --- /dev/null +++ b/packages/embeddedci/tests/test_http_redirect_auth.py @@ -0,0 +1,32 @@ +"""Credentials stay on their own origin across a redirect.""" + +import urllib.request + +from embeddedci._http import _DropAuthOnRedirect + + +class _Hdrs(dict): + def get_all(self, *_a): # what HTTPRedirectHandler may ask of the headers + return None + + +def _redirect(from_url, to_url): + req = urllib.request.Request(from_url, headers={"Authorization": "ApiKey eci_x", "X-Benchpod-Lease": "l1", "Accept": "a"}) + return _DropAuthOnRedirect().redirect_request(req, None, 302, "Found", _Hdrs(), to_url) + + +def test_a_redirect_to_another_host_drops_the_credentials(): + new = _redirect("https://embeddedci.com/api/x", "https://bucket.s3.example/x") + names = {k.lower() for k in new.headers} + assert "authorization" not in names and "x-benchpod-lease" not in names + assert "accept" in names + + +def test_a_redirect_on_the_same_origin_keeps_them(): + new = _redirect("https://embeddedci.com/api/x", "https://embeddedci.com/api/y") + assert {k.lower() for k in new.headers} >= {"authorization", "x-benchpod-lease"} + + +def test_a_downgrade_to_http_drops_them(): + new = _redirect("https://embeddedci.com/api/x", "http://embeddedci.com/api/x") + assert "authorization" not in {k.lower() for k in new.headers} diff --git a/packages/embeddedci/tests/test_lease.py b/packages/embeddedci/tests/test_lease.py index 3986f8c..09f0f5b 100644 --- a/packages/embeddedci/tests/test_lease.py +++ b/packages/embeddedci/tests/test_lease.py @@ -6,6 +6,7 @@ from embeddedci.benchpod.lease import DeviceLease from embeddedci.benchpod.errors import DeviceBusyError +from embeddedci import _http def _lease(posts): @@ -98,7 +99,7 @@ def fake(request, timeout=None): raise outcome return _Resp(b'{"lease_id":"x"}') - monkeypatch.setattr(urllib.request, "urlopen", fake) + monkeypatch.setattr(_http, "urlopen", fake) return calls diff --git a/packages/embeddedci/tests/test_one_source.py b/packages/embeddedci/tests/test_one_source.py index 9616830..2ed6480 100644 --- a/packages/embeddedci/tests/test_one_source.py +++ b/packages/embeddedci/tests/test_one_source.py @@ -22,6 +22,7 @@ from embeddedci.benchpod import cli_config from embeddedci.benchpod.connection import resolve_connection from embeddedci.benchpod.errors import ConnectionConfigError, FirmwareError +from embeddedci import _http DEVICE_ID = "8f1c0d2e-0000-4000-8000-000000000001" PROFILE = {"uart_rx": 3, "uart_tx": 6, "efuse": 2, "signals": [{"name": "READY", "la": 10}]} @@ -180,7 +181,7 @@ def __exit__(self, *exc: object) -> None: @pytest.fixture def http(monkeypatch): h = Http() - monkeypatch.setattr(urllib.request, "urlopen", h.urlopen) + monkeypatch.setattr(_http, "urlopen", h.urlopen) return h diff --git a/packages/embeddedci/tests/test_refusal_errors.py b/packages/embeddedci/tests/test_refusal_errors.py index 304e232..20a74f9 100644 --- a/packages/embeddedci/tests/test_refusal_errors.py +++ b/packages/embeddedci/tests/test_refusal_errors.py @@ -90,7 +90,7 @@ def deny(*_a, **_k): raise urllib.error.HTTPError("https://x.test", 403, "Forbidden", {}, io.BytesIO(b'{"error":"this needs an organization owner or admin"}')) - monkeypatch.setattr("urllib.request.urlopen", deny) + monkeypatch.setattr("embeddedci._http.urlopen", deny) api = ServerApi(api_base="https://x.test", api_key="eci_k") with pytest.raises(PermissionDeniedError) as info: api.list_devices() @@ -101,7 +101,7 @@ def _cloud_http_error(monkeypatch, code, body): def fail(*_a, **_k): raise urllib.error.HTTPError("https://x.test", code, "err", {}, io.BytesIO(body)) - monkeypatch.setattr("urllib.request.urlopen", fail) + monkeypatch.setattr("embeddedci._http.urlopen", fail) return CloudTransport("dev-a", api_base="https://x.test", token="t") From 32ee807cc4f5e76339cd025c960bf1d0084ebf40 Mon Sep 17 00:00:00 2001 From: Edward Viaene Date: Sat, 14 Nov 2026 17:22:08 -0600 Subject: [PATCH 2/2] MCP login refresh also keeps its token on its origin; a test's fake urlopen still applies embeddedci_mcp.cli_login uses embeddedci._http when the SDK has it (2.13.18), plain urllib with an older SDK. _http.urlopen calls a replaced urllib.request.urlopen, so tests that fake it keep working and never reach the network. Co-Authored-By: Claude Opus 5.5 --- .../embeddedci-mcp/src/embeddedci_mcp/cli_login.py | 6 +++++- packages/embeddedci/src/embeddedci/_http.py | 12 +++++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/cli_login.py b/packages/embeddedci-mcp/src/embeddedci_mcp/cli_login.py index 1dbad12..dff6070 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/cli_login.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/cli_login.py @@ -22,6 +22,10 @@ from embeddedci.benchpod.cloud_auth import USER_AGENT from embeddedci.benchpod.errors import CloudAuthError +try: # embeddedci 2.13.18+: credentials stay on their origin across redirects + from embeddedci import _http +except ImportError: # pragma: no cover - older SDK + _http = urllib.request # type: ignore[assignment] #: Same margin as the CLI: a token this close to expiry counts as expired. EXPIRY_SKEW = timedelta(seconds=60) @@ -100,7 +104,7 @@ def _refresh(api_base: str, refresh_token: str) -> Dict[str, Any]: "User-Agent": USER_AGENT}, ) try: - with urllib.request.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: + with _http.urlopen(request, timeout=_HTTP_TIMEOUT) as resp: body = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as exc: msg = exc.read().decode("utf-8", "replace") diff --git a/packages/embeddedci/src/embeddedci/_http.py b/packages/embeddedci/src/embeddedci/_http.py index 3b60f16..199da16 100644 --- a/packages/embeddedci/src/embeddedci/_http.py +++ b/packages/embeddedci/src/embeddedci/_http.py @@ -33,10 +33,20 @@ def redirect_request(self, req, fp, code, msg, headers, newurl): # type: ignore _opener = urllib.request.build_opener(_DropAuthOnRedirect) +_STDLIB_URLOPEN = urllib.request.urlopen def urlopen(req: Any, data: Optional[bytes] = None, timeout: Optional[float] = None, **kw: Any): - """:func:`urllib.request.urlopen` that keeps credentials on their own origin across redirects.""" + """:func:`urllib.request.urlopen` that keeps credentials on their own origin across redirects. + + A test that replaced ``urllib.request.urlopen`` with a fake gets its fake called.""" + if urllib.request.urlopen is not _STDLIB_URLOPEN: + args = {**kw} + if data is not None: + args["data"] = data + if timeout is not None: + args["timeout"] = timeout + return urllib.request.urlopen(req, **args) if timeout is None: return _opener.open(req, data, **kw) return _opener.open(req, data, timeout=timeout, **kw)