From 41389058814d42a8d83d1259d42ea7e49f5e1d4d Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Sat, 8 Aug 2026 16:34:42 -0700 Subject: [PATCH] feat(protocol): add bound server certificate profile --- README.md | 41 ++ smartthings_local/protocol/auth.py | 261 ++++++++++++- tests/test_certificate_profiles.py | 588 +++++++++++++++++++++++++++++ tests/test_public_api_contract.py | 14 + 4 files changed, 902 insertions(+), 2 deletions(-) create mode 100644 tests/test_certificate_profiles.py diff --git a/README.md b/README.md index 52c2f41..c2ace8c 100644 --- a/README.md +++ b/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` diff --git a/smartthings_local/protocol/auth.py b/smartthings_local/protocol/auth.py index 0e99712..e0fab08 100644 --- a/smartthings_local/protocol/auth.py +++ b/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[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,11 +364,14 @@ 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 @@ -137,11 +379,14 @@ 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", +] diff --git a/tests/test_certificate_profiles.py b/tests/test_certificate_profiles.py new file mode 100644 index 0000000..5114f2b --- /dev/null +++ b/tests/test_certificate_profiles.py @@ -0,0 +1,588 @@ +"""Synthetic security and wire-contract tests for certificate profiles.""" + +from __future__ import annotations + +from dataclasses import asdict +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from uuid import UUID + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID +from OpenSSL import SSL, crypto + +import smartthings_local.protocol.auth as auth_module +from smartthings_local.protocol.auth import ( + CertificateAuth, + SamsungServerProfile, +) + +_IDENTITY = UUID(bytes=b"\xab" * 16) +_OTHER_IDENTITY = UUID(bytes=b"\xcd" * 16) + + +def _build_certificate( + *, + subject: x509.Name, + issuer: x509.Name, + public_key, + issuer_key, + serial_number: int, + is_ca: bool | None, +) -> x509.Certificate: + now = datetime.now(UTC) + builder = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(public_key) + .serial_number(serial_number) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(hours=1)) + ) + if is_ca is not None: + builder = builder.add_extension( + x509.BasicConstraints(ca=is_ca, path_length=None), + critical=True, + ).add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=is_ca, + crl_sign=is_ca, + encipher_only=None, + decipher_only=None, + ), + critical=True, + ) + if is_ca is False: + builder = builder.add_extension( + x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), + critical=False, + ) + return builder.sign(issuer_key, hashes.SHA256()) + + +def _make_generated_chain( + identity: UUID, + *, + organizational_unit: str = "OCF HA Device", + intermediate_has_constraints: bool = True, +): + """Create a throwaway three-level chain unrelated to real devices.""" + root_key = ec.generate_private_key(ec.SECP256R1()) + root_name = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, "Synthetic profile root")] + ) + root = _build_certificate( + subject=root_name, + issuer=root_name, + public_key=root_key.public_key(), + issuer_key=root_key, + serial_number=101, + is_ca=True, + ) + + intermediate_key = ec.generate_private_key(ec.SECP256R1()) + intermediate_name = x509.Name( + [ + x509.NameAttribute( + NameOID.COMMON_NAME, + "Synthetic profile intermediate", + ) + ] + ) + intermediate = _build_certificate( + subject=intermediate_name, + issuer=root_name, + public_key=intermediate_key.public_key(), + issuer_key=root_key, + serial_number=102, + is_ca=True if intermediate_has_constraints else None, + ) + + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf_name = x509.Name( + [ + x509.NameAttribute(NameOID.COUNTRY_NAME, "KR"), + x509.NameAttribute( + NameOID.ORGANIZATION_NAME, + "Samsung Electronics", + ), + x509.NameAttribute( + NameOID.ORGANIZATIONAL_UNIT_NAME, + organizational_unit, + ), + x509.NameAttribute( + NameOID.COMMON_NAME, + f"OCF Device: Test ({identity})", + ), + ] + ) + leaf = _build_certificate( + subject=leaf_name, + issuer=intermediate_name, + public_key=leaf_key.public_key(), + issuer_key=intermediate_key, + serial_number=103, + is_ca=False, + ) + + root_pem = root.public_bytes(serialization.Encoding.PEM).decode() + intermediate_pem = intermediate.public_bytes(serialization.Encoding.PEM).decode() + leaf_pem = leaf.public_bytes(serialization.Encoding.PEM).decode() + key_pem = leaf_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + return SimpleNamespace( + root_pem=root_pem, + certificate_pem=leaf_pem + intermediate_pem, + private_key_pem=key_pem, + leaf=crypto.load_certificate(crypto.FILETYPE_PEM, leaf_pem), + intermediate=crypto.load_certificate( + crypto.FILETYPE_PEM, + intermediate_pem, + ), + ) + + +@pytest.fixture(scope="module") +def generated_chain(): + return _make_generated_chain(_IDENTITY) + + +def _configured_context(chain, profile=None) -> SSL.Context: + context = SSL.Context(SSL.DTLS_METHOD) + CertificateAuth.from_memory( + chain.certificate_pem, + chain.private_key_pem, + server_profile=profile, + ).configure_context(context) + return context + + +def _first_client_hello(context: SSL.Context) -> bytes: + connection = SSL.Connection(context, None) + connection.set_connect_state() + with pytest.raises(SSL.WantReadError): + connection.do_handshake() + chunks = [] + while True: + try: + chunks.append(connection.bio_read(65535)) + except SSL.WantReadError: + break + assert len(chunks) == 1 + return chunks[0] + + +def _parse_client_hello(datagram: bytes): + """Return cipher suites and extensions from one DTLS ClientHello.""" + assert datagram[0] == 22 + record_length = int.from_bytes(datagram[11:13], "big") + handshake = datagram[13 : 13 + record_length] + assert handshake[0] == 1 + body = handshake[12:] + + offset = 2 + 32 + session_id_length = body[offset] + offset += 1 + session_id_length + cookie_length = body[offset] + offset += 1 + cookie_length + + cipher_length = int.from_bytes(body[offset : offset + 2], "big") + offset += 2 + ciphers = [ + int.from_bytes(body[index : index + 2], "big") + for index in range(offset, offset + cipher_length, 2) + ] + offset += cipher_length + + compression_length = body[offset] + offset += 1 + compression_length + extensions_length = int.from_bytes(body[offset : offset + 2], "big") + offset += 2 + extensions_end = offset + extensions_length + extensions = {} + while offset < extensions_end: + extension_type = int.from_bytes(body[offset : offset + 2], "big") + extension_length = int.from_bytes( + body[offset + 2 : offset + 4], + "big", + ) + offset += 4 + assert extension_type not in extensions + extensions[extension_type] = body[offset : offset + extension_length] + offset += extension_length + assert offset == extensions_end == len(body) + return ciphers, extensions + + +def _vector_values(extension: bytes) -> list[int]: + vector_length = int.from_bytes(extension[:2], "big") + assert vector_length == len(extension) - 2 + return [ + int.from_bytes(extension[index : index + 2], "big") + for index in range(2, len(extension), 2) + ] + + +def _verify_chain(context: SSL.Context, chain) -> None: + crypto.X509StoreContext( + context.get_cert_store(), + chain.leaf, + [chain.intermediate], + ).verify_certificate() + + +def test_profile_accepts_only_canonical_nonzero_identity(): + assert repr(SamsungServerProfile.bound_device(_IDENTITY)) == ( + "SamsungServerProfile()" + ) + assert repr(SamsungServerProfile.bound_device(str(_IDENTITY))) == ( + "SamsungServerProfile()" + ) + + for invalid in ( + UUID(int=0), + str(UUID(int=0)), + str(_IDENTITY).upper(), + "{" + str(_IDENTITY) + "}", + "not-an-identity", + ): + with pytest.raises(ValueError, match="canonical non-zero UUID"): + SamsungServerProfile.bound_device(invalid) + for invalid in (True, 1, b"not-an-identity"): + with pytest.raises(TypeError, match="UUID or string"): + SamsungServerProfile.bound_device(invalid) + + +def test_profile_additional_ca_input_is_bounded_and_parsed(generated_chain): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + assert repr(profile) == "SamsungServerProfile()" + + with pytest.raises(TypeError, match="must be a string"): + SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem.encode(), + ) + for invalid in ( + "", + generated_chain.root_pem + "unexpected trailing material", + generated_chain.root_pem * 5, + generated_chain.root_pem * 2, + generated_chain.root_pem + (" " * (32 * 1024)), + generated_chain.certificate_pem, + "-----BEGIN CERTIFICATE-----\ninvalid\n-----END CERTIFICATE-----", + "non-ascii-\N{SNOWMAN}", + ): + with pytest.raises(ValueError): + SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=invalid, + ) + + +def test_profile_is_immutable_and_has_no_public_identity_or_ca_surface( + generated_chain, +): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + rendered = repr(profile) + assert str(_IDENTITY) not in rendered + assert "BEGIN CERTIFICATE" not in rendered + assert not hasattr(profile, "expected_device_identity") + assert not hasattr(profile, "additional_ca_pem") + with pytest.raises(TypeError): + vars(profile) + with pytest.raises(TypeError): + asdict(profile) + with pytest.raises(AttributeError, match="immutable"): + profile.expected_device_identity = _OTHER_IDENTITY + with pytest.raises(AttributeError, match="immutable"): + del profile._expected_device_identity + + +def test_certificate_auth_requires_the_exact_profile_type(generated_chain): + with pytest.raises(TypeError, match="SamsungServerProfile"): + CertificateAuth.from_memory( + generated_chain.certificate_pem, + generated_chain.private_key_pem, + server_profile=object(), + ) + + +def test_profile_emits_the_exact_client_hello_and_reuses_cold( + generated_chain, +): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + + first = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + second = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + assert first == second + + ciphers, extensions = first + # Older OpenSSL appends the non-negotiable renegotiation SCSV (0x00ff). + # No other negotiable cipher may enter the profile. + assert ciphers[0] == 0xC02B + assert set(ciphers) <= {0xC02B, 0x00FF} + assert _vector_values(extensions[10]) == [23] + assert _vector_values(extensions[13]) == [ + 0x0401, + 0x0403, + 0x0201, + 0x0203, + ] + assert 35 not in extensions + + +def test_python_floor_curve_fallback_has_the_same_wire_contract( + monkeypatch, + generated_chain, +): + signature_setter = auth_module._util.lib.SSL_CTX_set1_sigalgs_list + monkeypatch.setattr( + auth_module, + "_util", + SimpleNamespace( + lib=SimpleNamespace( + SSL_CTX_set1_sigalgs_list=signature_setter, + ) + ), + ) + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + + ciphers, extensions = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + + assert ciphers[0] == 0xC02B + assert set(ciphers) <= {0xC02B, 0x00FF} + assert _vector_values(extensions[10]) == [23] + assert _vector_values(extensions[13]) == [ + 0x0401, + 0x0403, + 0x0201, + 0x0203, + ] + assert 35 not in extensions + + +def test_profile_fails_closed_when_exact_openssl_support_is_unavailable( + monkeypatch, +): + class FallbackContext: + _context = object() + + def set_tmp_ecdh(self, _curve): + return None + + monkeypatch.setattr( + auth_module, + "_util", + SimpleNamespace(lib=SimpleNamespace()), + ) + profile = SamsungServerProfile.bound_device(_IDENTITY) + + with pytest.raises(RuntimeError, match="rejected"): + profile._configure_context(FallbackContext()) + + +def test_default_and_profile_verification_are_selected_per_provider( + monkeypatch, +): + valid_chain = _make_generated_chain(_IDENTITY) + mismatch_chain = _make_generated_chain(_OTHER_IDENTITY) + + class RecordingContext: + def __init__(self): + self._context = object() + 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",)) + + def set_options(self, options): + self.calls.append(("set_options", options)) + + profile_calls = [] + monkeypatch.setattr( + auth_module, + "_util", + SimpleNamespace( + lib=SimpleNamespace( + SSL_CTX_set1_curves_list=( + lambda handle, value: ( + profile_calls.append(("curves", handle, value)) or 1 + ) + ), + SSL_CTX_set1_sigalgs_list=( + lambda handle, value: ( + profile_calls.append(("signature_algorithms", handle, value)) + or 1 + ) + ), + ) + ), + ) + + default_context = RecordingContext() + CertificateAuth.from_files("/synthetic/cert", "/synthetic/key").configure_context( + default_context + ) + assert default_context.verify_callback(None, None, 0, 0, True) is True + assert default_context.verify_callback(None, None, 0, 0, False) is False + assert not profile_calls + assert all(call[0] != "set_options" for call in default_context.calls) + + profiled_context = RecordingContext() + profile = SamsungServerProfile.bound_device(_IDENTITY) + CertificateAuth.from_files( + "/synthetic/cert", + "/synthetic/key", + server_profile=profile, + ).configure_context(profiled_context) + + assert [call[0] for call in profile_calls] == [ + "curves", + "signature_algorithms", + ] + assert ("set_options", SSL.OP_NO_TICKET) in profiled_context.calls + callback = profiled_context.verify_callback + assert callback(None, valid_chain.leaf, 0, 0, True) is True + assert callback(None, mismatch_chain.leaf, 0, 0, True) is False + assert callback(None, valid_chain.leaf, 0, 0, False) is False + + +def test_profile_identity_verification_rejects_malformed_subjects(): + profile = SamsungServerProfile.bound_device(_IDENTITY) + wrong_role = _make_generated_chain( + _IDENTITY, + organizational_unit="Unexpected Device", + ) + + assert profile._verify_peer(None, wrong_role.leaf, 0, 0, True) is False + assert profile._verify_peer(None, None, 0, 0, True) is False + assert profile._verify_peer(None, wrong_role.leaf, 0, -1, True) is False + assert profile._verify_peer(None, object(), 0, 0, True) is False + assert profile._verify_peer(None, object(), 0, 1, True) is True + + duplicate_subject = x509.Name( + [ + x509.NameAttribute( + NameOID.COMMON_NAME, + f"OCF Device: First ({_IDENTITY})", + ), + x509.NameAttribute( + NameOID.COMMON_NAME, + f"OCF Device: Other ({_IDENTITY})", + ), + x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "OCF HA Device"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Samsung Electronics"), + x509.NameAttribute(NameOID.COUNTRY_NAME, "KR"), + ] + ) + duplicate_common_name = SimpleNamespace( + to_cryptography=lambda: SimpleNamespace(subject=duplicate_subject) + ) + assert ( + profile._verify_peer( + None, + duplicate_common_name, + 0, + 0, + True, + ) + is False + ) + + +def test_additional_ca_is_scoped_and_invalid_intermediate_is_rejected( + generated_chain, +): + default_context = _configured_context(generated_chain) + with pytest.raises(crypto.X509StoreContextError): + _verify_chain(default_context, generated_chain) + + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + profiled_context = _configured_context(generated_chain, profile) + _verify_chain(profiled_context, generated_chain) + + missing_constraints = _make_generated_chain( + _IDENTITY, + intermediate_has_constraints=False, + ) + missing_constraints_profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=missing_constraints.root_pem, + ) + missing_constraints_context = _configured_context( + missing_constraints, + missing_constraints_profile, + ) + with pytest.raises(crypto.X509StoreContextError): + _verify_chain(missing_constraints_context, missing_constraints) + + +def test_profile_errors_and_provider_repr_do_not_echo_inputs(generated_chain): + marker = "private" + "-profile-marker" + with pytest.raises(ValueError) as captured: + SamsungServerProfile.bound_device(marker) + assert marker not in str(captured.value) + + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + provider = CertificateAuth.from_memory( + generated_chain.certificate_pem, + generated_chain.private_key_pem, + server_profile=profile, + ) + rendered = repr(provider) + repr(profile) + assert str(_IDENTITY) not in rendered + assert generated_chain.root_pem not in rendered + assert generated_chain.private_key_pem not in rendered diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index f40e5b4..d4c33df 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -10,6 +10,7 @@ AuthenticationProvider, CertificateAuth, PskAuth, + SamsungServerProfile, ) from smartthings_local.protocol.dtls_session import DtlsCoapSession @@ -54,6 +55,19 @@ def test_certificate_auth_is_a_public_authentication_provider(): provider = CertificateAuth.from_files("/synthetic/cert.pem", "/synthetic/key") assert isinstance(provider, AuthenticationProvider) + for factory in (CertificateAuth.from_files, CertificateAuth.from_memory): + profile_parameter = inspect.signature(factory).parameters["server_profile"] + assert profile_parameter.kind is inspect.Parameter.KEYWORD_ONLY + assert profile_parameter.default is None + + +def test_samsung_server_profile_is_public_and_explicitly_bound(): + parameters = inspect.signature(SamsungServerProfile.bound_device).parameters + assert list(parameters) == ["expected_device_identity", "additional_ca_pem"] + assert parameters["expected_device_identity"].default is inspect.Parameter.empty + assert parameters["additional_ca_pem"].kind is inspect.Parameter.KEYWORD_ONLY + assert parameters["additional_ca_pem"].default is None + def test_psk_auth_is_a_public_authentication_provider(): provider = PskAuth(identity=b"i" * 16, key=b"k" * 16)