diff --git a/README.md b/README.md index 1e7fdff..45fa4bf 100644 --- a/README.md +++ b/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 diff --git a/smartthings_local/protocol/auth.py b/smartthings_local/protocol/auth.py index 14e87a1..2a14357 100644 --- a/smartthings_local/protocol/auth.py +++ b/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. diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index e73e8ce..b636cdb 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/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): diff --git a/tests/test_certificate_profiles.py b/tests/test_certificate_profiles.py index 91ba1fd..684b1cd 100644 --- a/tests/test_certificate_profiles.py +++ b/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() diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index 8358466..7880efc 100644 --- a/tests/test_public_api_contract.py +++ b/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 diff --git a/tests/test_session_connect_deadline.py b/tests/test_session_connect_deadline.py index c7c1f3b..8d74867 100644 --- a/tests/test_session_connect_deadline.py +++ b/tests/test_session_connect_deadline.py @@ -3,11 +3,12 @@ from __future__ import annotations import socket +from uuid import UUID import pytest from OpenSSL import SSL -from smartthings_local.errors import SessionTimeoutError +from smartthings_local.errors import AuthenticationError, SessionTimeoutError from smartthings_local.protocol import dtls_session from smartthings_local.protocol.dtls_session import DtlsCoapSession from smartthings_local.protocol.endpoint import ResolvedUdpEndpoint @@ -34,6 +35,17 @@ def configure_context(self, _context): self.clock.advance(self.configure_delay) +class _IdentityAuth(_Auth): + def __init__(self, identity): + super().__init__() + self.identity = identity + + def _authenticated_server_identity(self, _connection): + if isinstance(self.identity, Exception): + raise self.identity + return self.identity + + class _Connection: def __init__(self, outcomes=None, outputs=None, timer=None): self.outcomes = list(outcomes or ()) @@ -300,9 +312,58 @@ def test_successful_handshake_preserves_connected_session_state(monkeypatch): assert session.sock is sock assert session.endpoint is endpoint assert session.dest == endpoint.sockaddr + assert session.server_certificate_identity is None assert not sock.closed +def test_successful_profiled_handshake_publishes_server_identity(monkeypatch): + clock = _Clock() + identity = UUID(bytes=b"\xab" * 16) + connection, sock, endpoint, _open_calls = _install_handshake( + monkeypatch, + clock, + outcomes=("want-read", "success"), + inbound=(b"synthetic server flight",), + ) + session = _session(_IdentityAuth(identity)) + + session.connect(timeout=1.0) + + assert session.server_certificate_identity == identity + assert session.conn is connection + assert session.sock is sock + assert session.endpoint is endpoint + assert not sock.closed + + +@pytest.mark.parametrize( + "identity", + (ValueError("missing identity"), "not-a-uuid", UUID(int=0)), +) +def test_profiled_handshake_fails_closed_without_valid_server_identity( + monkeypatch, + identity, +): + clock = _Clock() + _connection, sock, _endpoint, _open_calls = _install_handshake( + monkeypatch, + clock, + outcomes=("want-read", "success"), + inbound=(b"synthetic server flight",), + ) + session = _session(_IdentityAuth(identity)) + + with pytest.raises(AuthenticationError) as captured: + session.connect(timeout=1.0) + + assert captured.value.code == "authentication" + assert str(identity) not in str(captured.value) + assert session.server_certificate_identity is None + assert session.conn is None + assert session.sock is None + assert sock.closed + + def test_connect_services_openssl_retransmit_timer(monkeypatch): clock = _Clock() outbound = b"\x16\xfe\xfd" + b"\x00" * 8 + b"\x00\x01x"