wardviaene · Nov 14, 2026 · Nov 14, 2026
@@ -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

@@ -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 <token>` (mandatory off loopback) |
+| `--auth-token` | `EMBEDDEDCI_MCP_TOKEN` | — | require `Authorization: Bearer <token>` (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) |
 

@@ -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"

@@ -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"

@@ -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 <token>' on every request (or set {TOKEN_ENV}). "
-                           "Mandatory when binding a non-loopback address.")
+                      help=f"Require 'Authorization: Bearer <token>' 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)

@@ -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")

@@ -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

@@ -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"

@@ -0,0 +1,52 @@
+"""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)
+_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.
+
+    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)
@@ -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:

@@ -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:<board name>``.
 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()

@@ -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:

@@ -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")

@@ -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")

@@ -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]

@@ -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

@@ -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:

@@ -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)