Add bound Samsung server certificate profile

README.md

@@ -56,6 +56,47 @@ auth = CertificateAuth.from_memory(cert_pem, key_pem)
 sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth)
 ```
 
+Some newer OCF-PKI home appliances require an exact Samsung DTLS offer and
+present a hardware certificate whose subject contains the appliance's OCF
+UUID. When the caller already has an authorized client certificate and a
+previously verified appliance UUID, opt in to both requirements explicitly:
+
+```python
+from smartthings_local.protocol.auth import (
+    CertificateAuth,
+    SamsungServerProfile,
+)
+
+server_profile = SamsungServerProfile.bound_device(
+    expected_device_uuid,
+    additional_ca_pem=additional_samsung_ca_pem,
+)
+auth = CertificateAuth.from_memory(
+    cert_pem,
+    key_pem,
+    server_profile=server_profile,
+)
+sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth)
+```
+
+The profile limits the ClientHello to P-256,
+`ECDHE-ECDSA-AES128-GCM-SHA256`, and the observed SHA-256/SHA-1 RSA/ECDSA
+signature set, disables session tickets, preserves certificate-chain
+verification, and requires the exact subject role
+`C=KR, O=Samsung Electronics, OU=OCF HA Device` with a common name ending in
+the expected UUID. `additional_ca_pem` is optional and accepts only a bounded
+PEM CA-certificate chain; it is applied only to this profiled context. Without
+a profile, `CertificateAuth` retains its existing verification behavior.
+
+This API deliberately does not discover, mint, authorize, provision, rotate,
+or persist credentials, and it performs no ownership transfer or OCF security
+resource writes. In particular, an unowned vendor-OTM appliance such as the
+one discussed in [issue #20](https://github.com/QuiteYellow/SmartThings-Local/issues/20)
+does not become connectable merely by selecting this profile. The already-owned
+new-PKI case in [issue #16](https://github.com/QuiteYellow/SmartThings-Local/issues/16)
+still requires an authorized client identity before this connection profile
+can be used.
+
 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`

smartthings_local/protocol/auth.py

@@ -3,15 +3,28 @@
 from __future__ import annotations
 
 import re
+import warnings
 from os import PathLike
 from pathlib import Path
 from typing import Protocol, runtime_checkable
+from uuid import UUID
 
+from cryptography.x509.oid import ExtensionOID, NameOID
 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"
