Lift certificate pinning into the connector contract

fluksio.flow.connector_tls: trust the certificate a device presented on
first contact, kept in ConnectorNode.store, and name the new fingerprint
when another one is refused. Additive, so no contract version bump; hue
and wfrac use it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ

backend/fluksio/flow/connector.py

@@ -23,7 +23,8 @@ What a connector gets from the base class:
   port rather than as the message itself;
 * :attr:`ConnectorNode.store`, for the few short strings that have to outlive
   a restart — a certificate pinned on first contact, a pairing token — shared
-  by every node of the connector's package.
+  by every node of the connector's package; :mod:`fluksio.flow.connector_tls`
+  does the pinning.
 
 The message schemas and the parameter model are the rest of the contract, and
 they are the same ones the built-in nodes use. See ``docs/connectors/`` for the

backend/fluksio/flow/connector_tls.py

@@ -0,0 +1,109 @@
+"""Trusting a device by the certificate it presented the first time, and no other.
+
+A device on the local network presents a certificate no system trusts — its
+own, or one signed by a vendor root — naming a serial or a MAC rather than the
+address it is reached at, so neither the system's CAs nor a hostname check can
+vouch for it. What can be checked is that it is still the same device. The
+certificate presented on first contact is kept in the connector's store
+(:attr:`ConnectorNode.store`), and every connection after that trusts that one
+certificate and nothing else: a different one is refused in the handshake,
+before a key or a command is sent.
+
+Fetching it on first contact sends a TLS hello and nothing more, so a device
+that is not the device learns nothing from being asked.
+
+*address* is ``host`` or ``host:port`` (443 if left out), as it goes in a URL.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import ssl
+from functools import lru_cache
+from typing import TYPE_CHECKING
+
+import httpx
+
+if TYPE_CHECKING:
+    from fluksio.flow.connector_store import ConnectorStore
+
+logger = logging.getLogger(__name__)
+
+
+class CertificateChanged(ConnectionError):
+    """The device presented a certificate other than the one pinned for it."""
+
+    def __init__(self, address: str, presented: str, node_type: str) -> None:
+        super().__init__(
+            f"{address} presented a different certificate (sha256 {presented}) from "
+            "the one pinned on first contact, so nothing was sent. If the device "
+            "was replaced or reset, clear the pin under Stored state in this "
+            "node's settings, or DELETE "
+            f"/api/v1/connectors/{node_type}/state/{pin_key(address)}."
+        )
+
+
+def pin_key(address: str) -> str:
+    """Where the certificate of the device at *address* is kept."""
+    return f"cert:{address}"
+
+
+def fingerprint(pem: str) -> str:
+    return hashlib.sha256(ssl.PEM_cert_to_DER_cert(pem)).hexdigest()
+
+
+def _presented(address: str, timeout: float) -> str:
+    """The certificate the device at *address* presents, as PEM."""
+    url = httpx.URL(f"https://{address}")
+    return ssl.get_server_certificate((url.host, url.port or 443), timeout=timeout)
+
+
+def pinned_certificate(store: ConnectorStore, address: str, timeout: float) -> str:
+    """The certificate pinned for *address*, pinning the one it presents if none is.
+
+    Blocking: it reads the store, and on first contact asks the device. A device
+    that does not complete a TLS hello raises :class:`OSError` and pins nothing.
+    """
+    pem = store.get(pin_key(address))
+    if pem is None:
+        pem = _presented(address, timeout)
+        store.set(pin_key(address), pem)
+        logger.info(
+            "pinned the certificate of %s: sha256 %s", address, fingerprint(pem)
+        )
+    return pem
+
+
+@lru_cache(maxsize=32)
+def context_for(pem: str) -> ssl.SSLContext:
+    """A context trusting this one certificate, whatever signed it."""
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    # The certificate names the device, not the address it is reached at.
+    context.check_hostname = False
+    # The pinned certificate is the trust anchor, not a chain up to a root.
+    context.verify_flags |= ssl.VERIFY_X509_PARTIAL_CHAIN
+    context.load_verify_locations(cadata=pem)
+    return context
+
+
+def raise_if_changed(
+    exc: BaseException, pem: str, address: str, timeout: float, node_type: str
+) -> None:
+    """Say so when *exc* is the handshake refusing a certificate other than *pem*.
+
+    Anything else — a device that is down, a pinned certificate that expired —
+    is left for the caller to re-raise as it was. *node_type* names the
+    connector in the error's way to clear the pin.
+    """
+    cause: BaseException | None = exc
+    while cause is not None and not isinstance(cause, ssl.SSLCertVerificationError):
+        cause = cause.__cause__ or cause.__context__
+    if cause is None:
+        return
+    try:
+        presented = fingerprint(_presented(address, timeout))
+    except (OSError, ValueError):
+        return
+    if presented != fingerprint(pem):
+        raise CertificateChanged(address, presented, node_type) from exc

docs/reference/connector-contract.md

@@ -240,6 +240,34 @@ curl -X DELETE $FLUKSIO/connectors/hue/state -H "Authorization: Bearer $TOKEN"
 So a connector has to cope with a key disappearing: clearing is how somebody
 tells it to start over.
 
+### Pinning a device's certificate
+
+A device on the local network presents a certificate no system trusts and that
+names a serial or a MAC rather than its address, so `verify=True` fails and
+`verify=False` trusts anyone. `fluksio.flow.connector_tls` trusts the
+certificate the device presented on first contact instead, kept in `self.store`
+under `cert:<address>`:
+
+```python
+from fluksio.flow.connector_tls import context_for, pinned_certificate, raise_if_changed
+
+address = f"{self.config.host}:{self.config.port}"
+pem = pinned_certificate(self.store, address, self.config.timeout)  # blocking
+try:
+    with httpx.Client(verify=context_for(pem)) as client:
+        ...
+except httpx.ConnectError as exc:
+    raise_if_changed(exc, pem, address, self.config.timeout, "my_node_type")
+    raise
+```
+
+- First contact sends a TLS hello and nothing else; a device that does not
+  complete one raises `OSError` and nothing is pinned.
+- Any other certificate is refused in the handshake, before a key or a command
+  is sent. `raise_if_changed` turns that refusal into `CertificateChanged`,
+  which names the new SHA-256 and the key to clear, under the node type given.
+- Clearing the key accepts whatever the device presents next.
+
 ## Lifecycle
 
 ```python