Merge pull request #31 from Moballo-LLC/codex/py-05-certificate-auth

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",
+    auth=auth,
-    key_path="certs/client.key",
 )
 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 @@ on a per-token Event the reader signals. OBSERVE notifications are
 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 @@ from .coap import (
     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 @@ _BLOCK_ACK_TIMEOUT  = 4.0
 # 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
+    Authentication comes from an immutable provider. For compatibility,
-    an in-memory PEM pair (cert_pem, key_pem) — exactly one pair required.
+    cert_path/key_path and cert_pem/key_pem create a CertificateAuth provider
-    The in-memory path exists for callers (e.g. an HA config flow) that
+    internally — exactly one legacy pair is required when auth is omitted.
-    mint a client cert at runtime and never write it to disk.
     """
 
     HANDSHAKE_TIMEOUT_S = 12.0
@@ -123,17 +97,24 @@ class DtlsCoapSession:
                  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):
+                 local_port=None, family=socket.AF_UNSPEC,
-        if (cert_path is not None or key_path is not None) and \
+                 auth: AuthenticationProvider | None = None):
-                (cert_pem is not None or key_pem is not 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
@@ -141,6 +122,12 @@ class DtlsCoapSession:
         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 @@ class DtlsCoapSession:
         """DTLS handshake. Blocks up to HANDSHAKE_TIMEOUT_S. Raises
         ConnectionError / TimeoutError on failure."""
         ctx = SSL.Context(SSL.DTLS_METHOD)
-
+        self.auth.configure_context(ctx)
-        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()
 
         conn = SSL.Connection(ctx, None)
         conn.set_connect_state()

tests/test_dtls_session_cert_loading.py

@@ -1,31 +1,69 @@
+import gc
+import traceback
+import weakref
+from dataclasses import asdict
+from datetime import datetime, timedelta, timezone
+
 import pytest
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
 from OpenSSL import SSL, crypto
 
+from smartthings_local.protocol.auth import CertificateAuth
 from smartthings_local.protocol.dtls_session import DtlsCoapSession, _load_pem_chain
 
 
-def _make_self_signed_pem_pair():
+def _make_generated_pem_chain():
-    """A throwaway self-signed cert + key, just to exercise PEM loading —
+    """Create a throwaway leaf + root chain unrelated to Samsung devices."""
-    not meant to resemble a real Samsung client cert."""
+    now = datetime.now(timezone.utc)
-    key = crypto.PKey()
+    root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
-    key.generate_key(crypto.TYPE_RSA, 2048)
+    root_name = x509.Name(
+        [x509.NameAttribute(NameOID.COMMON_NAME, "Synthetic test root")]
+    )
+    root_cert = (
+        x509.CertificateBuilder()
+        .subject_name(root_name)
+        .issuer_name(root_name)
+        .public_key(root_key.public_key())
+        .serial_number(1)
+        .not_valid_before(now - timedelta(minutes=1))
+        .not_valid_after(now + timedelta(hours=1))
+        .add_extension(x509.BasicConstraints(ca=True, path_length=None), True)
+        .sign(root_key, hashes.SHA256())
+    )
 
-    cert = crypto.X509()
+    leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
-    cert.get_subject().CN = "test"
+    leaf_name = x509.Name(
-    cert.set_serial_number(1)
+        [x509.NameAttribute(NameOID.COMMON_NAME, "Synthetic test client")]
-    cert.gmtime_adj_notBefore(0)
+    )
-    cert.gmtime_adj_notAfter(3600)
+    leaf_cert = (
-    cert.set_issuer(cert.get_subject())
+        x509.CertificateBuilder()
-    cert.set_pubkey(key)
+        .subject_name(leaf_name)
-    cert.sign(key, "sha256")
+        .issuer_name(root_name)
+        .public_key(leaf_key.public_key())
+        .serial_number(2)
+        .not_valid_before(now - timedelta(minutes=1))
+        .not_valid_after(now + timedelta(hours=1))
+        .add_extension(x509.BasicConstraints(ca=False, path_length=None), True)
+        .sign(root_key, hashes.SHA256())
+    )
 
-    cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode()
+    cert_pem = (
-    key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, key).decode()
+        leaf_cert.public_bytes(serialization.Encoding.PEM)
+        + root_cert.public_bytes(serialization.Encoding.PEM)
+    ).decode()
+    key_pem = leaf_key.private_bytes(
+        serialization.Encoding.PEM,
+        serialization.PrivateFormat.PKCS8,
+        serialization.NoEncryption(),
+    ).decode()
     return cert_pem, key_pem
 
 
 def test_load_pem_chain_loads_cert_and_key_in_memory():
-    cert_pem, key_pem = _make_self_signed_pem_pair()
+    cert_pem, key_pem = _make_generated_pem_chain()
     ctx = SSL.Context(SSL.DTLS_METHOD)
     _load_pem_chain(ctx, cert_pem, key_pem)
     ctx.check_privatekey()  # raises if cert/key don't match
@@ -37,7 +75,7 @@ def test_load_pem_chain_rejects_cert_pem_with_no_certificates():
 
 
 def test_session_requires_exactly_one_cert_source():
-    cert_pem, key_pem = _make_self_signed_pem_pair()
+    cert_pem, key_pem = _make_generated_pem_chain()
 
     with pytest.raises(ValueError):
         DtlsCoapSession("host", 1234)  # neither pair given
@@ -49,10 +87,226 @@ def test_session_requires_exactly_one_cert_source():
     with pytest.raises(ValueError):
         DtlsCoapSession("host", 1234, cert_pem=cert_pem)  # key_pem missing
 
+    with pytest.raises(ValueError):
+        DtlsCoapSession("host", 1234, cert_path="/a")  # key_path missing
+
+
+def test_session_rejects_provider_with_legacy_certificate_arguments():
+    cert_pem, key_pem = _make_generated_pem_chain()
+    auth = CertificateAuth.from_memory(cert_pem, key_pem)
+
+    with pytest.raises(ValueError, match="auth or legacy certificate"):
+        DtlsCoapSession(
+            "host",
+            1234,
+            cert_pem=cert_pem,
+            key_pem=key_pem,
+            auth=auth,
+        )
+
+
+def test_session_rejects_object_that_is_not_an_authentication_provider():
+    with pytest.raises(TypeError, match="AuthenticationProvider"):
+        DtlsCoapSession("host", 1234, auth=object())
+
+
+def test_session_accepts_explicit_certificate_provider():
+    cert_pem, key_pem = _make_generated_pem_chain()
+    auth = CertificateAuth.from_memory(cert_pem, key_pem)
+    session = DtlsCoapSession("host", 1234, auth=auth)
+
+    assert session.auth is auth
+    assert session.cert_path is None
+    assert session.key_path is None
+    assert session.cert_pem is None
+    assert session.key_pem is None
+
+
+def test_session_retains_authentication_provider_for_its_lifetime():
+    class RetainedProvider:
+        def configure_context(self, _context):
+            return None
+
+    auth = RetainedProvider()
+    reference = weakref.ref(auth)
+    session = DtlsCoapSession("host", 1234, auth=auth)
+
+    del auth
+    gc.collect()
+    assert reference() is session.auth
+
 
 def test_session_accepts_pem_pair():
-    cert_pem, key_pem = _make_self_signed_pem_pair()
+    cert_pem, key_pem = _make_generated_pem_chain()
     sess = DtlsCoapSession("host", 1234, cert_pem=cert_pem, key_pem=key_pem)
     assert sess.cert_path is None
     assert sess.key_path is None
     assert sess.cert_pem == cert_pem
+    assert isinstance(sess.auth, CertificateAuth)
+
+
+def test_session_routes_legacy_file_pair_through_certificate_auth(tmp_path):
+    cert_path = tmp_path / "client.pem"
+    key_path = tmp_path / "client-key.pem"
+    session = DtlsCoapSession(
+        "host",
+        1234,
+        cert_path=cert_path,
+        key_path=key_path,
+    )
+
+    assert isinstance(session.auth, CertificateAuth)
+    assert session.cert_path == str(cert_path)
+    assert session.key_path == str(key_path)
+
+
+def test_certificate_auth_loads_generated_chain_from_memory_and_files(tmp_path):
+    cert_pem, key_pem = _make_generated_pem_chain()
+
+    memory_context = SSL.Context(SSL.DTLS_METHOD)
+    CertificateAuth.from_memory(cert_pem, key_pem).configure_context(
+        memory_context
+    )
+    memory_context.check_privatekey()
+
+    cert_path = tmp_path / "client.pem"
+    key_path = tmp_path / "client-key.pem"
+    cert_path.write_text(cert_pem)
+    key_path.write_text(key_pem)
+    file_context = SSL.Context(SSL.DTLS_METHOD)
+    CertificateAuth.from_files(cert_path, key_path).configure_context(file_context)
+    file_context.check_privatekey()
+
+
+def test_certificate_auth_rejects_invalid_memory_material():
+    auth = CertificateAuth.from_memory("not a certificate", "not a key")
+    with pytest.raises(ValueError, match="No certificates found"):
+        auth.configure_context(SSL.Context(SSL.DTLS_METHOD))
+
+
+def test_invalid_certificate_error_does_not_include_credential_material():
+    marker = "credential" + "-marker"
+    certificate_blob = (
+        "-----BEGIN CERTIFICATE-----\n"
+        f"{marker}\n"
+        "-----END CERTIFICATE-----\n"
+    )
+    key_blob = "invalid-" + marker
+    auth = CertificateAuth.from_memory(certificate_blob, key_blob)
+
+    with pytest.raises(crypto.Error) as captured:
+        auth.configure_context(SSL.Context(SSL.DTLS_METHOD))
+    rendered = (
+        str(captured.value)
+        + repr(captured.value)
+        + "".join(traceback.format_exception(captured.value))
+    )
+    assert marker not in rendered
+
+
+def test_certificate_auth_rejects_invalid_file_material(tmp_path):
+    cert_path = tmp_path / "invalid.pem"
+    key_path = tmp_path / "invalid-key.pem"
+    cert_path.write_text("invalid")
+    key_path.write_text("invalid")
+
+    with pytest.raises(SSL.Error):
+        CertificateAuth.from_files(cert_path, key_path).configure_context(
+            SSL.Context(SSL.DTLS_METHOD)
+        )
+
+
+def test_certificate_auth_rejects_incomplete_or_mixed_sources():
+    with pytest.raises(ValueError):
+        CertificateAuth()
+    with pytest.raises(ValueError):
+        CertificateAuth(certificate_path="/synthetic/client.pem")
+    with pytest.raises(ValueError):
+        CertificateAuth(certificate_pem="certificate")
+
+    certificate_path = "/synthetic/client.pem"
+    key_path = "/synthetic/client-key.pem"
+    certificate_data = "certificate"
+    key_data = "key"
+    with pytest.raises(ValueError):
+        CertificateAuth(
+            certificate_path=certificate_path,
+            private_key_path=key_path,
+            certificate_pem=certificate_data,
+            private_key_pem=key_data,
+        )
+
+
+def test_certificate_auth_is_immutable_and_has_secret_safe_repr():
+    cert_pem, key_pem = _make_generated_pem_chain()
+    auth = CertificateAuth.from_memory(cert_pem, key_pem)
+
+    rendered = repr(auth)
+    assert rendered == "CertificateAuth()"
+    assert cert_pem not in rendered
+    assert key_pem not in rendered
+    with pytest.raises(AttributeError, match="immutable"):
+        auth.certificate_pem = None
+    with pytest.raises(AttributeError, match="immutable"):
+        del auth._certificate_pem
+
+
+def test_certificate_auth_has_no_public_or_dataclass_credential_surface():
+    cert_pem, key_pem = _make_generated_pem_chain()
+    auth = CertificateAuth.from_memory(cert_pem, key_pem)
+
+    assert not hasattr(auth, "certificate_pem")
+    assert not hasattr(auth, "private_key_pem")
+    with pytest.raises(TypeError):
+        vars(auth)
+    with pytest.raises(TypeError):
+        asdict(auth)
+
+
+def test_certificate_auth_context_setup_matches_legacy_happy_path():
+    class RecordingContext:
+        def __init__(self):
+            self.calls = []
+            self.verify_callback = None
+
+        def load_verify_locations(self, path):
+            self.calls.append(("load_verify_locations", path))
+
+        def set_verify(self, mode, callback):
+            self.calls.append(("set_verify", mode))
+            self.verify_callback = callback
+
+        def set_cipher_list(self, ciphers):
+            self.calls.append(("set_cipher_list", ciphers))
+
+        def use_certificate_chain_file(self, path):
+            self.calls.append(("use_certificate_chain_file", path))
+
+        def use_privatekey_file(self, path):
+            self.calls.append(("use_privatekey_file", path))
+
+        def check_privatekey(self):
+            self.calls.append(("check_privatekey",))
+
+    context = RecordingContext()
+    CertificateAuth.from_files(
+        "/synthetic/client.pem",
+        "/synthetic/client-key.pem",
+    ).configure_context(context)
+
+    assert [call[0] for call in context.calls] == [
+        "load_verify_locations",
+        "set_verify",
+        "set_cipher_list",
+        "use_certificate_chain_file",
+        "use_privatekey_file",
+        "check_privatekey",
+    ]
+    assert context.calls[1] == ("set_verify", SSL.VERIFY_PEER)
+    assert context.calls[2] == (
+        "set_cipher_list",
+        b"ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0",
+    )
+    callback = context.verify_callback
+    assert callback(None, None, 0, 0, True) is True
+    assert callback(None, None, 1, 0, False) is False

tests/test_public_api_contract.py

@@ -6,6 +6,7 @@ import inspect
 
 from smartthings_local.ocf.observe_refresh import ObserveRefreshTask
 from smartthings_local.ocf.state_cache import StateCache
+from smartthings_local.protocol.auth import AuthenticationProvider, CertificateAuth
 from smartthings_local.protocol.dtls_session import DtlsCoapSession
 
 
@@ -40,6 +41,14 @@ def test_dtls_session_constructor_keeps_file_memory_and_local_port_inputs():
             "local_port",
         ],
     )
+    auth_parameter = inspect.signature(DtlsCoapSession).parameters["auth"]
+    assert auth_parameter.kind is inspect.Parameter.KEYWORD_ONLY
+    assert auth_parameter.default is None
+
+
+def test_certificate_auth_is_a_public_authentication_provider():
+    provider = CertificateAuth.from_files("/synthetic/cert.pem", "/synthetic/key")
+    assert isinstance(provider, AuthenticationProvider)
 
 
 def test_dtls_session_keeps_current_consumer_methods():