refactor(protocol): add certificate authentication provider

README.md

@@ -27,12 +27,16 @@ session directly:
 
 ```python
 import cbor2
+from smartthings_local.protocol.auth import CertificateAuth
 from smartthings_local.protocol.dtls_session import DtlsCoapSession
 
+auth = CertificateAuth.from_files(
+    "certs/client_fullchain.pem",
+    "certs/client.key",
+)
 sess = DtlsCoapSession(
     "192.0.2.100", 49154,
-    cert_path="certs/client_fullchain.pem",
-    key_path="certs/client.key",
+    auth=auth,
 )
 sess.connect()
 sess.start_reader()
@@ -44,12 +48,19 @@ sess.subscribe(["operational", "state", "vs", "0"],          # OBSERVE
 sess.close()
 ```
 
-If the cert/key are minted at runtime and never written to disk (e.g. inside an HA config flow), pass them in memory instead of by path:
+If the cert/key are minted at runtime and never written to disk (e.g. inside
+an HA config flow), create the provider from memory instead:
 
 ```python
-sess = DtlsCoapSession("192.0.2.100", 49154, cert_pem=cert_pem, key_pem=key_pem)
+auth = CertificateAuth.from_memory(cert_pem, key_pem)
+sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth)
 ```
 
+For compatibility, the existing `cert_path` / `key_path` and `cert_pem` /
+`key_pem` session arguments remain supported without a deprecation warning.
+They are routed through `CertificateAuth` internally. Do not combine `auth`
+with those legacy arguments.
+
 ### Classified errors
 
 Runtime transport failures use the public types in
