Add authenticated server identity discovery

README.md

@@ -338,8 +338,32 @@ sess = DtlsCoapSession("192.0.2.100", 5684, auth=auth)
 `ServerCertificateAuth` is for a server-authenticated channel that does not
 send a client certificate, such as the initial DTLS carrier used by
 manufacturer-certificate OTM. It still verifies the CA chain, exact selected
-subject role, and pinned certificate UUID. It does not learn an identity from
-the first endpoint it reaches and cannot be combined with client credentials.
+subject role, and pinned certificate UUID. It cannot be combined with client
+credentials.
+
+An explicit first-use workflow may need to authenticate the Samsung hardware
+certificate before its subject UUID is known. Use the discovery profile only
+for that bounded step:
+
+```python
+server_profile = SamsungServerProfile.discover_device(
+    additional_ca_pem=additional_samsung_ca_pem,
+)
+auth = ServerCertificateAuth(server_profile=server_profile)
+sess = DtlsCoapSession("192.0.2.100", 5684, auth=auth)
+sess.connect()
+certificate_uuid = sess.server_certificate_identity
+```
+
+The discovery profile still verifies the CA chain and the complete selected
+Samsung subject role before `connect()` exposes the non-zero certificate UUID.
+It does not trust an arbitrary first certificate, and neither the immutable
+profile nor its provider retains the learned identity. The caller must bind
+that UUID to independently authenticated device evidence, such as `/oic/d`
+read over the same authenticated session, before persisting it. Subsequent
+connections should use `bound_device()` with that verified binding. The
+certificate UUID and the OCF device UUID are separate identities and must not
+be assumed equal.
 
 This API deliberately does not discover, mint, authorize, provision, rotate,
 or persist credentials, and it performs no ownership transfer or OCF security

smartthings_local/protocol/auth.py

@@ -93,6 +93,19 @@ def __init__(
         role: SamsungServerRole = SamsungServerRole.HOME_APPLIANCE,
         additional_ca_pem: str | None = None,
     ) -> None:
+        parsed_identity = self._parse_expected_identity(
+            expected_certificate_identity
+        )
+        self._initialize(
+            expected_certificate_identity=parsed_identity,
+            role=role,
+            additional_ca_pem=additional_ca_pem,
+        )
+
+    @staticmethod
+    def _parse_expected_identity(
+        expected_certificate_identity: UUID | str,
+    ) -> UUID:
         if type(expected_certificate_identity) is UUID:
             parsed_identity = expected_certificate_identity
         elif type(expected_certificate_identity) is str:
@@ -117,6 +130,15 @@ def __init__(
                 "expected_certificate_identity must be a canonical "
                 "non-zero UUID"
             )
+        return parsed_identity
+
+    def _initialize(
+        self,
+        *,
+        expected_certificate_identity: UUID | None,
+        role: SamsungServerRole,
+        additional_ca_pem: str | None,
+    ) -> None:
         if type(role) is not SamsungServerRole:
             raise TypeError("role must be a SamsungServerRole")
 
