feat(protocol): add PSK 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,34 @@ 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
+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.
+
+An existing OCF PSK credential can be supplied through `PskAuth`:
 
 ```python
-sess = DtlsCoapSession("192.0.2.100", 49154, cert_pem=cert_pem, key_pem=key_pem)
+from smartthings_local.protocol.auth import PskAuth
+
+auth = PskAuth(identity=psk_identity, key=psk_key)
+sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth)
 ```
 
+The identity must be the raw 16-byte OCF UUID and cannot contain a NUL byte;
+the key must be exactly 16 or 32 bytes. `PskAuth` selects only
+`ECDHE-PSK-AES128-CBC-SHA256` and does not acquire, derive, provision, rotate,
+or persist credentials. Ownership transfer and credential discovery are
+outside this package.
+
 ### Classified errors
 
 Runtime transport failures use the public types in
@@ -479,6 +505,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,246 @@
+"""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, _util, crypto
+
+_OCF_ROOT_CA = str(Path(__file__).with_name("ocf_root_ca.pem"))
+_DTLS_CIPHERS = b"ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0"
+_DTLS_PSK_CIPHERS = b"ECDHE-PSK-AES128-CBC-SHA256:@SECLEVEL=0"
+_PSK_CLIENT_CALLBACK_CDEF = (
+    "unsigned int (*)(SSL *, char *, char *, unsigned int, "
+    "unsigned char *, unsigned int)"
+)
+_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 a context while this provider remains session-owned."""
+
+
+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()
+
+
+class PskAuth:
+    """DTLS authentication using an existing OCF PSK credential.
+
+    The identity must be a raw 16-byte OCF UUID. The key must contain 16 or
+    32 bytes. Credential material is intentionally not exposed as public
+    attributes and is never included in this provider's representation. A
+    configured context must not outlive this provider; ``DtlsCoapSession``
+    enforces that lifetime by retaining its provider.
+    """
+
+    __slots__ = ("_callback",)
+
+    def __init__(self, *, identity: bytes, key: bytes) -> None:
+        if type(identity) is not bytes or type(key) is not bytes:
+            raise TypeError("identity and key must be bytes")
+        if len(identity) != 16:
+            raise ValueError("identity must be a raw 16-byte OCF UUID")
+        if b"\x00" in identity:
+            raise ValueError("identity cannot contain a NUL byte")
+        if len(key) not in (16, 32):
+            raise ValueError("key must be 16 or 32 bytes")
+
+        ffi = _util.ffi
+
+        @ffi.callback(_PSK_CLIENT_CALLBACK_CDEF)
+        def client_callback(
+            _ssl,
+            _identity_hint,
+            identity_buffer,
+            max_identity_length,
+            key_buffer,
+            max_key_length,
+        ):
+            # OpenSSL callbacks cannot propagate Python exceptions. Fail
+            # before touching either destination when a buffer is unavailable
+            # or too small for the complete credential.
+            if (
+                identity_buffer == ffi.NULL
+                or key_buffer == ffi.NULL
+                or len(identity) + 1 > max_identity_length
+                or len(key) > max_key_length
+            ):
+                return 0
+            ffi.memmove(
+                identity_buffer,
+                identity + b"\x00",
+                len(identity) + 1,
+            )
+            ffi.memmove(key_buffer, key, len(key))
+            return len(key)
+
+        object.__setattr__(self, "_callback", client_callback)
+
+    def __setattr__(self, _name: str, _value: object) -> None:
+        raise AttributeError("PskAuth is immutable")
+
+    def __delattr__(self, _name: str) -> None:
+        raise AttributeError("PskAuth is immutable")
+
+    def __repr__(self) -> str:
+        """Return a representation that never includes credential material."""
+        return "PskAuth()"
+
+    def configure_context(self, context: SSL.Context) -> None:
+        """Configure one context for the narrow Samsung OCF PSK profile."""
+        setter = getattr(_util.lib, "SSL_CTX_set_psk_client_callback", None)
+        if setter is None:
+            raise RuntimeError(
+                "the installed OpenSSL binding does not support DTLS PSK"
+            )
+        context.set_cipher_list(_DTLS_PSK_CIPHERS)
+        setter(context._context, self._callback)  # noqa: SLF001
+
+
+__all__ = ["AuthenticationProvider", "CertificateAuth", "PskAuth"]