@@ -479,6 +490,7 @@ smartthings_local/                   The installable library — `pip install sm
   __init__.py
   protocol/                          DTLS-CoAP transport (reusable by any consumer, not just MQTT)
     __init__.py
+    auth.py                          Immutable DTLS authentication providers
     coap.py                          CoAP wire protocol: message encode/decode, token handling
     dtls_session.py                  DTLS session: handshake, client-cert auth (file or in-memory PEM), Block2, liveness
     dtls_probe.py                    Stateless DTLS liveness + opt-in stateful diagnostic

smartthings_local/protocol/auth.py

@@ -0,0 +1,167 @@
+"""Immutable authentication providers for DTLS sessions."""
+
+from __future__ import annotations
+
+import re
+from os import PathLike
+from pathlib import Path
+from typing import Protocol, runtime_checkable
+
+from OpenSSL import SSL, crypto
+
+_OCF_ROOT_CA = str(Path(__file__).with_name("ocf_root_ca.pem"))
+_DTLS_CIPHERS = b"ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0"
+_PEM_CERT_RE = re.compile(
+    rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
+    re.DOTALL,
+)
+
+
+def _verify_peer(_connection, _certificate, _error, _depth, ok):
+    """Keep pyOpenSSL's existing verification result unchanged."""
+    return ok
+
+
+def _load_pem_chain(ctx: SSL.Context, cert_pem: str, key_pem: str) -> None:
+    """Load a PEM certificate chain and private key into a context in memory."""
+    certificates = _PEM_CERT_RE.findall(cert_pem.encode())
+    if not certificates:
+        raise ValueError("No certificates found in cert_pem")
+    ctx.use_certificate(
+        crypto.load_certificate(crypto.FILETYPE_PEM, certificates[0])
+    )
+    for extra in certificates[1:]:
+        ctx.add_extra_chain_cert(
+            crypto.load_certificate(crypto.FILETYPE_PEM, extra)
+        )
+    ctx.use_privatekey(
+        crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem.encode())
+    )
+    ctx.check_privatekey()
+
+
+@runtime_checkable
+class AuthenticationProvider(Protocol):
+    """Configure authentication for a newly created DTLS context."""
+
+    def configure_context(self, context: SSL.Context) -> None:
+        """Configure trust, verification, ciphers, and client credentials."""
+
+
+class CertificateAuth:
+    """Certificate authentication loaded from files or in-memory PEM data.
+
+    Use :meth:`from_files` or :meth:`from_memory` to create an instance.
+    Credential sources are intentionally not exposed as public attributes.
+    """
+
+    __slots__ = (
+        "_certificate_path",
+        "_certificate_pem",
+        "_private_key_path",
+        "_private_key_pem",
+    )
+
+    def __init__(
+        self,
+        *,
+        certificate_path: str | PathLike[str] | None = None,
+        private_key_path: str | PathLike[str] | None = None,
+        certificate_pem: str | None = None,
+        private_key_pem: str | None = None,
+    ) -> None:
+        file_supplied = (
+            certificate_path is not None or private_key_path is not None
+        )
+        memory_supplied = (
+            certificate_pem is not None or private_key_pem is not None
+        )
+        if file_supplied and memory_supplied:
+            raise ValueError(
+                "pass either certificate_path/private_key_path or "
+                "certificate_pem/private_key_pem, not both"
+            )
+        if file_supplied:
+            if certificate_path is None or private_key_path is None:
+                raise ValueError(
+                    "certificate_path and private_key_path must be passed together"
+                )
+        elif memory_supplied:
+            if certificate_pem is None or private_key_pem is None:
+                raise ValueError(
+                    "certificate_pem and private_key_pem must be passed together"
+                )
+        else:
+            raise ValueError(
+                "must pass either certificate_path/private_key_path or "
+                "certificate_pem/private_key_pem"
+            )
+        object.__setattr__(
+            self,
+            "_certificate_path",
+            str(certificate_path) if certificate_path is not None else None,
+        )
+        object.__setattr__(
+            self,
+            "_private_key_path",
+            str(private_key_path) if private_key_path is not None else None,
+        )
+        object.__setattr__(self, "_certificate_pem", certificate_pem)
+        object.__setattr__(self, "_private_key_pem", private_key_pem)
+
+    def __setattr__(self, _name: str, _value: object) -> None:
+        raise AttributeError("CertificateAuth is immutable")
+
+    def __delattr__(self, _name: str) -> None:
+        raise AttributeError("CertificateAuth is immutable")
+
+    @classmethod
+    def from_files(
+        cls,
+        certificate_path: str | PathLike[str],
+        private_key_path: str | PathLike[str],
+    ) -> CertificateAuth:
+        """Create a provider backed by certificate-chain and key files."""
+        return cls(
+            certificate_path=certificate_path,
+            private_key_path=private_key_path,
+        )
+
+    @classmethod
+    def from_memory(
+        cls,
+        certificate_pem: str,
+        private_key_pem: str,
+    ) -> CertificateAuth:
+        """Create a provider backed by an in-memory PEM chain and key."""
+        return cls(
+            certificate_pem=certificate_pem,
+            private_key_pem=private_key_pem,
+        )
+
+    def __repr__(self) -> str:
+        """Return a representation that never includes credential material."""
+        return "CertificateAuth()"
+
+    def configure_context(self, context: SSL.Context) -> None:
+        """Apply the existing certificate authentication profile to a context."""
+        context.load_verify_locations(_OCF_ROOT_CA)
+        context.set_verify(SSL.VERIFY_PEER, _verify_peer)
+        # @SECLEVEL=0 permits SHA-1 in Samsung's server cert chain (AC14K_M
+        # intermediate is SHA-1 signed). This is the only channel that reaches
+        # the OpenSSL instance cryptography bundles; ctypes and cffi bindings
+        # do not expose SSL_CTX_set_security_level on this build.
+        context.set_cipher_list(_DTLS_CIPHERS)
+        if self._certificate_pem is not None:
+            _load_pem_chain(
+                context,
+                self._certificate_pem,
+                self._private_key_pem,
+            )
+        else:
+            context.use_certificate_chain_file(self._certificate_path)
+            context.use_privatekey_file(self._private_key_path)
+            context.check_privatekey()
+
+
+__all__ = ["AuthenticationProvider", "CertificateAuth"]

smartthings_local/protocol/dtls_session.py

@@ -19,11 +19,9 @@
 delivered via the on_notification callback.
 """
 import os
-import re as _re
 import socket
 import threading
 import time
-from pathlib import Path
 
 from OpenSSL import SSL
 
@@ -42,15 +40,18 @@
     encode_options, parse_coap, build_coap, block_value, fmt_code,
     split_dtls as _split_dtls,
 )
+from .auth import (
+    AuthenticationProvider,
+    CertificateAuth,
+    _DTLS_CIPHERS,
+    _OCF_ROOT_CA,
+    _load_pem_chain,
+)
 from .endpoint import open_connected_udp_socket
 import logging
 
 logger = logging.getLogger(__name__)
 
-_OCF_ROOT_CA = str(Path(__file__).parent / 'ocf_root_ca.pem')
-_DTLS_CIPHERS = b'ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0'
-
-
 # Diagnostic logging — when DEBUG_BRIDGE=1 in env, the bridge dumps
 # every received CoAP frame, every /operational/state/vs/0 + /oven/vs/0
 # + /power/vs/0 + /mode/vs/0-options rep change, the full link tree at
@@ -71,32 +72,6 @@
 # once the ceiling is measured empirically.
 _DEFAULT_RATE_LIMIT_RPS = 5.0
 
-_PEM_CERT_RE = _re.compile(
-    rb'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----',
-    _re.DOTALL,
-)
-
-
-def _load_pem_chain(ctx: SSL.Context, cert_pem: str, key_pem: str) -> None:
-    """Load a PEM cert chain and private key into an SSL context in memory.
-
-    Parses all certificate blocks from cert_pem: the first is the leaf
-    (use_certificate), the rest are intermediates (add_extra_chain_cert).
-    No temp files are written.
-    """
-    from OpenSSL import crypto
-    certs = _PEM_CERT_RE.findall(cert_pem.encode())
-    if not certs:
-        raise ValueError("No certificates found in cert_pem")
-    ctx.use_certificate(crypto.load_certificate(crypto.FILETYPE_PEM, certs[0]))
-    for extra in certs[1:]:
-        ctx.add_extra_chain_cert(
-            crypto.load_certificate(crypto.FILETYPE_PEM, extra)
-        )
-    ctx.use_privatekey(crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem.encode()))
-    ctx.check_privatekey()
-
-
 class DtlsCoapSession:
     """Single sustained DTLS-CoAP session.
 
@@ -109,10 +84,9 @@ class DtlsCoapSession:
         code, _    = sess.post(['mode','vs','0'], cbor)
         sess.close()
 
-    Cert material comes from either a file pair (cert_path, key_path) or
-    an in-memory PEM pair (cert_pem, key_pem) — exactly one pair required.
-    The in-memory path exists for callers (e.g. an HA config flow) that
-    mint a client cert at runtime and never write it to disk.
+    Authentication comes from an immutable provider. For compatibility,
+    cert_path/key_path and cert_pem/key_pem create a CertificateAuth provider
+    internally — exactly one legacy pair is required when auth is omitted.
     """
 
     HANDSHAKE_TIMEOUT_S = 12.0
@@ -123,24 +97,37 @@ def __init__(self, host, port, cert_path=None, key_path=None, *,
                  cert_pem=None, key_pem=None,
                  on_notification=None, mtu=1200,
                  rate_limit_rps: float = _DEFAULT_RATE_LIMIT_RPS,
-                 local_port=None, family=socket.AF_UNSPEC):
-        if (cert_path is not None or key_path is not None) and \
-                (cert_pem is not None or key_pem is not None):
+                 local_port=None, family=socket.AF_UNSPEC,
+                 auth: AuthenticationProvider | None = None):
+        file_supplied = cert_path is not None or key_path is not None
+        memory_supplied = cert_pem is not None or key_pem is not None
+        if auth is not None and (file_supplied or memory_supplied):
+            raise ValueError(
+                "pass auth or legacy certificate arguments, not both")
+        if auth is None and file_supplied and memory_supplied:
             raise ValueError(
                 "pass either cert_path/key_path or cert_pem/key_pem, not both")
-        if cert_pem is not None or key_pem is not None:
+        if auth is None and memory_supplied:
             if cert_pem is None or key_pem is None:
                 raise ValueError("cert_pem and key_pem must be passed together")
-        elif cert_path is None or key_path is None:
+        elif auth is None and (cert_path is None or key_path is None):
             raise ValueError(
                 "must pass either cert_path/key_path or cert_pem/key_pem")
+        if auth is not None and not isinstance(auth, AuthenticationProvider):
+            raise TypeError("auth must implement AuthenticationProvider")
 
         self.host = host
         self.port = port
         self.cert_path = str(cert_path) if cert_path is not None else None
         self.key_path  = str(key_path) if key_path is not None else None
         self.cert_pem = cert_pem
         self.key_pem  = key_pem
+        if auth is None:
+            if cert_pem is not None:
+                auth = CertificateAuth.from_memory(cert_pem, key_pem)
+            else:
+                auth = CertificateAuth.from_files(self.cert_path, self.key_path)
+        self.auth = auth
         self.on_notification = on_notification  # fn(href, payload_bytes)
         self.mtu = mtu
         self._min_req_interval = 1.0 / rate_limit_rps
@@ -196,20 +183,7 @@ def connect(self):
         """DTLS handshake. Blocks up to HANDSHAKE_TIMEOUT_S. Raises
         ConnectionError / TimeoutError on failure."""
         ctx = SSL.Context(SSL.DTLS_METHOD)
-
-        ctx.load_verify_locations(_OCF_ROOT_CA)
-        ctx.set_verify(SSL.VERIFY_PEER, lambda conn, cert, err, depth, ok: ok)
-        # @SECLEVEL=0 permits SHA-1 in Samsung's server cert chain (AC14K_M
-        # intermediate is SHA-1 signed). This is the only channel that reaches
-        # the OpenSSL instance cryptography bundles — ctypes and cffi bindings
-        # do not expose SSL_CTX_set_security_level on this build.
-        ctx.set_cipher_list(_DTLS_CIPHERS)
-        if self.cert_pem is not None:
-            _load_pem_chain(ctx, self.cert_pem, self.key_pem)
-        else:
-            ctx.use_certificate_chain_file(self.cert_path)
-            ctx.use_privatekey_file(self.key_path)
-            ctx.check_privatekey()
+        self.auth.configure_context(ctx)
 
         conn = SSL.Connection(ctx, None)
         conn.set_connect_state()