@@ -176,7 +198,7 @@ def __init__(
         object.__setattr__(
             self,
             "_expected_certificate_identity",
-            parsed_identity,
+            expected_certificate_identity,
         )
         object.__setattr__(self, "_role", role)
         object.__setattr__(self, "_additional_ca_certificates", certificates)
@@ -202,6 +224,27 @@ def bound_device(
             additional_ca_pem=additional_ca_pem,
         )
 
+    @classmethod
+    def discover_device(
+        cls,
+        *,
+        role: SamsungServerRole = SamsungServerRole.HOME_APPLIANCE,
+        additional_ca_pem: str | None = None,
+    ) -> SamsungServerProfile:
+        """Verify a Samsung hardware leaf before learning its certificate UUID.
+
+        This profile is intended only for an explicit first-use workflow. The
+        caller must bind the returned session identity to independently
+        authenticated device evidence before persisting it.
+        """
+        profile = object.__new__(cls)
+        profile._initialize(
+            expected_certificate_identity=None,
+            role=role,
+            additional_ca_pem=additional_ca_pem,
+        )
+        return profile
+
     def __repr__(self) -> str:
         """Return a representation without device or trust-chain details."""
         return "SamsungServerProfile()"
@@ -273,6 +316,14 @@ def _verify_peer(
             return False
         if depth > 0:
             return True
+        identity = self._certificate_identity(certificate)
+        return identity is not None and (
+            self._expected_certificate_identity is None
+            or identity == self._expected_certificate_identity
+        )
+
+    def _certificate_identity(self, certificate) -> UUID | None:
+        """Extract a UUID only from the selected Samsung subject role."""
         try:
             with warnings.catch_warnings():
                 # pyOpenSSL deprecates this API in favor of cryptography, but
@@ -291,7 +342,7 @@ def _verify_peer(
             crypto.Error,
         ):
             logger.warning("Unable to parse Samsung server certificate subject")
-            return False
+            return None
 
         common_names = [value for name, value in components if name == "CN"]
         organizational_units = [
@@ -308,13 +359,12 @@ def _verify_peer(
             or organizations != ["Samsung Electronics"]
             or countries != ["KR"]
         ):
-            return False
+            return None
         match = _SAMSUNG_SERVER_CN_RE.fullmatch(common_names[0])
-        return (
-            match is not None
-            and UUID(match.group("device_identity"))
-            == self._expected_certificate_identity
-        )
+        if match is None:
+            return None
+        identity = UUID(match.group("device_identity"))
+        return identity if identity.int != 0 else None
 
 
 def _configure_certificate_server(
@@ -362,6 +412,13 @@ def configure_context(self, context: SSL.Context) -> None:
         """Verify the selected server profile without loading client material."""
         _configure_certificate_server(context, self._server_profile)
 
+    def _authenticated_server_identity(self, connection) -> UUID | None:
+        certificate = connection.get_peer_certificate()
+        identity = self._server_profile._certificate_identity(certificate)
+        if identity is None:
+            raise ValueError("verified server identity was unavailable")
+        return identity
+
 
 class CertificateAuth:
     """Certificate authentication loaded from files or in-memory PEM data.
@@ -486,6 +543,15 @@ def configure_context(self, context: SSL.Context) -> None:
             context.use_privatekey_file(self._private_key_path)
             context.check_privatekey()
 
+    def _authenticated_server_identity(self, connection) -> UUID | None:
+        if self._server_profile is None:
+            return None
+        certificate = connection.get_peer_certificate()
+        identity = self._server_profile._certificate_identity(certificate)
+        if identity is None:
+            raise ValueError("verified server identity was unavailable")
+        return identity
+
 
 class PskAuth:
     """DTLS authentication using an existing OCF PSK credential.

smartthings_local/protocol/dtls_session.py

@@ -34,10 +34,12 @@
 import time
 from collections.abc import Mapping
 from dataclasses import dataclass
+from uuid import UUID
 
 from OpenSSL import SSL
 
 from ..errors import (
+    AuthenticationError,
     BlockwiseError,
     EndpointError,
     MalformedMessageError,
@@ -462,6 +464,10 @@ def __init__(self, host, port, cert_path=None, key_path=None, *,
         self.conn = None
         self.dest = None
         self.endpoint = None
+        # Populated only when a SamsungServerProfile verified the peer. The
+        # UUID comes from the authenticated hardware-certificate subject; it
+        # is not necessarily the OCF device UUID returned by /oic/d.
+        self.server_certificate_identity = None
 
         # Terminal session shutdown has its own wake signal so
         # quiesce_for_close() can interrupt connect() even when the caller did
@@ -678,6 +684,30 @@ def connect(
             sock.close()
             raise SessionTimeoutError()
 
+        try:
+            identity_reader = getattr(
+                self.auth,
+                "_authenticated_server_identity",
+                None,
+            )
+            if identity_reader is not None:
+                server_certificate_identity = identity_reader(conn)
+                if server_certificate_identity is not None and (
+                    type(server_certificate_identity) is not UUID
+                    or server_certificate_identity.int == 0
+                ):
+                    raise ValueError(
+                        "verified server identity was unavailable"
+                    )
+            else:
+                server_certificate_identity = None
+        except Exception:
+            self._send_close_notify(conn, sock)
+            sock.close()
+            raise AuthenticationError() from ConnectionError(
+                "verified server identity was unavailable"
+            )
+
         with self._lifecycle_lock:
             if self._lifecycle_cancel.is_set():
                 sock.close()
@@ -686,6 +716,7 @@ def connect(
             self.conn = conn
             self.dest = dest
             self.endpoint = endpoint
+            self.server_certificate_identity = server_certificate_identity
             self._stop.clear()
 
     def start_reader(self):

tests/test_certificate_profiles.py

@@ -453,6 +453,56 @@ def test_profile_accepts_only_canonical_nonzero_identity():
             SamsungServerProfile.bound_device(invalid)
 
 
+def test_discovery_profile_verifies_role_before_exposing_identity():
+    first_chain = _make_generated_chain(_IDENTITY)
+    second_chain = _make_generated_chain(_OTHER_IDENTITY)
+    zero_chain = _make_generated_chain(UUID(int=0))
+    wrong_role = _make_generated_chain(
+        _IDENTITY,
+        organizational_unit="OCF VD Device",
+    )
+    profile = SamsungServerProfile.discover_device()
+
+    assert repr(profile) == "SamsungServerProfile()"
+    assert profile._verify_peer(None, first_chain.leaf, 0, 0, True) is True
+    assert profile._verify_peer(None, second_chain.leaf, 0, 0, True) is True
+    assert profile._verify_peer(None, zero_chain.leaf, 0, 0, True) is False
+    assert profile._verify_peer(None, wrong_role.leaf, 0, 0, True) is False
+
+    for invalid in ("OCF HA Device", None, object()):
+        with pytest.raises(TypeError, match="SamsungServerRole"):
+            SamsungServerProfile.discover_device(role=invalid)
+
+
+def test_discovery_profile_is_reusable_and_does_not_retain_peer_identity():
+    first_chain = _make_generated_chain(_IDENTITY)
+    second_chain = _make_generated_chain(_OTHER_IDENTITY)
+    profile = SamsungServerProfile.discover_device()
+    provider = ServerCertificateAuth(server_profile=profile)
+
+    first = provider._authenticated_server_identity(
+        SimpleNamespace(get_peer_certificate=lambda: first_chain.leaf)
+    )
+    second = provider._authenticated_server_identity(
+        SimpleNamespace(get_peer_certificate=lambda: second_chain.leaf)
+    )
+
+    assert first == _IDENTITY
+    assert second == _OTHER_IDENTITY
+    assert not hasattr(profile, "certificate_identity")
+    assert str(_IDENTITY) not in repr(profile)
+    assert str(_OTHER_IDENTITY) not in repr(profile)
+
+    certificate_provider = CertificateAuth.from_memory(
+        first_chain.certificate_pem,
+        first_chain.private_key_pem,
+        server_profile=profile,
+    )
+    assert certificate_provider._authenticated_server_identity(
+        SimpleNamespace(get_peer_certificate=lambda: second_chain.leaf)
+    ) == _OTHER_IDENTITY
+
+
 def test_profile_roles_are_explicit_and_fail_closed():
     home_chain = _make_generated_chain(_IDENTITY)
     video_chain = _make_generated_chain(
@@ -894,6 +944,7 @@ def set_options(self, options):
 
 def test_profile_identity_verification_rejects_malformed_subjects(caplog):
     profile = SamsungServerProfile.bound_device(_IDENTITY)
+    discovery_profile = SamsungServerProfile.discover_device()
     wrong_role = _make_generated_chain(
         _IDENTITY,
         organizational_unit="Unexpected Device",
@@ -934,6 +985,10 @@ def test_profile_identity_verification_rejects_malformed_subjects(caplog):
     caplog.clear()
     with caplog.at_level(logging.WARNING, logger=auth_module.__name__):
         assert profile._verify_peer(None, object(), 0, 0, True) is False
+        assert (
+            discovery_profile._verify_peer(None, object(), 0, 0, True)
+            is False
+        )
         assert (
             profile._verify_peer(
                 None,
@@ -947,6 +1002,7 @@ def test_profile_identity_verification_rejects_malformed_subjects(caplog):
     assert caplog.messages == [
         "Unable to parse Samsung server certificate subject",
         "Unable to parse Samsung server certificate subject",
+        "Unable to parse Samsung server certificate subject",
     ]
 
     caplog.clear()

tests/test_public_api_contract.py

@@ -125,6 +125,20 @@ def test_samsung_server_profile_is_public_and_explicitly_bound():
     assert parameters["additional_ca_pem"].kind is inspect.Parameter.KEYWORD_ONLY
     assert parameters["additional_ca_pem"].default is None
 
+    discovery_parameters = inspect.signature(
+        SamsungServerProfile.discover_device
+    ).parameters
+    assert list(discovery_parameters) == ["role", "additional_ca_pem"]
+    assert all(
+        parameter.kind is inspect.Parameter.KEYWORD_ONLY
+        for parameter in discovery_parameters.values()
+    )
+    assert (
+        discovery_parameters["role"].default
+        is SamsungServerRole.HOME_APPLIANCE
+    )
+    assert discovery_parameters["additional_ca_pem"].default is None
+
 
 def test_server_certificate_auth_is_a_public_authentication_provider():
     profile = SamsungServerProfile.bound_device(
@@ -139,6 +153,7 @@ def test_server_certificate_auth_is_a_public_authentication_provider():
     assert session.key_path is None
     assert session.cert_pem is None
     assert session.key_pem is None
+    assert session.server_certificate_identity is None
     parameters = inspect.signature(ServerCertificateAuth).parameters
     assert list(parameters) == ["server_profile"]
     assert parameters["server_profile"].kind is inspect.Parameter.KEYWORD_ONLY