From 02793eb6d5c5584ee7c14c09e78799d8e7878ce7 Mon Sep 17 00:00:00 2001 From: Edward Viaene Date: Fri, 16 Oct 2026 10:50:51 -0500 Subject: [PATCH 1/3] LAN API over TLS: SDK transport with pinning, lan_tls methods, MCP lan_tls tool The TCP transport connects to 8443 with TLS when the pod has it (firmware 3.9.3+): default port tries TLS first and falls back to plaintext only when 8443 refuses and nothing is pinned; :8443 is TLS only; other ports and the USB link stay plaintext. Pins live in the CLI's known_pods.json (same format), TlsPinMismatchError on a changed certificate, BENCHPOD_TLS / BENCHPOD_TLS_CA. BenchPod.lan_tls/set_lan_tls_https_only/regenerate_lan_tls/set_lan_tls_cert (upload as OTA target tls on one connection: TcpTransport.command_sequence), trust_lan_tls/forget_lan_tls, LanTls. MCP: lan_tls tool. READMEs and changelogs. Co-Authored-By: Claude Opus 5.5 --- packages/embeddedci-mcp/CHANGELOG.md | 6 + packages/embeddedci-mcp/README.md | 7 + .../src/embeddedci_mcp/models.py | 15 + .../src/embeddedci_mcp/server.py | 29 ++ packages/embeddedci-mcp/tests/test_tools.py | 18 +- .../embeddedci-mcp/tests/tools_surface.json | 34 +++ packages/embeddedci/CHANGELOG.md | 10 + packages/embeddedci/README.md | 37 ++- .../src/embeddedci/benchpod/__init__.py | 8 +- .../src/embeddedci/benchpod/client.py | 84 +++++- .../src/embeddedci/benchpod/errors.py | 19 ++ .../src/embeddedci/benchpod/results.py | 34 ++- .../src/embeddedci/benchpod/transport/tcp.py | 78 ++++- .../src/embeddedci/benchpod/transport/tls.py | 266 ++++++++++++++++++ packages/embeddedci/tests/api_surface.json | 32 ++- packages/embeddedci/tests/test_lan_tls.py | 124 ++++++++ 16 files changed, 787 insertions(+), 14 deletions(-) create mode 100644 packages/embeddedci/src/embeddedci/benchpod/transport/tls.py create mode 100644 packages/embeddedci/tests/test_lan_tls.py diff --git a/packages/embeddedci-mcp/CHANGELOG.md b/packages/embeddedci-mcp/CHANGELOG.md index a8b3eb9..538ab22 100644 --- a/packages/embeddedci-mcp/CHANGELOG.md +++ b/packages/embeddedci-mcp/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog — `embeddedci-mcp` +## Unreleased + +- New tool `lan_tls`: the pod's LAN API over TLS (firmware 3.9.3+), HTTPS-only and regenerate. + `connect` to a network address uses TLS when the pod has it and pins its certificate (from the + SDK). + ## 2.13.1 - Requires embeddedci 2.13.1: `Capabilities.scope` is False on the digital-only board, so `status` diff --git a/packages/embeddedci-mcp/README.md b/packages/embeddedci-mcp/README.md index e98697c..a0aba07 100644 --- a/packages/embeddedci-mcp/README.md +++ b/packages/embeddedci-mcp/README.md @@ -55,6 +55,12 @@ lists the pods it heard (name and address) and what to pass instead: `host[:port `claude mcp add benchpod -- uvx embeddedci-mcp` is enough for a single pod on your network: just ask the agent to connect. +**TLS.** A network address uses TLS on port 8443 when the pod has it (firmware 3.9.3+) and pins +the pod's certificate on first use, in the same `known_pods.json` as `benchpod-cli`; older pods +stay on plaintext 8080. `BENCHPOD_TLS=on|off` and `BENCHPOD_TLS_CA` work as in the SDK. A changed +certificate stops `connect` with both fingerprints: after a deliberate change, the user runs +`benchpod lan-tls trust`. The `lan_tls` tool shows the certificate and switches HTTPS-only. + ### Claude Desktop / Cursor `claude_desktop_config.json` (Claude Desktop) or `.cursor/mcp.json` (Cursor): @@ -261,6 +267,7 @@ the pod's second UART, so `uart_open` keeps the console. Needs gateware v48 (`gp | --- | --- | --- | | `set_pull` | Engage or release the bias resistors: LA1-LA6 pull up, LA7/LA8 pull down. | **`las`**, **`enabled`** | | `pull_status` | State, direction and value of the bias resistor on LA1-LA8. | none | +| `lan_tls` | The pod's LAN API over TLS (firmware 3.9.3+): port 8443, the HTTPS-only setting and the certificate's sha256 (what clients pin). Turning HTTPS-only off and regenerating need a TLS, USB or cloud connection. | `https_only`, `regenerate` | | `usb_network` | Read the pod's USB network setting (firmware 3.9.0+), or turn the adapter off over the USB link. Only the device changes it; turning it back on is `benchpod usb-network on` on the USB console. | `set=off` | ### GPIO on the LA pins diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/models.py b/packages/embeddedci-mcp/src/embeddedci_mcp/models.py index f462791..a8ce2c9 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/models.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/models.py @@ -658,6 +658,21 @@ class StepResult(BaseModel): status: str = "started" +class LanTlsResult(BaseModel): + """The pod's LAN API over TLS (firmware 3.9.3+).""" + port: int = Field(description="The TLS port (8443).") + https_only: bool = Field(description="True: plaintext 8080 is refused on Ethernet and Wi-Fi (the USB link keeps it).") + sessions: int = Field(description="TLS connections open now (at most 2).") + origin: str = Field("", description='"generated" (the pod made it) or "user" (installed).') + key: str = Field("", description='The key type, e.g. "EC P-256".') + subject: str = Field("", description="The certificate's subject.") + san: List[str] = Field(default_factory=list, description="The certificate's DNS names.") + not_after: str = Field("", description="Valid until (YYYY-MM-DD).") + sha256: str = Field("", description="SHA-256 of the certificate: the value clients pin.") + connected_over_tls: bool = Field(False, description="Whether this MCP server's connection to the pod uses TLS.") + error: str = Field("", description="Why there is no certificate in use.") + + class UsbNetworkResult(BaseModel): """The pod's USB network setting (firmware 3.9.0+).""" setting: str = Field(description='"on" or "off": whether the pod\'s USB port is also a network adapter.') diff --git a/packages/embeddedci-mcp/src/embeddedci_mcp/server.py b/packages/embeddedci-mcp/src/embeddedci_mcp/server.py index b7a98b5..4f51bae 100644 --- a/packages/embeddedci-mcp/src/embeddedci_mcp/server.py +++ b/packages/embeddedci-mcp/src/embeddedci_mcp/server.py @@ -1097,6 +1097,35 @@ def op() -> m.UsbNetworkResult: return await _call(op) +@mcp.tool(annotations=_ann("LAN TLS", destructive=True)) +async def lan_tls( + https_only: Annotated[Optional[bool], Field(description=( + "Omit to read. True: the pod refuses plaintext port 8080 on Ethernet and Wi-Fi, so only TLS " + "(8443) crosses the LAN; its USB link keeps 8080. False turns it back off, which the pod " + "only takes over TLS, USB or the cloud."))] = None, + regenerate: Annotated[bool, Field(description=( + "True: the pod makes a new key and self-signed certificate. Clients that pinned the old one " + "must trust the new one (`benchpod lan-tls trust`); this server's pin follows."))] = False, +) -> m.LanTlsResult: + """The pod's LAN API over TLS (firmware 3.9.3+): port 8443, HTTPS-only, and the certificate + clients pin (sha256). Connections to a network address use TLS by default when the pod has it. + Installing your own certificate: `benchpod lan-tls set-cert` or the SDK's set_lan_tls_cert.""" + def op() -> m.LanTlsResult: + pod = SESSION.require() + if regenerate: + info = pod.regenerate_lan_tls() + elif https_only is not None: + info = pod.set_lan_tls_https_only(https_only) + else: + info = pod.lan_tls() + return m.LanTlsResult(port=info.port, https_only=info.https_only, sessions=info.sessions, + origin=info.origin, key=info.key, subject=info.subject, san=list(info.san), + not_after=info.not_after, sha256=info.sha256, error=info.error, + connected_over_tls=bool(getattr(pod.transport, "last_tls", False))) + + return await _call(op) + + @mcp.tool(annotations=_ann("Pull resistor status", read_only=True)) async def pull_status() -> m.PullStatusResult: """State, direction and value of the bias resistor on LA1-LA8.""" diff --git a/packages/embeddedci-mcp/tests/test_tools.py b/packages/embeddedci-mcp/tests/test_tools.py index be33300..3a4f214 100644 --- a/packages/embeddedci-mcp/tests/test_tools.py +++ b/packages/embeddedci-mcp/tests/test_tools.py @@ -24,7 +24,7 @@ "enable_i2c_sensor", "set_i2c_sensor", "disable_i2c_sensor", "i2c_sensor_status", "i2c_sensor_regs", "i2c_sensor_capture", "i2c_sensor_types", "enable_gps", "set_gps", "disable_gps", "gps_status", - "set_pull", "pull_status", "usb_network", + "set_pull", "pull_status", "usb_network", "lan_tls", "analog_path", "dac_output", "current_out", "adc_read", "calibration", "calibrate", "capture_adc", "capture_la", "capture_correlated", "decode_la", "la_timing", "generate", "dac_stop", "replay", "list_waveforms", "replay_waveform", @@ -612,6 +612,22 @@ def test_la_step_and_command(connected): call("command", request={"samples": 1}) +def test_lan_tls_reads_and_changes(fake_open, fake_transport, monkeypatch): + from embeddedci.benchpod import LanTls + call("connect", connection="usb", la_voltage=3.3) + pod = SESSION.require() + seen = [] + info = LanTls(port=8443, https_only=False, sessions=0, sha256="ab" * 32, san=("benchpod-x",)) + monkeypatch.setattr(pod, "lan_tls", lambda: seen.append("read") or info, raising=False) + monkeypatch.setattr(pod, "set_lan_tls_https_only", lambda on: seen.append(on) or info, raising=False) + monkeypatch.setattr(pod, "regenerate_lan_tls", lambda: seen.append("regen") or info, raising=False) + res = call("lan_tls") + assert res["port"] == 8443 and res["sha256"] == "ab" * 32 and res["san"] == ["benchpod-x"] + call("lan_tls", https_only=True) + call("lan_tls", regenerate=True) + assert seen == ["read", True, "regen"] + + def test_usb_network_reads_and_turns_off(fake_open, fake_transport, monkeypatch): call("connect", connection="usb", la_voltage=3.3) pod = SESSION.require() diff --git a/packages/embeddedci-mcp/tests/tools_surface.json b/packages/embeddedci-mcp/tests/tools_surface.json index 498e990..ea3bc28 100644 --- a/packages/embeddedci-mcp/tests/tools_surface.json +++ b/packages/embeddedci-mcp/tests/tools_surface.json @@ -2572,6 +2572,40 @@ "type": "object" } }, + "lan_tls": { + "annotations": { + "destructiveHint": true, + "idempotentHint": false, + "openWorldHint": false, + "readOnlyHint": false, + "title": "LAN TLS" + }, + "input": { + "properties": { + "https_only": { + "anyOf": [ + { + "type": "boolean" + }, + { + "type": "null" + } + ], + "default": null, + "description": "Omit to read. True: the pod refuses plaintext port 8080 on Ethernet and Wi-Fi, so only TLS (8443) crosses the LAN; its USB link keeps 8080. False turns it back off, which the pod only takes over TLS, USB or the cloud.", + "title": "Https Only" + }, + "regenerate": { + "default": false, + "description": "True: the pod makes a new key and self-signed certificate. Clients that pinned the old one must trust the new one (`benchpod lan-tls trust`); this server's pin follows.", + "title": "Regenerate", + "type": "boolean" + } + }, + "title": "lan_tlsArguments", + "type": "object" + } + }, "list_waveforms": { "annotations": { "idempotentHint": false, diff --git a/packages/embeddedci/CHANGELOG.md b/packages/embeddedci/CHANGELOG.md index 94d6433..ea3c5e8 100644 --- a/packages/embeddedci/CHANGELOG.md +++ b/packages/embeddedci/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog — `embeddedci` +## Unreleased + +- The LAN API over TLS (pod firmware 3.9.3+): a network address on the default port connects to + 8443 with TLS when the pod has it and falls back to plaintext 8080 only when 8443 refuses and + nothing is pinned. The pod's certificate is pinned on first use in `known_pods.json`, shared with + `benchpod-cli`; a different one raises `TlsPinMismatchError`. `BENCHPOD_TLS` (auto/on/off) and + `BENCHPOD_TLS_CA` (verify against a CA instead). New: `BenchPod.lan_tls()`, + `set_lan_tls_https_only()`, `regenerate_lan_tls()`, `set_lan_tls_cert()`, `LanTls`, + `trust_lan_tls()`, `forget_lan_tls()`, `TcpTransport.last_tls`, `TcpTransport.command_sequence()`. + ## 2.13.1 - `Capabilities.scope` is False on the digital-only board. Current firmware lists `signal` on every diff --git a/packages/embeddedci/README.md b/packages/embeddedci/README.md index e7c44e1..b4c9229 100644 --- a/packages/embeddedci/README.md +++ b/packages/embeddedci/README.md @@ -254,6 +254,36 @@ power moved to the network too; see the compatibility table above.) | `replay()` / client-side `replay_waveform()` (streams the waveform to the pod) | yes | yes | no | yes | | Device lease | — | — | — | yes | +### TLS on the LAN + +Pods with firmware 3.9.3 or later also serve the API on port 8443 with TLS. For a network address +on the default port (`192.168.1.5`, `:8080`, a pod name) the SDK connects to 8443 first and falls +back to plaintext 8080 only when the pod has no TLS port and nothing is pinned for it, so older +pods keep working. `:8443` means TLS only; any other port and the USB network adapter (`usb`, a +physical link) stay plaintext. `BENCHPOD_TLS=on|off` overrides (`auto` is the default). +`bp.transport.last_tls` says which one the last connection used. + +The pod's certificate is its own, so the SDK pins its SHA-256 the first time (trust on first use) +in `known_pods.json` next to `benchpod-cli`'s config; the CLI reads and writes the same file. A +different certificate later raises `TlsPinMismatchError` (`addr`, `pinned`, `got`). After a +deliberate change elsewhere, `trust_lan_tls(host, sha256=None)` pins the certificate the pod +presents now (`sha256` only accepts that one) and `forget_lan_tls(host)` drops the pin. With +`BENCHPOD_TLS_CA=ca.pem` the certificate and host name are verified against that CA instead (for a +certificate you installed, signed by your own CA). + +```python +bp = BenchPod("192.168.1.220") +info = bp.lan_tls() # LanTls: port, https_only, sessions, subject, san, sha256, ... +bp.set_lan_tls_https_only(True) # plaintext 8080 off on Ethernet and Wi-Fi (USB keeps it) +bp.set_lan_tls_cert(open("pod.pem").read(), open("pod.key").read()) +bp.regenerate_lan_tls() +``` + +`set_lan_tls_https_only(False)`, `regenerate_lan_tls()` and `set_lan_tls_cert()` need a TLS, USB or +cloud connection; over plaintext Ethernet or Wi-Fi the pod refuses (`PodLockedError`). +`set_lan_tls_cert` needs a network connection (it uploads the bundle); after it and after +`regenerate_lan_tls()` the SDK pins the new certificate for the address it used. + ### Environment variables | Variable | Used when | Purpose | @@ -266,6 +296,8 @@ power moved to the network too; see the compatibility table above.) | `BENCHPOD_WIRING` | no `wiring` argument / `--benchpod-wiring` / `benchpod_wiring` fixture | wiring profile file (`.json`/`.toml`), see [Wiring profile](#wiring-profile) | | `BENCHPOD_WIRING_SOURCE` | no `wiring_source` argument | `local` never asks embeddedci.com for the pod's stored wiring profile; `auto` (default) does, see [Wiring profile](#wiring-profile) | | `BENCHPOD_LIFT_DAC_LIMITS` | pytest sessions | `1` clears the pod's DAC output limits for the session and restores the same limits at the end, even when tests fail (only with the output stage switched off) | +| `BENCHPOD_TLS` | network connections | `auto` (default), `on` or `off`: TLS to the pod's LAN API on 8443, see [TLS on the LAN](#tls-on-the-lan) | +| `BENCHPOD_TLS_CA` | network connections over TLS | PEM file of CA certificates: verify the pod's certificate and host name instead of pinning it | | `BENCHPOD_STALL_TIMEOUT` | flashing | seconds with no SWD traffic before a flash attempt is aborted and retried (default 60) | ## API conventions and stability @@ -1501,6 +1533,7 @@ Invalid arguments raise `ValueError` before anything is sent. Everything else ra | `TransportError` | the pod or tunnel could not be reached / talked to | | | `TransportTimeout` (a `TransportError` and a `TimeoutError`) | the pod or tunnel did not answer in time | | | `ConnectionClosedError` (a `TransportError`) | the connection ended or reset before the reply | | +| `TlsPinMismatchError` (a `TransportError`) | the pod's TLS certificate is not the one pinned for it (see [TLS on the LAN](#tls-on-the-lan)) | `addr`, `pinned`, `got` | | `FirmwareError` | the pod replied with an error (e.g. "la voltage not set"); a fix the firmware spells as a raw protocol command is rewritten into the SDK call in the message, and `firmware_message` keeps the pod's own text | `firmware_message`, `cmd` | | `FlashError` | OpenOCD failed, is missing, or lacks the TCP backend | | | `TargetUnreachableError` (a `FlashError`) | the probe worked but no target answered on SWD | | @@ -1545,7 +1578,7 @@ lease_wait=600.0, lease_ttl=120, wiring=None)`. A context manager; `close()` is | Area | Methods and properties | |---|---| -| Connection and device | `ping()`, `status()`, `capabilities`, `refresh_capabilities()` (drop the cached capabilities and read them again), `close()`, `leased` (true while this client holds the cloud lease), `wiring`, `save_wiring(wiring=None)` (store a cloud device's profile on embeddedci.com), `signal(name)`, `fpga_image(image)`, `usb_cc()`, `usb_network(on=None)` (the USB network setting; `False` turns the adapter off over the USB link) | +| Connection and device | `ping()`, `status()`, `capabilities`, `refresh_capabilities()` (drop the cached capabilities and read them again), `close()`, `leased` (true while this client holds the cloud lease), `wiring`, `save_wiring(wiring=None)` (store a cloud device's profile on embeddedci.com), `signal(name)`, `fpga_image(image)`, `usb_cc()`, `usb_network(on=None)` (the USB network setting; `False` turns the adapter off over the USB link), `lan_tls()`, `set_lan_tls_https_only(on)`, `regenerate_lan_tls()`, `set_lan_tls_cert(cert_pem, key_pem)` (the LAN API over TLS, see [TLS on the LAN](#tls-on-the-lan)) | | LA voltage | `set_la_voltage(volts)`, `get_la_voltage()` | | Power and reset | `power_on(efuse=None, *, delay=None)`, `power_off(efuse=None, *, delay=None)`, `target_power(efuse=None, *, on, delay=None)`, `target_status()`, `power_status()`, `measure_power(duration, *, efuse=None, rate_hz=500.0, keep_samples=0)`, `power_profile(*, efuse=None, rate_hz=500.0, keep_samples=4096, max_duration=60.0)`, `reset_target(*, pulse=0.1)`, `set_reset(asserted)`, `reset_state()` | | Flashing | `flash(...)`, `spi_flash(image, addr=0, *, erase=True, verify=True, hold_reset=False, sck=, mosi=, miso=, cs=)`, `open_spi(...)`, `open_motor_emulator(...)` | @@ -1577,6 +1610,7 @@ few built from other shapes); `to_dict()` goes the other way where a type is sen | `PowerProfile` | `measure_power`, `PowerProfileSession.result` | see [Power profiles](#power-profiles) | | `PowerProfileSession` | `power_profile` (a context manager) | `start()`, `stop()`, `status()`, `result` | | `ResetState` | `reset_target`, `set_reset`, `reset_state` | `asserted`, `supported` | +| `LanTls` | `lan_tls`, `set_lan_tls_https_only`, `regenerate_lan_tls`, `set_lan_tls_cert` | `port`, `https_only`, `sessions`, `origin` (`generated`/`user`), `key`, `subject`, `san`, `not_after`, `sha256` (what clients pin), `error` (no certificate) | | `UsbCcStatus` | `usb_cc` | `orientation`, `advertised`, `advertised_current` (A), `cc1_voltage`, `cc2_voltage` (V) | | `PullState` | `pull_state`, `set_pull` | `la`, `enabled`, `direction`, `ohms`, `available` | | `LaPinState` | `la_pins`, `configure_gpio` | `la`, `function`, `gpio`, `level`, `pull`, `pull_ohms`, `pull_on`, `in_use` | @@ -1677,6 +1711,7 @@ timeout=60.0, lease_id=None)`. Errors raise `ServerApiError` (`status`), and an |---|---| | `parse_connection(raw)` | parse a connection string into a `ConnSpec` (raises `ConnectionConfigError`) | | `resolve_connection(connection=None)` | the same, falling back to `BENCHPOD_CONNECTION` | +| `trust_lan_tls(host, port=8080, *, sha256=None)`, `forget_lan_tls(host, port=8080)` | pin the TLS certificate the pod presents now (returns its SHA-256), or drop the pin (see [TLS on the LAN](#tls-on-the-lan)) | | `make_build_reporter(*, api_base=None, audience=None, target="", name="")` | a `BuildReporter` inside GitHub Actions, else a `NoopBuildReporter` | | `build_panel_curve(voc_code, sharpness=4.0, points=256)`, `build_linear_curve(max_code, rising=True, points=256)`, `build_constant_curve(value_code, points=256)` | control-loop curves (see [In-fabric DAC control loop](#in-fabric-dac-control-loop)) | | `curve_output_at(curve, input_code)`, `input_percent_to_code(percent)` | evaluate a curve the way the gateware does; an input percentage as a code | diff --git a/packages/embeddedci/src/embeddedci/benchpod/__init__.py b/packages/embeddedci/src/embeddedci/benchpod/__init__.py index 891fc0f..8e70991 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/__init__.py +++ b/packages/embeddedci/src/embeddedci/benchpod/__init__.py @@ -95,6 +95,7 @@ def test_boots(benchpod_target, firmware): # fixtures from the pytest plugin PodLockedError, PullConflictError, TargetUnreachableError, + TlsPinMismatchError, TransportError, TransportTimeout, TriggerTimeout, @@ -109,7 +110,8 @@ def test_boots(benchpod_target, firmware): # fixtures from the pytest plugin from .lowlevel import LowLevel from .power import PowerProfile, PowerProfileSession from .replay import DacHandle, Fault, ReplayHandle, Segment -from .results import Capture, CorrelatedCapture, LaCapture, Trigger +from .results import Capture, CorrelatedCapture, LaCapture, LanTls, Trigger +from .transport.tls import delete_pin as forget_lan_tls, trust_lan_tls from .server_api import ServerApi, ServerApiError, ServerPermissionDeniedError from .state import ( AdcReading, @@ -190,6 +192,10 @@ def test_boots(benchpod_target, firmware): # fixtures from the pytest plugin "dsp", "Capabilities", "Capture", + "LanTls", + "TlsPinMismatchError", + "trust_lan_tls", + "forget_lan_tls", "LaCapture", "CorrelatedCapture", "Trigger", diff --git a/packages/embeddedci/src/embeddedci/benchpod/client.py b/packages/embeddedci/src/embeddedci/benchpod/client.py index cca7286..9c3d8ea 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/client.py +++ b/packages/embeddedci/src/embeddedci/benchpod/client.py @@ -92,7 +92,7 @@ from .lowlevel import LowLevel from .power import PowerProfile, PowerProfileSession from .replay import DacHandle, Fault, ReplayHandle, normalize_fault -from .results import Capture, CorrelatedCapture, LaCapture, Trigger +from .results import Capture, CorrelatedCapture, LaCapture, LanTls, Trigger from .wiring import Signal, Wiring from .state import ( AdcReading, @@ -710,6 +710,88 @@ def usb_network(self, on: Optional[bool] = None) -> bool: data = self.command(req) return isinstance(data, dict) and data.get("setting") == "on" + # -- LAN TLS ------------------------------------------------------------ + + def lan_tls(self) -> "LanTls": + """The pod's LAN API over TLS (firmware 3.9.3+): port 8443, the HTTPS-only setting and the + certificate in use (its SHA-256 is what clients pin). Raises + :class:`UnsupportedFeatureError` on older firmware.""" + return LanTls.from_reply(self.command({"cmd": "lan_tls"})) + + def set_lan_tls_https_only(self, on: bool) -> "LanTls": + """Turn HTTPS-only on (plaintext 8080 refused on Ethernet and Wi-Fi; the USB link keeps it) + or off. Turning it off needs a TLS, USB or cloud connection: over plaintext Ethernet or Wi-Fi + the pod refuses with :class:`PodLockedError`. On a cloud connection it needs an + organization owner or admin.""" + return LanTls.from_reply(self.command({"cmd": "lan_tls", "https_only": bool(on)})) + + def regenerate_lan_tls(self) -> "LanTls": + """Have the pod make a new key and self-signed certificate. Same transports as turning + HTTPS-only off. On a network connection this computer's pin follows.""" + info = LanTls.from_reply(self.command({"cmd": "lan_tls", "regenerate": True})) + self._repin(info) + return info + + def set_lan_tls_cert(self, cert_pem: Union[str, bytes], key_pem: Union[str, bytes]) -> "LanTls": + """Install a certificate (PEM, optionally followed by its chain) and its unencrypted private + key (EC P-256/P-384 or RSA) for port 8443. The pod checks both and that they match before it + stores them. Network connections only (TLS, or the USB network adapter); this computer's pin + follows.""" + from .transport.tcp import TcpTransport + + if not isinstance(self._transport, TcpTransport): + raise UnsupportedFeatureError( + "set_lan_tls_cert needs a network connection (TLS or the USB network adapter)", + cmd="lan_tls", feature="lan_tls") + c = cert_pem.encode() if isinstance(cert_pem, str) else bytes(cert_pem) + k = key_pem.encode() if isinstance(key_pem, str) else bytes(key_pem) + if b"ENCRYPTED" in k: + raise ValueError("the private key is encrypted; the pod needs it without a passphrase") + bundle = c.rstrip(b"\n") + b"\n" + k + if len(bundle) > 7664: + raise ValueError(f"certificate and key are {len(bundle)} bytes; the pod takes at most 7664") + import hashlib + + reqs: List[Dict[str, Any]] = [ + {"cmd": "lan_tls"}, + {"cmd": "ota_begin", "size": len(bundle), "sha256": hashlib.sha256(bundle).hexdigest(), + "target": "tls"}, + ] + for off in range(0, len(bundle), 768): + reqs.append({"cmd": "ota_data", "offset": off, + "data": base64.urlsafe_b64encode(bundle[off:off + 768]).rstrip(b"=").decode()}) + reqs += [{"cmd": "ota_end"}, {"cmd": "ota_commit"}, {"cmd": "lan_tls"}] + try: + replies = self._transport.command_sequence(reqs) + except FirmwareError: + try: + self.command({"cmd": "ota_abort"}) + except BenchPodError: + pass + raise + for name, rep in (("ota_end", replies[-3]), ("ota_commit", replies[-2])): + state = rep.get("state") if isinstance(rep, dict) else None + want = "verified" if name == "ota_end" else "installed" + if state != want: + why = rep.get("error") if isinstance(rep, dict) else rep + raise FirmwareError(f"{name}: {why or state}", cmd=name) + info = LanTls.from_reply(replies[-1]) + self._repin(info) + return info + + def _repin(self, info: "LanTls") -> None: + """After this computer changed the pod's certificate: pin the new one for its address.""" + from .transport import tls as _tls + from .transport.tcp import TcpTransport + + t = self._transport + if not isinstance(t, TcpTransport) or not info.sha256: + return + host, port = t._split_addr() + if _tls.is_usb_link(host): + return + _tls.set_pin(host, info.sha256, port=port, subject=info.subject) + # -- flash -------------------------------------------------------------- def flash( diff --git a/packages/embeddedci/src/embeddedci/benchpod/errors.py b/packages/embeddedci/src/embeddedci/benchpod/errors.py index dc7c2d5..acf04a7 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/errors.py +++ b/packages/embeddedci/src/embeddedci/benchpod/errors.py @@ -30,6 +30,24 @@ class ConnectionClosedError(TransportError): """The connection ended (EOF, reset or a closed tunnel) before the full reply arrived.""" +class TlsPinMismatchError(TransportError): + """The pod presented a TLS certificate other than the one pinned for it (LAN TLS, firmware + 3.9.3+). Either its certificate was replaced or regenerated, or something is intercepting the + connection. :attr:`addr` is the pod's TLS address, :attr:`pinned` and :attr:`got` the two + SHA-256 fingerprints. Trust the new one with :func:`embeddedci.benchpod.trust_lan_tls` or + ``benchpod lan-tls trust``.""" + + def __init__(self, addr: str, pinned: str, got: str) -> None: + super().__init__( + f"the pod at {addr} presented TLS certificate sha256 {got[:16]}..., but " + f"{pinned[:16]}... is pinned for it. If its certificate was replaced or regenerated, " + f"trust the new one (`benchpod lan-tls trust`); otherwise something may be " + f"intercepting the connection") + self.addr = addr + self.pinned = pinned + self.got = got + + class DeviceBusyError(BenchPodError): """The shared cloud BenchPod is held by another consumer and did not free within the wait timeout. Raised when acquiring the device lease times out.""" @@ -223,6 +241,7 @@ def __init__(self, message: str, *, cmd: Optional[str] = None, _LEASED = re.compile(r"busy: a cloud job holds this pod \((.*), (\d+) s left\)") #: Policy changes the LAN may not make (pod_policy.c). _POLICY_ELSEWHERE = ("change it from the cloud or the USB console", + "change it over TLS (port 8443), over USB or from the cloud", "change it from the cloud or over USB", "change it on the device", "only the USB console can loosen") diff --git a/packages/embeddedci/src/embeddedci/benchpod/results.py b/packages/embeddedci/src/embeddedci/benchpod/results.py index 7ef1cd8..0211fbc 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/results.py +++ b/packages/embeddedci/src/embeddedci/benchpod/results.py @@ -16,7 +16,7 @@ import math from dataclasses import dataclass, field -from typing import List, Optional, Sequence, Union +from typing import Any, List, Optional, Sequence, Tuple, Union from .constants import TRIGGER_EDGES, Edge, TriggerEdge @@ -321,3 +321,35 @@ class CorrelatedCapture: def __len__(self) -> int: return max(len(self.adc), len(self.la)) + + +@dataclass(frozen=True) +class LanTls: + """The pod's LAN API over TLS (:meth:`BenchPod.lan_tls`, firmware 3.9.3+).""" + + port: int + https_only: bool + sessions: int + origin: str = "" # "generated" or "user" + key: str = "" # "EC P-256", "RSA 2048", ... + subject: str = "" # "CN=benchpod-baea06" + san: Tuple[str, ...] = () + not_after: str = "" # "9999-12-31" + sha256: str = "" # of the certificate (DER): the value clients pin + error: str = "" # why there is no certificate in use + + @classmethod + def from_reply(cls, data: Any) -> "LanTls": + d = data if isinstance(data, dict) else {} + return cls( + port=int(d.get("port") or 8443), + https_only=bool(d.get("https_only")), + sessions=int(d.get("sessions") or 0), + origin=str(d.get("origin") or ""), + key=str(d.get("key") or ""), + subject=str(d.get("subject") or ""), + san=tuple(str(x) for x in (d.get("san") or [])), + not_after=str(d.get("not_after") or ""), + sha256=str(d.get("sha256") or ""), + error=str(d.get("error") or ""), + ) diff --git a/packages/embeddedci/src/embeddedci/benchpod/transport/tcp.py b/packages/embeddedci/src/embeddedci/benchpod/transport/tcp.py index 56fbe4e..ec5e210 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/transport/tcp.py +++ b/packages/embeddedci/src/embeddedci/benchpod/transport/tcp.py @@ -1,4 +1,5 @@ -"""JSON-over-TCP transport (wifi/network), port 8080 by default. +"""JSON-over-TCP transport (wifi/network), port 8080 by default, TLS on 8443 when the pod has it +(firmware 3.9.3+, see :mod:`.tls`). Mirrors the Go ``tcpclient``: each command dials a fresh connection (the firmware serves one client at a time), disables Nagle, sends one JSON line and @@ -10,6 +11,7 @@ from __future__ import annotations import json +import os import socket import time from contextlib import contextmanager @@ -24,6 +26,7 @@ ) from ..protocol import encode_request, parse_reply, raise_for_status from .base import RawLink, Transport, request_timeout +from . import tls as _tls DEFAULT_DIAL_TIMEOUT = 10.0 # total budget to establish a connection _DIAL_ATTEMPT_TIMEOUT = 0.5 @@ -97,13 +100,22 @@ class TcpTransport(Transport): #: with its USB network off or down); ``""`` otherwise. usb_lan_fallback: str = "" + #: Whether the last connection used TLS (port 8443). + last_tls: bool = False + def __init__(self, addr: str, *, timeout: float = 30.0, - dial_timeout: float = DEFAULT_DIAL_TIMEOUT) -> None: + dial_timeout: float = DEFAULT_DIAL_TIMEOUT, tls: Optional[str] = None, + tls_ca: Optional[str] = None) -> None: + """``tls``: ``"auto"`` (default, also ``BENCHPOD_TLS``), ``"on"`` or ``"off"``; + ``tls_ca``: a PEM file of CA certificates to verify the pod's certificate against instead + of pinning it (also ``BENCHPOD_TLS_CA``). See :mod:`.tls`.""" if not addr: raise TransportError("TCP transport requires a host:port address") self.addr = addr self.timeout = timeout self.dial_timeout = dial_timeout + self.tls = _tls.tls_mode(tls) + self.tls_ca = tls_ca if tls_ca is not None else (os.environ.get("BENCHPOD_TLS_CA") or None) # -- connection helpers ------------------------------------------------- @@ -124,20 +136,53 @@ def _dial(self, timeout: Optional[float] = None, kind: Optional[str] = None) -> ``"flash"``). The cloud transport passes it to the server so a UART console and a capture can share the pod as they do here; a LAN socket needs no such hint and ignores it.""" host, port = self._split_addr() - deadline = time.monotonic() + self.dial_timeout + io_timeout = self.timeout if timeout is None else timeout + p = _tls.plan(host, port, self.tls) + tls_to = p.tls + if p.fallback and _tls.known_no_tls(p.plain) and not _tls.get_pin(host, port): + tls_to = None + if tls_to is None: + self.last_tls = False + return self._connect(p.plain, self.dial_timeout, io_timeout) + if p.fallback and not _tls.get_pin(host, port): + # One quick probe: a pod without LAN TLS answers 8443 with a reset. + try: + raw = self._connect(tls_to, _DIAL_ATTEMPT_TIMEOUT, io_timeout, retry=False) + except TransportError as exc: + if isinstance(exc.__cause__, ConnectionRefusedError): + _tls.remember_no_tls(p.plain) + self.last_tls = False + return self._connect(p.plain, self.dial_timeout, io_timeout) + else: + raw = self._connect(tls_to, self.dial_timeout, io_timeout) + try: + sock = _tls.wrap(raw, tls_to[0], tls_to[1], ca=self.tls_ca) + except BenchPodError: + raw.close() + raise + except (OSError, ValueError) as exc: # ssl.SSLError is an OSError + raw.close() + raise TransportError(f"TLS handshake with {tls_to[0]}:{tls_to[1]}: {exc}") from exc + self.last_tls = True + return sock + + def _connect(self, addr: "tuple[str, int]", budget: float, io_timeout: Optional[float], + *, retry: bool = True) -> socket.socket: + """A TCP connection to ``addr``, re-probing within ``budget`` seconds.""" + deadline = time.monotonic() + budget last: Optional[Exception] = None while True: try: - sock = socket.create_connection((host, port), timeout=_DIAL_ATTEMPT_TIMEOUT) + sock = socket.create_connection(addr, timeout=_DIAL_ATTEMPT_TIMEOUT) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) - sock.settimeout(self.timeout if timeout is None else timeout) + sock.settimeout(io_timeout) return sock except OSError as exc: last = exc - if time.monotonic() >= deadline: + if not retry or time.monotonic() >= deadline: break time.sleep(_DIAL_RETRY_BACKOFF) - raise TransportError(f"could not connect to {self.addr}: {last}") + raise TransportError(f"could not connect to {addr[0]}:{addr[1]}: {last}") from last @staticmethod def _recv_line(sock: socket.socket, buf: bytearray) -> bytes: @@ -189,6 +234,25 @@ def _recv_line_exact(sock: socket.socket) -> bytes: # -- Transport API ------------------------------------------------------ + def command_sequence(self, reqs: List[dict]) -> List[Any]: + """Send several JSON commands on ONE connection and return their ``data`` in order + (raises on the first error). For what the pod ties to a connection, such as an upload + (``ota_*``).""" + timeout = max((request_timeout(r, self.timeout) for r in reqs), default=self.timeout) + sock = self._dial(timeout) + out: List[Any] = [] + buf = bytearray() + try: + for req in reqs: + with _io_errors(f"{req.get('cmd')}", timeout): + sock.sendall(encode_request(req)) + reply = parse_reply(self._recv_line(sock, buf)) + raise_for_status(reply, cmd=req.get("cmd")) + out.append(reply.data) + return out + finally: + sock.close() + def command(self, req: dict) -> Any: """Send one JSON command and return its ``data`` (raises on error).""" timeout = request_timeout(req, self.timeout) diff --git a/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py b/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py new file mode 100644 index 0000000..bb4b3b4 --- /dev/null +++ b/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py @@ -0,0 +1,266 @@ +"""The pod's LAN API over TLS (firmware 3.9.3+), as ``benchpod`` does it (its ``tcpclient/tls.go``). + +The pod serves the same API on TCP 8443 with TLS 1.2 and its own certificate (self-signed on first +boot, or one the user installed). How :class:`~.tcp.TcpTransport` connects: + +- port 8443: TLS only. +- port 8080 (the default): TLS on 8443 first; when 8443 refuses the connection (firmware without + LAN TLS) and nothing is pinned for the pod, plaintext 8080 as before. With a pin, a pod that stops + answering TLS is an error, never a silent downgrade. +- any other port, and the pod's USB network adapter (10.237.0.0/16, a physical link): plaintext. + +``tls="on"`` / ``"off"`` (or ``BENCHPOD_TLS``) overrides ``"auto"``. + +Trust: the SHA-256 of the pod's certificate (DER) is pinned on first use per address in +``known_pods.json`` next to ``benchpod-cli``'s config (the CLI reads and writes the same file). A +different certificate later raises :class:`~..errors.TlsPinMismatchError`. With a CA file +(``tls_ca`` or ``BENCHPOD_TLS_CA``) the certificate is verified against it and the host name +instead, and nothing is pinned. +""" + +from __future__ import annotations + +import hashlib +import ipaddress +import json +import os +import socket +import ssl +import tempfile +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, NamedTuple, Optional, Tuple + +from ..cli_config import config_dir + +PLAIN_PORT = 8080 +TLS_PORT = 8443 +_USB_LINK = ipaddress.ip_network("10.237.0.0/16") +_lock = threading.Lock() +#: Pods (host:port) that refused 8443 in this process, so each command does not probe again. +_no_tls: set = set() + + +def tls_mode(value: Optional[str] = None) -> str: + """``"auto"``, ``"on"`` or ``"off"`` from ``value``, else ``BENCHPOD_TLS``, else ``"auto"``.""" + v = (value if value is not None else os.environ.get("BENCHPOD_TLS", "")).strip().lower() + if v in ("", "auto"): + return "auto" + if v in ("on", "true", "yes", "1"): + return "on" + if v in ("off", "false", "no", "0"): + return "off" + raise ValueError(f"tls: {v!r} is not auto, on or off") + + +def is_usb_link(host: str) -> bool: + """Whether ``host`` is on the pod's USB network adapter.""" + try: + return ipaddress.ip_address(host) in _USB_LINK + except ValueError: + return False + + +def tls_addr(host: str, port: int) -> Tuple[str, int]: + """Where a TLS connection to this pod goes, also the key of its pin: ``host:8443`` for the + default port, else the address as given.""" + return (host, TLS_PORT) if port == PLAIN_PORT else (host, port) + + +class DialPlan(NamedTuple): + tls: Optional[Tuple[str, int]] # None: no TLS attempt + plain: Optional[Tuple[str, int]] # None: no plaintext + fallback: bool # plaintext only when TLS is refused and nothing is pinned + + +def plan(host: str, port: int, mode: str) -> DialPlan: + """How to reach ``host:port`` in ``mode`` (see the module docstring).""" + if mode == "off": + return DialPlan(None, (host, port), False) + if port == TLS_PORT: + return DialPlan((host, port), None, False) + if mode == "on": + return DialPlan(tls_addr(host, port), None, False) + if port != PLAIN_PORT or is_usb_link(host): + return DialPlan(None, (host, port), False) + return DialPlan((host, TLS_PORT), (host, port), True) + + +def remember_no_tls(addr: Tuple[str, int]) -> None: + _no_tls.add(addr) + + +def known_no_tls(addr: Tuple[str, int]) -> bool: + return addr in _no_tls + + +# -- pins --------------------------------------------------------------------------------------- + + +def pin_store_path() -> Path: + """``known_pods.json`` in ``benchpod-cli``'s config directory.""" + return config_dir() / "known_pods.json" + + +def _key(host: str, port: int) -> str: + h, p = tls_addr(host, port) + return f"[{h}]:{p}" if ":" in h else f"{h}:{p}" + + +def _load(path: Path) -> Dict: + try: + data = json.loads(path.read_text()) + except FileNotFoundError: + return {"version": 1, "pods": {}} + if not isinstance(data, dict): + raise ValueError(f"{path}: not a JSON object") + data.setdefault("version", 1) + if not isinstance(data.get("pods"), dict): + data["pods"] = {} + return data + + +def _save(path: Path, data: Dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, tmp = tempfile.mkstemp(prefix=".known_pods-", dir=str(path.parent)) + try: + with os.fdopen(fd, "w") as f: + json.dump(data, f, indent=2) + f.write("\n") + os.chmod(tmp, 0o600) + os.replace(tmp, path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def get_pin(host: str, port: int = PLAIN_PORT) -> Optional[str]: + """The pinned SHA-256 (lowercase hex) for the pod at ``host``, if any.""" + with _lock: + try: + pod = _load(pin_store_path())["pods"].get(_key(host, port)) + except (OSError, ValueError): + return None + return pod.get("sha256") if isinstance(pod, dict) else None + + +def set_pin(host: str, sha256: str, port: int = PLAIN_PORT, subject: str = "") -> None: + """Pin ``sha256`` for the pod at ``host`` (what ``benchpod lan-tls trust`` does).""" + with _lock: + path = pin_store_path() + data = _load(path) + data["pods"][_key(host, port)] = { + "sha256": sha256.lower(), + "subject": subject, + "added": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + } + _save(path, data) + + +def delete_pin(host: str, port: int = PLAIN_PORT) -> bool: + """Forget the pod's pin. ``False`` when there was none.""" + with _lock: + path = pin_store_path() + data = _load(path) + if data["pods"].pop(_key(host, port), None) is None: + return False + _save(path, data) + return True + + +def fingerprint(der: bytes) -> str: + """Lowercase hex SHA-256 of a certificate (DER): the value the pod reports as ``sha256``.""" + return hashlib.sha256(der).hexdigest() + + +def _check_pin(host: str, port: int, der: bytes) -> None: + from ..errors import TlsPinMismatchError + + got = fingerprint(der) + with _lock: + path = pin_store_path() + data = _load(path) + key = _key(host, port) + pod = data["pods"].get(key) + if isinstance(pod, dict) and pod.get("sha256"): + if pod["sha256"].lower() != got: + raise TlsPinMismatchError(key, pod["sha256"].lower(), got) + return + data["pods"][key] = {"sha256": got, "subject": "", + "added": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")} + _save(path, data) + + +# -- the handshake ------------------------------------------------------------------------------ + + +def wrap(sock: socket.socket, host: str, port: int, *, ca: Optional[str] = None, + pin: bool = True, timeout: float = 10.0) -> ssl.SSLSocket: + """TLS over a connected socket to the pod at ``host:port``. + + With ``ca`` (a PEM file of CA certificates) the certificate chain and host name are verified. + Otherwise the certificate is checked against (or, on first use, recorded as) the pin; with + ``pin=False`` any certificate is accepted (to show or trust it).""" + is_ip = True + try: + ipaddress.ip_address(host) + except ValueError: + is_ip = False + if ca and pin: + ctx = ssl.create_default_context(cafile=ca) + check_name = True + else: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE # the pin below is the check + check_name = False + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + old = sock.gettimeout() + sock.settimeout(timeout) + tls = ctx.wrap_socket(sock, server_hostname=host if (check_name or not is_ip) else None) + try: + if not ca and pin: + der = tls.getpeercert(binary_form=True) + if not der: + raise ssl.SSLError("the pod sent no certificate") + _check_pin(host, port, der) + except BaseException: + tls.close() + raise + tls.settimeout(old) + return tls + + +def trust_lan_tls(host: str, port: int = PLAIN_PORT, *, sha256: Optional[str] = None, + timeout: float = 10.0) -> str: + """Pin the certificate the pod at ``host`` presents now (after a deliberate change), like + ``benchpod lan-tls trust``. With ``sha256`` only that certificate is trusted. Returns the + pinned SHA-256.""" + der = peer_certificate(host, port, timeout=timeout) + got = fingerprint(der) + if sha256 and sha256.strip().lower() != got: + raise ValueError(f"the pod presents sha256 {got}, not {sha256.strip().lower()}; not trusted") + set_pin(host, got, port=port) + return got + + +def peer_certificate(host: str, port: int = PLAIN_PORT, *, timeout: float = 10.0) -> bytes: + """The certificate (DER) the pod at ``host`` presents on its TLS port, unchecked.""" + th, tp = tls_addr(host, port) + sock = socket.create_connection((th, tp), timeout=timeout) + try: + tls = wrap(sock, th, tp, pin=False, timeout=timeout) + except BaseException: + sock.close() + raise + try: + der = tls.getpeercert(binary_form=True) + finally: + tls.close() + if not der: + raise ssl.SSLError("the pod sent no certificate") + return der diff --git a/packages/embeddedci/tests/api_surface.json b/packages/embeddedci/tests/api_surface.json index f3efea4..5ae2fc2 100644 --- a/packages/embeddedci/tests/api_surface.json +++ b/packages/embeddedci/tests/api_surface.json @@ -96,6 +96,7 @@ "i2c_sensor_types": "(self) -> 'List[Dict[str, Any]]'", "la_pins": "(self) -> 'List[LaPinState]'", "la_step": "(self, la: 'Union[Pin, int]', *, steps: 'int', delay: 'float', dir_la: 'Optional[Union[Pin, int]]' = None, direction: 'int' = 0) -> 'Dict[str, Any]'", + "lan_tls": "(self) -> \"'LanTls'\"", "leased": "property", "loop_input": "(self, input_code: 'Optional[int]' = None, *, source: 'Optional[LoopSource]' = None, step: 'Optional[int]' = None) -> 'LoopState'", "loop_probe": "(self) -> '_control_loop.IVPoint'", @@ -114,6 +115,7 @@ "pull_state": "(self, la: 'Union[Pin, int]') -> 'PullState'", "read_gpio": "(self, la: 'Union[Pin, int, str]') -> 'int'", "refresh_capabilities": "(self) -> 'Capabilities'", + "regenerate_lan_tls": "(self) -> \"'LanTls'\"", "release_gpio": "(self, *las: 'Union[Pin, int, str]') -> 'None'", "replay": "(self, source: 'Union[Capture, Sequence[float], Sequence[int]]', *, dac_path: 'Optional[DacPath]' = None, mapping: 'ReplayMapping' = 'faithful', sample_rate_hz: 'Optional[float]' = None, deep: 'Optional[bool]' = None, fault: 'Union[Fault, Dict[str, Any], None]' = None, are_codes: 'bool' = False, on_capture: 'bool' = False, route: 'bool' = True, switch_image: 'bool' = True) -> 'ReplayHandle'", "replay_waveform": "(self, waveform: \"Union[str, 'Waveform']\", *, dac_path: 'Optional[DacPath]' = None, mapping: 'ReplayMapping' = 'faithful', sample_rate_hz: 'Optional[float]' = None, window_start: 'int' = 0, window_len: 'int' = 0, target_samples: 'int' = 0, fault: 'Union[Fault, Dict[str, Any], None]' = None, deep: 'Optional[bool]' = None, server_side: 'Optional[bool]' = None, on_capture: 'bool' = False, switch_image: 'bool' = True) -> 'ReplayHandle'", @@ -126,6 +128,8 @@ "set_gps": "(self, *, utc: 'Union[_dt.datetime, str, int, None]' = None, **fix: 'Any') -> 'Dict[str, Any]'", "set_i2c_sensor": "(self, *, temperature_c: 'Optional[float]' = None, pressure_pa: 'Optional[float]' = None, **values: 'float') -> 'Dict[str, Any]'", "set_la_voltage": "(self, voltage: 'float') -> 'LaVoltage'", + "set_lan_tls_cert": "(self, cert_pem: 'Union[str, bytes]', key_pem: 'Union[str, bytes]') -> \"'LanTls'\"", + "set_lan_tls_https_only": "(self, on: 'bool') -> \"'LanTls'\"", "set_pull": "(self, la: 'Union[Pin, int]', enabled: 'bool') -> 'PullState'", "set_reset": "(self, asserted: 'bool') -> 'ResetState'", "signal": "(self, name: 'str') -> 'GpioPin'", @@ -524,6 +528,21 @@ "from_reply": "(cls, reply: 'Any') -> \"'LaVoltage'\"", "is_set": "property" }, + "LanTls": { + "__fields__": [ + "port", + "https_only", + "sessions", + "origin", + "key", + "subject", + "san", + "not_after", + "sha256", + "error" + ], + "from_reply": "(cls, data: 'Any') -> \"'LanTls'\"" + }, "LoopInputMap": { "__fields__": [ "mv_per_unit", @@ -814,6 +833,9 @@ "from_reply": "(cls, reply: 'Any') -> \"'TargetStatus'\"" }, "TargetUnreachableError": {}, + "TlsPinMismatchError": { + "__init__": "(self, addr: 'str', pinned: 'str', got: 'str') -> 'None'" + }, "TransportError": {}, "TransportTimeout": {}, "Trigger": { @@ -1007,6 +1029,7 @@ "LaCapture", "LaPinState", "LaVoltage", + "LanTls", "LoopInputMap", "LoopSource", "LoopState", @@ -1055,6 +1078,7 @@ "SpiStreamResult", "TargetStatus", "TargetUnreachableError", + "TlsPinMismatchError", "TransportError", "TransportTimeout", "Trigger", @@ -1080,12 +1104,14 @@ "decode", "dsp", "encode_curve_b64url", + "forget_lan_tls", "i2c", "input_percent_to_code", "make_build_reporter", "motor_emulator", "parse_connection", - "resolve_connection" + "resolve_connection", + "trust_lan_tls" ], "build_constant_curve": "(value_code: 'int', points: 'int' = 256) -> 'List[int]'", "build_linear_curve": "(max_code: 'int', rising: 'bool' = True, points: 'int' = 256) -> 'List[int]'", @@ -1096,10 +1122,12 @@ "decode": "module", "dsp": "module", "encode_curve_b64url": "(codes: 'Sequence[int]') -> 'str'", + "forget_lan_tls": "(host: 'str', port: 'int' = 8080) -> 'bool'", "i2c": "module", "input_percent_to_code": "(percent: 'float') -> 'int'", "make_build_reporter": "(*, api_base: 'Optional[str]' = None, audience: 'Optional[str]' = None, target: 'str' = '', name: 'str' = '') -> 'Any'", "motor_emulator": "module", "parse_connection": "(raw: 'str') -> 'ConnSpec'", - "resolve_connection": "(connection: \"'str | None'\" = None) -> 'ConnSpec'" + "resolve_connection": "(connection: \"'str | None'\" = None) -> 'ConnSpec'", + "trust_lan_tls": "(host: 'str', port: 'int' = 8080, *, sha256: 'Optional[str]' = None, timeout: 'float' = 10.0) -> 'str'" } diff --git a/packages/embeddedci/tests/test_lan_tls.py b/packages/embeddedci/tests/test_lan_tls.py new file mode 100644 index 0000000..f49b658 --- /dev/null +++ b/packages/embeddedci/tests/test_lan_tls.py @@ -0,0 +1,124 @@ +"""The LAN API over TLS (transport/tls.py): dial plan, trust on first use, pin mismatch, re-trust, +and the pin file the CLI shares (firmware 3.9.3+).""" + +import json +import shutil +import socket +import ssl +import subprocess +import threading + +import pytest + +from embeddedci.benchpod import TlsPinMismatchError, forget_lan_tls, trust_lan_tls +from embeddedci.benchpod.transport import tls as T +from embeddedci.benchpod.transport.tcp import TcpTransport + + +@pytest.fixture(autouse=True) +def pin_dir(tmp_path, monkeypatch): + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path)) + monkeypatch.delenv("BENCHPOD_TLS", raising=False) + monkeypatch.delenv("BENCHPOD_TLS_CA", raising=False) + return tmp_path + + +def test_plan(): + assert T.plan("192.168.1.5", 8080, "auto") == T.DialPlan(("192.168.1.5", 8443), ("192.168.1.5", 8080), True) + assert T.plan("192.168.1.5", 8443, "auto") == T.DialPlan(("192.168.1.5", 8443), None, False) + assert T.plan("192.168.1.5", 9000, "auto") == T.DialPlan(None, ("192.168.1.5", 9000), False) + assert T.plan("10.237.8.85", 8080, "auto") == T.DialPlan(None, ("10.237.8.85", 8080), False) + assert T.plan("192.168.1.5", 8443, "off") == T.DialPlan(None, ("192.168.1.5", 8443), False) + assert T.plan("10.237.8.85", 8080, "on") == T.DialPlan(("10.237.8.85", 8443), None, False) + assert T.tls_mode("ON") == "on" and T.tls_mode("") == "auto" + with pytest.raises(ValueError): + T.tls_mode("maybe") + + +def _make_cert(tmp_path, name): + if not shutil.which("openssl"): + pytest.skip("openssl not installed") + key, crt = tmp_path / f"{name}.key", tmp_path / f"{name}.pem" + subprocess.run(["openssl", "req", "-x509", "-newkey", "ec", "-pkeyopt", "ec_paramgen_curve:P-256", + "-nodes", "-keyout", str(key), "-out", str(crt), "-days", "1", "-subj", f"/CN={name}"], + check=True, capture_output=True) + return crt, key + + +class _TlsPod: + """A TLS JSON server answering every line with an ok reply; its certificate can change.""" + + def __init__(self, tmp_path): + self.tmp = tmp_path + self.n = 0 + self.rotate() + self.sock = socket.socket() + self.sock.bind(("127.0.0.1", 0)) + self.sock.listen() + self.port = self.sock.getsockname()[1] + threading.Thread(target=self._serve, daemon=True).start() + + def rotate(self): + self.n += 1 + crt, key = _make_cert(self.tmp, f"benchpod-test{self.n}") + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.load_cert_chain(str(crt), str(key)) + self.ctx = ctx + self.der = ssl.PEM_cert_to_DER_cert(crt.read_text()) + + def _serve(self): + while True: + try: + c, _ = self.sock.accept() + except OSError: + return + try: + t = self.ctx.wrap_socket(c, server_side=True) + f = t.makefile("rb") + for _ in f: + t.sendall(b'{"status":"ok","data":"pong"}\n') + except (OSError, ssl.SSLError): + pass + finally: + c.close() + + def close(self): + self.sock.close() + + +def test_pin_on_first_use_mismatch_and_trust(tmp_path, pin_dir): + pod = _TlsPod(tmp_path) + try: + t = TcpTransport(f"127.0.0.1:{pod.port}", tls="on", timeout=5, dial_timeout=2) + assert t.command({"cmd": "ping"}) == "pong" + assert t.last_tls + assert T.get_pin("127.0.0.1", pod.port) == T.fingerprint(pod.der) + # The CLI's file format: version + pods keyed by host:port. + data = json.loads((pin_dir / "benchpod-cli" / "known_pods.json").read_text()) + assert data["pods"][f"127.0.0.1:{pod.port}"]["sha256"] == T.fingerprint(pod.der) + + assert t.command({"cmd": "ping"}) == "pong" + old = pod.der + pod.rotate() + with pytest.raises(TlsPinMismatchError) as ei: + t.command({"cmd": "ping"}) + assert ei.value.pinned == T.fingerprint(old) + assert ei.value.got == T.fingerprint(pod.der) + + with pytest.raises(ValueError): + trust_lan_tls("127.0.0.1", pod.port, sha256="00" * 32) + assert trust_lan_tls("127.0.0.1", pod.port, sha256=ei.value.got) == ei.value.got + assert t.command({"cmd": "ping"}) == "pong" + assert t.command_sequence([{"cmd": "ping"}, {"cmd": "ping"}]) == ["pong", "pong"] + + assert forget_lan_tls("127.0.0.1", pod.port) + assert not forget_lan_tls("127.0.0.1", pod.port) + assert T.get_pin("127.0.0.1", pod.port) is None + finally: + pod.close() + + +def test_pin_key_for_the_default_port_is_the_tls_port(): + T.set_pin("192.168.1.5", "AB" * 32) + assert T.get_pin("192.168.1.5", 8080) == "ab" * 32 + assert T.get_pin("192.168.1.5", 8443) == "ab" * 32 From e4a20237a0d624258e61da669b545a4aa9e4c80f Mon Sep 17 00:00:00 2001 From: Edward Viaene Date: Fri, 16 Oct 2026 11:31:55 -0500 Subject: [PATCH 2/3] TLS transport: resume sessions with the pod One SSLContext per CA mode and the last session per pod address: later connections resume in ~35 ms instead of a ~0.46 s full handshake (HW, a1f6f2). A pin change drops the sessions. Co-Authored-By: Claude Opus 5.5 --- .../src/embeddedci/benchpod/transport/tls.py | 52 ++++++++++++++----- 1 file changed, 40 insertions(+), 12 deletions(-) diff --git a/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py b/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py index bb4b3b4..a52cbb7 100644 --- a/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py +++ b/packages/embeddedci/src/embeddedci/benchpod/transport/tls.py @@ -150,6 +150,7 @@ def get_pin(host: str, port: int = PLAIN_PORT) -> Optional[str]: def set_pin(host: str, sha256: str, port: int = PLAIN_PORT, subject: str = "") -> None: """Pin ``sha256`` for the pod at ``host`` (what ``benchpod lan-tls trust`` does).""" + _sessions.clear() with _lock: path = pin_store_path() data = _load(path) @@ -198,43 +199,70 @@ def _check_pin(host: str, port: int, der: bytes) -> None: # -- the handshake ------------------------------------------------------------------------------ +#: One SSLContext per CA mode (a session can only resume on the context that made it), and the +#: last session per pod address: the pod resumes TLS 1.2 sessions (by ticket or id), so later +#: connections skip its ~460 ms ECDHE + ECDSA and take ~20 ms. A resumed session proves the same +#: server as the full handshake whose certificate was pinned. +_contexts: Dict[Tuple[Optional[str], bool], ssl.SSLContext] = {} +_sessions: Dict[Tuple[str, int, Optional[str]], ssl.SSLSession] = {} + + +def _context(ca: Optional[str], verify: bool) -> ssl.SSLContext: + key = (ca if verify else None, verify) + with _lock: + ctx = _contexts.get(key) + if ctx is None: + if ca and verify: + ctx = ssl.create_default_context(cafile=ca) + else: + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE # the pin is the check + ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + _contexts[key] = ctx + return ctx + + def wrap(sock: socket.socket, host: str, port: int, *, ca: Optional[str] = None, pin: bool = True, timeout: float = 10.0) -> ssl.SSLSocket: """TLS over a connected socket to the pod at ``host:port``. With ``ca`` (a PEM file of CA certificates) the certificate chain and host name are verified. Otherwise the certificate is checked against (or, on first use, recorded as) the pin; with - ``pin=False`` any certificate is accepted (to show or trust it).""" + ``pin=False`` any certificate is accepted (to show or trust it) and no session is reused.""" is_ip = True try: ipaddress.ip_address(host) except ValueError: is_ip = False - if ca and pin: - ctx = ssl.create_default_context(cafile=ca) - check_name = True - else: - ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) - ctx.check_hostname = False - ctx.verify_mode = ssl.CERT_NONE # the pin below is the check - check_name = False - ctx.minimum_version = ssl.TLSVersion.TLSv1_2 + verify = bool(ca and pin) + ctx = _context(ca, verify) + skey = (host, port, ca if verify else None) + session = _sessions.get(skey) if pin else None old = sock.gettimeout() sock.settimeout(timeout) - tls = ctx.wrap_socket(sock, server_hostname=host if (check_name or not is_ip) else None) + tls = ctx.wrap_socket(sock, server_hostname=host if (verify or not is_ip) else None, session=session) try: - if not ca and pin: + if not verify and pin: der = tls.getpeercert(binary_form=True) if not der: raise ssl.SSLError("the pod sent no certificate") _check_pin(host, port, der) except BaseException: + _sessions.pop(skey, None) tls.close() raise + if pin and tls.session is not None: + _sessions[skey] = tls.session tls.settimeout(old) return tls +def forget_sessions() -> None: + """Drop the resumable sessions (after a pin change, say).""" + _sessions.clear() + + def trust_lan_tls(host: str, port: int = PLAIN_PORT, *, sha256: Optional[str] = None, timeout: float = 10.0) -> str: """Pin the certificate the pod at ``host`` presents now (after a deliberate change), like From 5c8c69d8d533143d88195a3642ef1c349aad1d88 Mon Sep 17 00:00:00 2001 From: Edward Viaene Date: Fri, 16 Oct 2026 11:49:18 -0500 Subject: [PATCH 3/3] embeddedci 2.13.2, embeddedci-mcp 2.13.2 (requires embeddedci 2.13.2) Release prep for the LAN TLS work; not tagged. The MCP server's lan_tls tool needs the SDK's LanTls and lan_tls methods, so its minimum SDK moves to 2.13.2 (mcp-min-sdk tests the in-repo SDK until 2.13.2 is on PyPI). Co-Authored-By: Claude Opus 5.5 --- packages/embeddedci-mcp/CHANGELOG.md | 3 ++- packages/embeddedci-mcp/pyproject.toml | 4 ++-- packages/embeddedci-mcp/server.json | 4 ++-- packages/embeddedci/CHANGELOG.md | 4 +++- packages/embeddedci/pyproject.toml | 2 +- 5 files changed, 10 insertions(+), 7 deletions(-) diff --git a/packages/embeddedci-mcp/CHANGELOG.md b/packages/embeddedci-mcp/CHANGELOG.md index 538ab22..f6eab4e 100644 --- a/packages/embeddedci-mcp/CHANGELOG.md +++ b/packages/embeddedci-mcp/CHANGELOG.md @@ -1,7 +1,8 @@ # Changelog — `embeddedci-mcp` -## Unreleased +## 2.13.2 +- Requires embeddedci 2.13.2. - New tool `lan_tls`: the pod's LAN API over TLS (firmware 3.9.3+), HTTPS-only and regenerate. `connect` to a network address uses TLS when the pod has it and pins its certificate (from the SDK). diff --git a/packages/embeddedci-mcp/pyproject.toml b/packages/embeddedci-mcp/pyproject.toml index e2143db..909caf9 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.1" +version = "2.13.2" description = "MCP server that lets AI agents drive an EmbeddedCI BenchPod hardware-in-the-loop bench" readme = "README.md" requires-python = ">=3.10" @@ -26,7 +26,7 @@ classifiers = [ dependencies = [ # cloud: the embeddedci: destination; discovery: connect("discover") over mDNS; # analysis: dominant frequency in capture summaries. - "embeddedci[cloud,discovery,analysis]>=2.13.1,<3", + "embeddedci[cloud,discovery,analysis]>=2.13.2,<3", "mcp>=1.12,<2", "anyio>=4", "pydantic>=2.7", diff --git a/packages/embeddedci-mcp/server.json b/packages/embeddedci-mcp/server.json index f7bd3d8..67ee396 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.1", + "version": "2.13.2", "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.1", + "version": "2.13.2", "runtimeHint": "uvx", "transport": { "type": "stdio" diff --git a/packages/embeddedci/CHANGELOG.md b/packages/embeddedci/CHANGELOG.md index ea3c5e8..c3be0c6 100644 --- a/packages/embeddedci/CHANGELOG.md +++ b/packages/embeddedci/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog — `embeddedci` -## Unreleased +## 2.13.2 - The LAN API over TLS (pod firmware 3.9.3+): a network address on the default port connects to 8443 with TLS when the pod has it and falls back to plaintext 8080 only when 8443 refuses and @@ -9,6 +9,8 @@ `BENCHPOD_TLS_CA` (verify against a CA instead). New: `BenchPod.lan_tls()`, `set_lan_tls_https_only()`, `regenerate_lan_tls()`, `set_lan_tls_cert()`, `LanTls`, `trust_lan_tls()`, `forget_lan_tls()`, `TcpTransport.last_tls`, `TcpTransport.command_sequence()`. + The transport keeps the last TLS session per pod, so later connections resume in ~35 ms instead + of the pod's ~0.46 s full handshake. ## 2.13.1 diff --git a/packages/embeddedci/pyproject.toml b/packages/embeddedci/pyproject.toml index 79e40fc..0f23264 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.1" +version = "2.13.2" description = "BenchPod SDK and pytest plugin for EmbeddedCI hardware-in-the-loop testing" readme = "README.md" requires-python = ">=3.10"