Do not check a pinned certificate's dates

A pin trusts one exact certificate, so an expired one or an engine clock
that is off no longer locks a device out (X509_V_FLAG_NO_CHECK_TIME).

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

backend/fluksio/flow/connector_tls.py

@@ -83,6 +83,10 @@ def context_for(pem: str) -> ssl.SSLContext:
     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
+    # X509_V_FLAG_NO_CHECK_TIME, which `ssl` has no name for. A pin trusts one
+    # exact certificate, so its dates add nothing, and a device's certificate
+    # that expired, or a clock that is off, would otherwise lock it out.
+    context.verify_flags |= 0x200000
     context.load_verify_locations(cadata=pem)
     return context
 

backend/tests/flow/test_connector_tls.py

@@ -0,0 +1,110 @@
+"""A pinned certificate is trusted for what it is, not for its dates."""
+
+import datetime
+import socket
+import ssl
+import threading
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.x509.oid import NameOID
+
+from fluksio.flow.connector_tls import (
+    CertificateChanged,
+    context_for,
+    pin_key,
+    pinned_certificate,
+    raise_if_changed,
+)
+
+
+def _expired(name: str) -> tuple[ec.EllipticCurvePrivateKey, str]:
+    """A self-signed certificate that expired a month ago."""
+    key = ec.generate_private_key(ec.SECP256R1())
+    now = datetime.datetime.now(datetime.UTC)
+    subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)])
+    certificate = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(subject)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(now - datetime.timedelta(days=60))
+        .not_valid_after(now - datetime.timedelta(days=30))
+        .sign(key, hashes.SHA256())
+    )
+    return key, certificate.public_bytes(serialization.Encoding.PEM).decode()
+
+
+DEVICE = _expired("00005e005301")
+IMPOSTOR = _expired("00005e005302")
+
+
+class Store:
+    def __init__(self) -> None:
+        self.values: dict[str, str] = {}
+
+    def get(self, key: str) -> str | None:
+        return self.values.get(key)
+
+    def set(self, key: str, value: str) -> None:
+        self.values[key] = value
+
+
+@pytest.fixture
+def device(tmp_path: Path) -> Iterator[str]:
+    """A TLS server presenting the expired certificate; yields its address."""
+    key, pem = DEVICE
+    (tmp_path / "cert.pem").write_text(pem)
+    (tmp_path / "key.pem").write_bytes(
+        key.private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        )
+    )
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.load_cert_chain(tmp_path / "cert.pem", tmp_path / "key.pem")
+    listener = socket.create_server(("127.0.0.1", 0))
+
+    def serve() -> None:
+        while True:
+            try:
+                connection, _ = listener.accept()
+            except OSError:
+                return
+            try:
+                context.wrap_socket(connection, server_side=True).close()
+            except OSError:
+                connection.close()
+
+    threading.Thread(target=serve, daemon=True).start()
+    yield f"127.0.0.1:{listener.getsockname()[1]}"
+    listener.close()
+
+
+def _handshake(address: str, pem: str) -> None:
+    host, port = address.split(":")
+    with socket.create_connection((host, int(port)), timeout=5) as raw:
+        context_for(pem).wrap_socket(raw, server_hostname=host).close()
+
+
+def test_an_expired_pinned_certificate_is_still_trusted(device: str) -> None:
+    store = Store()
+    pem = pinned_certificate(store, device, 5.0)
+
+    assert store.get(pin_key(device)) == DEVICE[1]
+    _handshake(device, pem)  # raises if the dates were checked
+
+
+def test_another_certificate_is_still_refused(device: str) -> None:
+    pinned = IMPOSTOR[1]
+    with pytest.raises(ssl.SSLCertVerificationError) as refused:
+        _handshake(device, pinned)
+
+    with pytest.raises(CertificateChanged, match="/connectors/test/state/cert:"):
+        raise_if_changed(refused.value, pinned, device, 5.0, "test")

docs/reference/connector-contract.md

@@ -266,6 +266,9 @@ except httpx.ConnectError as exc:
 - 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.
+- The pinned certificate's dates are not checked: a pin trusts that exact
+  certificate, so one that expired, or an engine clock that is off, does not
+  lock the device out.
 - Clearing the key accepts whatever the device presents next.
 
 ## Lifecycle