+_SAMSUNG_SERVER_CURVES = b"prime256v1"
+_SAMSUNG_SERVER_SIGNATURE_ALGORITHMS = (
+    b"RSA+SHA256:ECDSA+SHA256:RSA+SHA1:ECDSA+SHA1"
+)
+_SAMSUNG_SERVER_CN_RE = re.compile(
+    r"\AOCF Device: [^()\r\n]{1,96} "
+    r"\((?P<device_identity>[0-9a-f]{8}-[0-9a-f]{4}-"
+    r"[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\)\Z",
+    re.IGNORECASE,
+)
 _PSK_CLIENT_CALLBACK_CDEF = (
     "unsigned int (*)(SSL *, char *, char *, unsigned int, "
     "unsigned char *, unsigned int)"
@@ -53,6 +66,224 @@ def configure_context(self, context: SSL.Context) -> None:
         """Configure a context while this provider remains session-owned."""
 
 
+class SamsungServerProfile:
+    """Opt-in Samsung home-appliance server verification profile."""
+
+    __slots__ = ("_additional_ca_certificates", "_expected_device_identity")
+
+    def __init__(
+        self,
+        *,
+        expected_device_identity: UUID | str,
+        additional_ca_pem: str | None = None,
+    ) -> None:
+        if type(expected_device_identity) is UUID:
+            parsed_identity = expected_device_identity
+        elif type(expected_device_identity) is str:
+            try:
+                parsed_identity = UUID(expected_device_identity)
+            except ValueError:
+                raise ValueError(
+                    "expected_device_identity must be a canonical non-zero UUID"
+                ) from None
+            if expected_device_identity != str(parsed_identity):
+                raise ValueError(
+                    "expected_device_identity must be a canonical non-zero UUID"
+                )
+        else:
+            raise TypeError("expected_device_identity must be a UUID or string")
+        if parsed_identity.int == 0:
+            raise ValueError(
+                "expected_device_identity must be a canonical non-zero UUID"
+            )
+
+        certificates: tuple[bytes, ...] = ()
+        if additional_ca_pem is not None:
+            if type(additional_ca_pem) is not str:
+                raise TypeError("additional_ca_pem must be a string")
+            try:
+                raw_ca_pem = additional_ca_pem.encode("ascii")
+            except UnicodeEncodeError:
+                raise ValueError(
+                    "additional_ca_pem must contain ASCII PEM certificates"
+                ) from None
+            parsed_certificates = tuple(_PEM_CERT_RE.findall(raw_ca_pem))
+            if (
+                not 1 <= len(parsed_certificates) <= 4
+                or len(raw_ca_pem) > 32 * 1024
+                or _PEM_CERT_RE.sub(b"", raw_ca_pem).strip()
+            ):
+                raise ValueError(
+                    "additional_ca_pem must contain one to four PEM certificates"
+                )
+            try:
+                loaded_certificates = [
+                    crypto.load_certificate(crypto.FILETYPE_PEM, certificate)
+                    for certificate in parsed_certificates
+                ]
+                basic_constraints = [
+                    [
+                        extension
+                        for extension in certificate.to_cryptography().extensions
+                        if extension.oid == ExtensionOID.BASIC_CONSTRAINTS
+                    ]
+                    for certificate in loaded_certificates
+                ]
+            except (crypto.Error, ValueError):
+                raise ValueError(
+                    "additional_ca_pem contains an invalid certificate"
+                ) from None
+            if any(
+                len(constraints) != 1 or not constraints[0].value.ca
+                for constraints in basic_constraints
+            ):
+                raise ValueError(
+                    "additional_ca_pem must contain only CA certificates"
+                )
+            fingerprints = {
+                crypto.dump_certificate(crypto.FILETYPE_ASN1, certificate)
+                for certificate in loaded_certificates
+            }
+            if len(fingerprints) != len(loaded_certificates):
+                raise ValueError(
+                    "additional_ca_pem must not contain duplicate certificates"
+                )
+            certificates = parsed_certificates
+
+        object.__setattr__(self, "_expected_device_identity", parsed_identity)
+        object.__setattr__(self, "_additional_ca_certificates", certificates)
+
+    def __setattr__(self, _name: str, _value: object) -> None:
+        raise AttributeError("SamsungServerProfile is immutable")
+
+    def __delattr__(self, _name: str) -> None:
+        raise AttributeError("SamsungServerProfile is immutable")
+
+    @classmethod
+    def bound_device(
+        cls,
+        expected_device_identity: UUID | str,
+        *,
+        additional_ca_pem: str | None = None,
+    ) -> SamsungServerProfile:
+        """Bind a verified Samsung hardware certificate to one OCF UUID."""
+        return cls(
+            expected_device_identity=expected_device_identity,
+            additional_ca_pem=additional_ca_pem,
+        )
+
+    def __repr__(self) -> str:
+        """Return a representation without device or trust-chain details."""
+        return "SamsungServerProfile()"
+
+    def _configure_context(self, context: SSL.Context) -> None:
+        curve_setter = getattr(_util.lib, "SSL_CTX_set1_curves_list", None)
+        if curve_setter is not None:
+            if curve_setter(context._context, _SAMSUNG_SERVER_CURVES) != 1:
+                raise RuntimeError(
+                    "OpenSSL rejected the Samsung server certificate profile"
+                )
+        else:
+            # pyOpenSSL 23.1 does not expose SSL_CTX_set1_curves_list. Its
+            # public set_tmp_ecdh fallback produces the same single P-256
+            # supported-groups ClientHello extension; wire-level tests protect
+            # that compatibility path. Newer pyOpenSSL uses the exact setter.
+            try:
+                with warnings.catch_warnings():
+                    warnings.simplefilter("ignore", DeprecationWarning)
+                    curve = crypto.get_elliptic_curve("prime256v1")
+                    context.set_tmp_ecdh(curve)
+            except (AttributeError, TypeError, ValueError, SSL.Error):
+                raise RuntimeError(
+                    "OpenSSL rejected the Samsung server certificate profile"
+                ) from None
+
+        signature_setter = getattr(
+            _util.lib,
+            "SSL_CTX_set1_sigalgs_list",
+            None,
+        )
+        if (
+            signature_setter is None
+            or signature_setter(
+                context._context,
+                _SAMSUNG_SERVER_SIGNATURE_ALGORITHMS,
+            )
+            != 1
+        ):
+            raise RuntimeError(
+                "OpenSSL rejected the Samsung server certificate profile"
+            )
+        context.set_options(SSL.OP_NO_TICKET)
+
+        if self._additional_ca_certificates:
+            store = context.get_cert_store()
+            try:
+                for certificate in self._additional_ca_certificates:
+                    store.add_cert(
+                        crypto.load_certificate(
+                            crypto.FILETYPE_PEM,
+                            certificate,
+                        )
+                    )
+            except crypto.Error:
+                raise RuntimeError(
+                    "OpenSSL rejected the Samsung server trust profile"
+                ) from None
+
+    def _verify_peer(
+        self,
+        _connection,
+        certificate,
+        _error,
+        depth,
+        ok,
+    ) -> bool:
+        if not ok or certificate is None or depth < 0:
+            return False
+        if depth > 0:
+            return True
+        try:
+            components = [
+                (attribute.oid, attribute.value)
+                for relative_name in certificate.to_cryptography().subject.rdns
+                for attribute in relative_name
+            ]
+            common_names = [
+                value for name, value in components if name == NameOID.COMMON_NAME
+            ]
+            organizational_units = [
+                value
+                for name, value in components
+                if name == NameOID.ORGANIZATIONAL_UNIT_NAME
+            ]
+            organizations = [
+                value
+                for name, value in components
+                if name == NameOID.ORGANIZATION_NAME
+            ]
+            countries = [
+                value for name, value in components if name == NameOID.COUNTRY_NAME
+            ]
+            if (
+                len(common_names) != 1
+                or organizational_units != ["OCF HA Device"]
+                or organizations != ["Samsung Electronics"]
+                or countries != ["KR"]
+            ):
+                return False
+            match = _SAMSUNG_SERVER_CN_RE.fullmatch(common_names[0])
+            return (
+                match is not None
+                and UUID(match.group("device_identity"))
+                == self._expected_device_identity
+            )
+        except Exception:  # noqa: BLE001 - never cross the OpenSSL callback
+            # Verification callbacks cannot propagate Python exceptions into
+            # OpenSSL. Any malformed or unexpected subject fails closed.
+            return False
+
+
 class CertificateAuth:
     """Certificate authentication loaded from files or in-memory PEM data.
 
@@ -65,6 +296,7 @@ class CertificateAuth:
         "_certificate_pem",
         "_private_key_path",
         "_private_key_pem",
+        "_server_profile",
     )
 
     def __init__(
@@ -74,6 +306,7 @@ def __init__(
         private_key_path: str | PathLike[str] | None = None,
         certificate_pem: str | None = None,
         private_key_pem: str | None = None,
+        server_profile: SamsungServerProfile | None = None,
     ) -> None:
         file_supplied = (
             certificate_path is not None or private_key_path is not None
@@ -101,6 +334,11 @@ def __init__(
                 "must pass either certificate_path/private_key_path or "
                 "certificate_pem/private_key_pem"
             )
+        if (
+            server_profile is not None
+            and type(server_profile) is not SamsungServerProfile
+        ):
+            raise TypeError("server_profile must be a SamsungServerProfile")
         object.__setattr__(
             self,
             "_certificate_path",
@@ -113,6 +351,7 @@ def __init__(
         )
         object.__setattr__(self, "_certificate_pem", certificate_pem)
         object.__setattr__(self, "_private_key_pem", private_key_pem)
+        object.__setattr__(self, "_server_profile", server_profile)
 
     def __setattr__(self, _name: str, _value: object) -> None:
         raise AttributeError("CertificateAuth is immutable")
@@ -125,23 +364,29 @@ def from_files(
         cls,
         certificate_path: str | PathLike[str],
         private_key_path: str | PathLike[str],
+        *,
+        server_profile: SamsungServerProfile | None = None,
     ) -> CertificateAuth:
         """Create a provider backed by certificate-chain and key files."""
         return cls(
             certificate_path=certificate_path,
             private_key_path=private_key_path,
+            server_profile=server_profile,
         )
 
     @classmethod
     def from_memory(
         cls,
         certificate_pem: str,
         private_key_pem: str,
+        *,
+        server_profile: SamsungServerProfile | None = None,
     ) -> 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,
+            server_profile=server_profile,
         )
 
     def __repr__(self) -> str:
@@ -151,7 +396,14 @@ def __repr__(self) -> str:
     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)
+        if self._server_profile is None:
+            context.set_verify(SSL.VERIFY_PEER, _verify_peer)
+        else:
+            self._server_profile._configure_context(context)
+            context.set_verify(
+                SSL.VERIFY_PEER,
+                self._server_profile._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
@@ -243,4 +495,9 @@ def configure_context(self, context: SSL.Context) -> None:
         setter(context._context, self._callback)  # noqa: SLF001
 
 
-__all__ = ["AuthenticationProvider", "CertificateAuth", "PskAuth"]
+__all__ = [
+    "AuthenticationProvider",
+    "CertificateAuth",
+    "PskAuth",
+    "SamsungServerProfile",
+]