From 83a59734345726432ca2e1b78dbe3315b6b08944 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 | 70 ++ smartthings_local/protocol/auth.py | 343 +++++++++- tests/test_certificate_profiles.py | 1005 ++++++++++++++++++++++++++++ tests/test_public_api_contract.py | 44 ++ 4 files changed, 1453 insertions(+), 9 deletions(-) create mode 100644 tests/test_certificate_profiles.py diff --git a/README.md b/README.md index 52c2f41..2983272 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,76 @@ auth = CertificateAuth.from_memory(cert_pem, key_pem) sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth) ``` +Some newer OCF-PKI devices require an exact Samsung DTLS offer and present a +hardware certificate whose subject contains a certificate UUID. That UUID can +be distinct from the runtime OCF device UUID reported by `/oic/d`, so callers +must obtain and verify the certificate identity independently. When the caller +already has an authorized client certificate and a previously verified +hardware-certificate UUID, opt in to both requirements explicitly: + +```python +from smartthings_local.protocol.auth import ( + CertificateAuth, + SamsungServerProfile, +) + +server_profile = SamsungServerProfile.bound_device( + expected_certificate_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 default profile is restricted to Samsung home-appliance leaves with +`OU=OCF HA Device`. 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 certificate 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. + +Samsung VD-family devices can present the same wire profile with the distinct +`OU=OCF VD Device` role. Select that role explicitly; profiles never fall back +between device classes: + +```python +from smartthings_local.protocol.auth import ( + SamsungServerProfile, + SamsungServerRole, + ServerCertificateAuth, +) + +server_profile = SamsungServerProfile.bound_device( + expected_certificate_uuid, + role=SamsungServerRole.VD_DEVICE, +) +auth = ServerCertificateAuth(server_profile=server_profile) +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. + +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, the server-only provider can authenticate the +initial manufacturer-certificate channel, but it does not implement the OTM +that follows. The already-owned new-PKI case in +[issue #16](https://github.com/QuiteYellow/SmartThings-Local/issues/16) still +requires an authorized client identity before ordinary protected resources +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..14e87a1 100644 --- a/smartthings_local/protocol/auth.py +++ b/smartthings_local/protocol/auth.py @@ -2,16 +2,33 @@ from __future__ import annotations +import logging import re +import warnings +from enum import Enum from os import PathLike from pathlib import Path from typing import Protocol, runtime_checkable +from uuid import UUID +from cryptography.x509.oid import ExtensionOID from OpenSSL import SSL, _util, crypto +logger = logging.getLogger(__name__) + _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 +70,299 @@ def configure_context(self, context: SSL.Context) -> None: """Configure a context while this provider remains session-owned.""" +class SamsungServerRole(Enum): + """Known Samsung OCF hardware-certificate subject roles.""" + + HOME_APPLIANCE = "OCF HA Device" + VD_DEVICE = "OCF VD Device" + + +class SamsungServerProfile: + """Opt-in Samsung hardware-certificate verification profile.""" + + __slots__ = ( + "_additional_ca_certificates", + "_expected_certificate_identity", + "_role", + ) + + def __init__( + self, + *, + expected_certificate_identity: UUID | str, + role: SamsungServerRole = SamsungServerRole.HOME_APPLIANCE, + additional_ca_pem: str | None = None, + ) -> None: + if type(expected_certificate_identity) is UUID: + parsed_identity = expected_certificate_identity + elif type(expected_certificate_identity) is str: + try: + parsed_identity = UUID(expected_certificate_identity) + except ValueError: + raise ValueError( + "expected_certificate_identity must be a canonical " + "non-zero UUID" + ) from None + if expected_certificate_identity != str(parsed_identity): + raise ValueError( + "expected_certificate_identity must be a canonical " + "non-zero UUID" + ) + else: + raise TypeError( + "expected_certificate_identity must be a UUID or string" + ) + if parsed_identity.int == 0: + raise ValueError( + "expected_certificate_identity must be a canonical " + "non-zero UUID" + ) + if type(role) is not SamsungServerRole: + raise TypeError("role must be a SamsungServerRole") + + 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_certificate_identity", + parsed_identity, + ) + object.__setattr__(self, "_role", role) + 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_certificate_identity: UUID | str, + *, + role: SamsungServerRole = SamsungServerRole.HOME_APPLIANCE, + additional_ca_pem: str | None = None, + ) -> SamsungServerProfile: + """Bind a verified Samsung hardware leaf to its certificate UUID.""" + return cls( + expected_certificate_identity=expected_certificate_identity, + role=role, + 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: + with warnings.catch_warnings(): + # pyOpenSSL deprecates this API in favor of cryptography, but + # reparsing Samsung's non-DER factory leaves with cryptography + # rejects certificates that OpenSSL has already verified. + warnings.simplefilter("ignore", DeprecationWarning) + components = [ + (name.decode("ascii"), value.decode("ascii")) + for name, value in certificate.get_subject().get_components() + ] + except ( + AttributeError, + TypeError, + UnicodeDecodeError, + ValueError, + crypto.Error, + ): + logger.warning("Unable to parse Samsung server certificate subject") + return False + + common_names = [value for name, value in components if name == "CN"] + organizational_units = [ + value for name, value in components if name == "OU" + ] + organizations = [value for name, value in components if name == "O"] + countries = [value for name, value in components if name == "C"] + # This deliberately pins the complete Samsung subject role. The OCF + # reference implementation reads only the UUID-bearing CN, but + # relaxing C/O/OU here could accept a different certificate cohort. + if ( + len(common_names) != 1 + or organizational_units != [self._role.value] + 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_certificate_identity + ) + + +def _configure_certificate_server( + context: SSL.Context, + server_profile: SamsungServerProfile | None, +) -> None: + """Configure certificate-server verification for one DTLS context.""" + context.load_verify_locations(_OCF_ROOT_CA) + if server_profile is None: + context.set_verify(SSL.VERIFY_PEER, _verify_peer) + else: + server_profile._configure_context(context) + context.set_verify( + SSL.VERIFY_PEER, + 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 + # do not expose SSL_CTX_set_security_level on this build. + context.set_cipher_list(_DTLS_CIPHERS) + + +class ServerCertificateAuth: + """Verify a pinned Samsung server without a client certificate.""" + + __slots__ = ("_server_profile",) + + def __init__(self, *, server_profile: SamsungServerProfile) -> None: + if type(server_profile) is not SamsungServerProfile: + raise TypeError("server_profile must be a SamsungServerProfile") + object.__setattr__(self, "_server_profile", server_profile) + + def __setattr__(self, _name: str, _value: object) -> None: + raise AttributeError("ServerCertificateAuth is immutable") + + def __delattr__(self, _name: str) -> None: + raise AttributeError("ServerCertificateAuth is immutable") + + def __repr__(self) -> str: + """Return a representation without server identity or trust details.""" + return "ServerCertificateAuth()" + + def configure_context(self, context: SSL.Context) -> None: + """Verify the selected server profile without loading client material.""" + _configure_certificate_server(context, self._server_profile) + + class CertificateAuth: """Certificate authentication loaded from files or in-memory PEM data. @@ -65,6 +375,7 @@ class CertificateAuth: "_certificate_pem", "_private_key_path", "_private_key_pem", + "_server_profile", ) def __init__( @@ -74,6 +385,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 +413,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 +430,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 +443,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 +458,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: @@ -150,13 +474,7 @@ 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) - # @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) + _configure_certificate_server(context, self._server_profile) if self._certificate_pem is not None: _load_pem_chain( context, @@ -240,7 +558,14 @@ def configure_context(self, context: SSL.Context) -> None: "the installed OpenSSL binding does not support DTLS PSK" ) context.set_cipher_list(_DTLS_PSK_CIPHERS) - setter(context._context, self._callback) # noqa: SLF001 + setter(context._context, self._callback) -__all__ = ["AuthenticationProvider", "CertificateAuth", "PskAuth"] +__all__ = [ + "AuthenticationProvider", + "CertificateAuth", + "PskAuth", + "SamsungServerProfile", + "SamsungServerRole", + "ServerCertificateAuth", +] diff --git a/tests/test_certificate_profiles.py b/tests/test_certificate_profiles.py new file mode 100644 index 0000000..91ba1fd --- /dev/null +++ b/tests/test_certificate_profiles.py @@ -0,0 +1,1005 @@ +"""Synthetic security and wire-contract tests for certificate profiles.""" + +from __future__ import annotations + +import logging +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, + SamsungServerRole, + ServerCertificateAuth, +) + +_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 _encode_der_length(length: int) -> bytes: + if length < 0x80: + return bytes([length]) + encoded = length.to_bytes((length.bit_length() + 7) // 8, "big") + return bytes([0x80 | len(encoded)]) + encoded + + +def _encode_der_element(tag: int, contents: bytes) -> bytes: + return bytes([tag]) + _encode_der_length(len(contents)) + contents + + +def _read_der_element( + encoded: bytes, + offset: int, +) -> tuple[int, int, int, int]: + tag = encoded[offset] + first_length_octet = encoded[offset + 1] + if first_length_octet < 0x80: + length_octets = 0 + length = first_length_octet + else: + length_octets = first_length_octet & 0x7F + assert 0 < length_octets <= 4 + length = int.from_bytes( + encoded[offset + 2 : offset + 2 + length_octets], + "big", + ) + contents_start = offset + 2 + length_octets + contents_end = contents_start + length + assert contents_end <= len(encoded) + return tag, contents_start, contents_end, contents_end + + +def _add_tbs_signature_algorithm_trailing_null( + certificate: x509.Certificate, + issuer_key, +) -> bytes: + """Make a signed synthetic leaf with Samsung-style algorithm extra data.""" + encoded = certificate.public_bytes(serialization.Encoding.DER) + outer_tag, outer_start, _outer_end, certificate_end = _read_der_element( + encoded, + 0, + ) + assert outer_tag == 0x30 + assert certificate_end == len(encoded) + + tbs_tag, tbs_start, tbs_end, tbs_next = _read_der_element( + encoded, + outer_start, + ) + assert tbs_tag == 0x30 + signature_algorithm_offset = tbs_start + for _ in range(2): + _, _, _, signature_algorithm_offset = _read_der_element( + encoded, + signature_algorithm_offset, + ) + ( + signature_algorithm_tag, + signature_algorithm_start, + signature_algorithm_end, + _, + ) = _read_der_element(encoded, signature_algorithm_offset) + assert signature_algorithm_tag == 0x30 + + malformed_signature_algorithm = _encode_der_element( + 0x30, + encoded[signature_algorithm_start:signature_algorithm_end] + + b"\x05\x00", + ) + malformed_tbs = _encode_der_element( + 0x30, + encoded[tbs_start:signature_algorithm_offset] + + malformed_signature_algorithm + + encoded[signature_algorithm_end:tbs_end], + ) + + ( + outer_signature_tag, + outer_signature_start, + outer_signature_end, + _, + ) = _read_der_element( + encoded, + tbs_next, + ) + assert outer_signature_tag == 0x30 + malformed_outer_signature_algorithm = _encode_der_element( + 0x30, + encoded[outer_signature_start:outer_signature_end] + b"\x05\x00", + ) + signature = issuer_key.sign(malformed_tbs, ec.ECDSA(hashes.SHA256())) + return _encode_der_element( + 0x30, + malformed_tbs + + malformed_outer_signature_algorithm + + _encode_der_element(0x03, b"\x00" + signature), + ) + + +def _make_generated_chain( + identity: UUID, + *, + organizational_unit: str = "OCF HA Device", + intermediate_has_constraints: bool = True, + leaf_signature_algorithm_trailing_null: bool = False, +): + """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() + if leaf_signature_algorithm_trailing_null: + leaf_der = _add_tbs_signature_algorithm_trailing_null( + leaf, + intermediate_key, + ) + openssl_leaf = crypto.load_certificate(crypto.FILETYPE_ASN1, leaf_der) + leaf_pem = crypto.dump_certificate( + crypto.FILETYPE_PEM, + openssl_leaf, + ).decode() + else: + leaf_pem = leaf.public_bytes(serialization.Encoding.PEM).decode() + openssl_leaf = crypto.load_certificate(crypto.FILETYPE_PEM, leaf_pem) + 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=openssl_leaf, + 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, *, server_only=False) -> SSL.Context: + context = SSL.Context(SSL.DTLS_METHOD) + if server_only: + ServerCertificateAuth(server_profile=profile).configure_context(context) + else: + CertificateAuth.from_memory( + chain.certificate_pem, + chain.private_key_pem, + server_profile=profile, + ).configure_context(context) + return context + + +def _configured_certificate_server_context(chain) -> SSL.Context: + context = SSL.Context(SSL.DTLS_METHOD) + certificates = auth_module._PEM_CERT_RE.findall( + chain.certificate_pem.encode() + ) + context.use_certificate( + crypto.load_certificate(crypto.FILETYPE_PEM, certificates[0]) + ) + for certificate in certificates[1:]: + context.add_extra_chain_cert( + crypto.load_certificate(crypto.FILETYPE_PEM, certificate) + ) + context.use_privatekey( + crypto.load_privatekey( + crypto.FILETYPE_PEM, + chain.private_key_pem.encode(), + ) + ) + context.check_privatekey() + # Request a client certificate but permit an empty certificate message, + # matching a server-authenticated manufacturer-certificate OTM carrier. + context.set_verify(SSL.VERIFY_PEER, lambda *_args: True) + 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 _drive_memory_bio_handshake( + client_context: SSL.Context, + server_context: SSL.Context, +) -> tuple[SSL.Connection, SSL.Connection]: + client = SSL.Connection(client_context, None) + client.set_connect_state() + client.set_ciphertext_mtu(1200) + server = SSL.Connection(server_context, None) + server.set_accept_state() + server.set_ciphertext_mtu(1200) + client_done = False + server_done = False + + for _ in range(20): + if not client_done: + try: + client.do_handshake() + client_done = True + except (SSL.WantReadError, SSL.WantWriteError): + pass + while True: + try: + server.bio_write(client.bio_read(65535)) + except SSL.WantReadError: + break + + if not server_done: + try: + server.do_handshake() + server_done = True + except (SSL.WantReadError, SSL.WantWriteError): + pass + while True: + try: + client.bio_write(server.bio_read(65535)) + except SSL.WantReadError: + break + + if client_done and server_done: + return client, server + + raise AssertionError("synthetic DTLS handshake did not complete") + + +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_roles_are_explicit_and_fail_closed(): + home_chain = _make_generated_chain(_IDENTITY) + video_chain = _make_generated_chain( + _IDENTITY, + organizational_unit="OCF VD Device", + ) + mismatched_video_chain = _make_generated_chain( + _OTHER_IDENTITY, + organizational_unit="OCF VD Device", + ) + home_profile = SamsungServerProfile.bound_device(_IDENTITY) + video_profile = SamsungServerProfile.bound_device( + _IDENTITY, + role=SamsungServerRole.VD_DEVICE, + ) + + assert home_profile._verify_peer(None, home_chain.leaf, 0, 0, True) is True + assert home_profile._verify_peer(None, video_chain.leaf, 0, 0, True) is False + assert video_profile._verify_peer(None, video_chain.leaf, 0, 0, True) is True + assert video_profile._verify_peer(None, home_chain.leaf, 0, 0, True) is False + assert ( + video_profile._verify_peer( + None, + mismatched_video_chain.leaf, + 0, + 0, + True, + ) + is False + ) + + for invalid in ("OCF VD Device", "video_device", None, object()): + with pytest.raises(TypeError, match="SamsungServerRole"): + SamsungServerProfile.bound_device(_IDENTITY, role=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_certificate_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_certificate_identity = _OTHER_IDENTITY + with pytest.raises(AttributeError, match="immutable"): + del profile._expected_certificate_identity + + +def test_server_certificate_auth_is_explicit_immutable_and_redacted( + generated_chain, +): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + role=SamsungServerRole.VD_DEVICE, + additional_ca_pem=generated_chain.root_pem, + ) + provider = ServerCertificateAuth(server_profile=profile) + + assert repr(provider) == "ServerCertificateAuth()" + assert str(_IDENTITY) not in repr(provider) + assert generated_chain.root_pem not in repr(provider) + with pytest.raises(TypeError): + vars(provider) + with pytest.raises(TypeError): + asdict(provider) + with pytest.raises(AttributeError, match="immutable"): + provider.server_profile = profile + with pytest.raises(AttributeError, match="immutable"): + del provider._server_profile + for invalid in (None, object()): + with pytest.raises(TypeError, match="SamsungServerProfile"): + ServerCertificateAuth(server_profile=invalid) + + +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_server_only_profile_emits_the_same_client_hello_without_credentials( + generated_chain, +): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + role=SamsungServerRole.VD_DEVICE, + additional_ca_pem=generated_chain.root_pem, + ) + + certificate_hello = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + server_only_hello = _parse_client_hello( + _first_client_hello( + _configured_context( + generated_chain, + profile, + server_only=True, + ) + ) + ) + + assert server_only_hello == certificate_hello + + +def test_server_only_profile_completes_without_a_client_certificate(): + video_chain = _make_generated_chain( + _IDENTITY, + organizational_unit="OCF VD Device", + ) + profile = SamsungServerProfile.bound_device( + _IDENTITY, + role=SamsungServerRole.VD_DEVICE, + additional_ca_pem=video_chain.root_pem, + ) + client_context = _configured_context( + video_chain, + profile, + server_only=True, + ) + + client, server = _drive_memory_bio_handshake( + client_context, + _configured_certificate_server_context(video_chain), + ) + + assert client.get_peer_certificate() is not None + assert server.get_peer_certificate() is None + + +def test_profile_verifies_openssl_accepted_non_der_vd_leaf(): + video_chain = _make_generated_chain( + _IDENTITY, + organizational_unit="OCF VD Device", + leaf_signature_algorithm_trailing_null=True, + ) + try: + video_chain.leaf.to_cryptography() + except ValueError as error: + # cryptography 50 rejects this shape; the dependency floor accepts it. + assert "TbsCertificate" in str(error) + assert "signature_alg" in str(error) + + video_profile = SamsungServerProfile.bound_device( + _IDENTITY, + role=SamsungServerRole.VD_DEVICE, + additional_ca_pem=video_chain.root_pem, + ) + client_context = _configured_context( + video_chain, + video_profile, + server_only=True, + ) + _verify_chain(client_context, video_chain) + + client, server = _drive_memory_bio_handshake( + client_context, + _configured_certificate_server_context(video_chain), + ) + + assert client.get_peer_certificate() is not None + assert server.get_peer_certificate() is None + assert ( + SamsungServerProfile.bound_device( + _OTHER_IDENTITY, + role=SamsungServerRole.VD_DEVICE, + )._verify_peer(None, video_chain.leaf, 0, 0, True) + is False + ) + assert ( + SamsungServerProfile.bound_device(_IDENTITY)._verify_peer( + None, + video_chain.leaf, + 0, + 0, + True, + ) + is False + ) + + +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 + + profile_calls.clear() + video_chain = _make_generated_chain( + _IDENTITY, + organizational_unit="OCF VD Device", + ) + server_only_context = RecordingContext() + video_profile = SamsungServerProfile.bound_device( + _IDENTITY, + role=SamsungServerRole.VD_DEVICE, + ) + ServerCertificateAuth(server_profile=video_profile).configure_context( + server_only_context + ) + assert [call[0] for call in profile_calls] == [ + "curves", + "signature_algorithms", + ] + assert all( + call[0] + not in { + "use_certificate_chain_file", + "use_privatekey_file", + "check_privatekey", + } + for call in server_only_context.calls + ) + assert ( + server_only_context.verify_callback( + None, + video_chain.leaf, + 0, + 0, + True, + ) + is True + ) + assert ( + server_only_context.verify_callback( + None, + valid_chain.leaf, + 0, + 0, + True, + ) + is False + ) + + +def test_profile_identity_verification_rejects_malformed_subjects(caplog): + 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, 1, True) is True + + duplicate_common_name = SimpleNamespace( + get_subject=lambda: SimpleNamespace( + get_components=lambda: [ + (b"CN", f"OCF Device: First ({_IDENTITY})".encode()), + (b"CN", f"OCF Device: Other ({_IDENTITY})".encode()), + (b"OU", b"OCF HA Device"), + (b"O", b"Samsung Electronics"), + (b"C", b"KR"), + ] + ) + ) + assert ( + profile._verify_peer( + None, + duplicate_common_name, + 0, + 0, + True, + ) + is False + ) + + invalid_encoding = SimpleNamespace( + get_subject=lambda: SimpleNamespace( + get_components=lambda: [(b"CN", b"\xff")] + ) + ) + caplog.clear() + with caplog.at_level(logging.WARNING, logger=auth_module.__name__): + assert profile._verify_peer(None, object(), 0, 0, True) is False + assert ( + profile._verify_peer( + None, + invalid_encoding, + 0, + 0, + True, + ) + is False + ) + assert caplog.messages == [ + "Unable to parse Samsung server certificate subject", + "Unable to parse Samsung server certificate subject", + ] + + caplog.clear() + assert profile._verify_peer(None, wrong_role.leaf, 0, 0, True) is False + assert not caplog.records + + +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..f3f97c5 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -10,6 +10,9 @@ AuthenticationProvider, CertificateAuth, PskAuth, + SamsungServerProfile, + SamsungServerRole, + ServerCertificateAuth, ) from smartthings_local.protocol.dtls_session import DtlsCoapSession @@ -54,6 +57,47 @@ 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_certificate_identity", + "role", + "additional_ca_pem", + ] + assert ( + parameters["expected_certificate_identity"].default + is inspect.Parameter.empty + ) + assert parameters["role"].kind is inspect.Parameter.KEYWORD_ONLY + assert parameters["role"].default is SamsungServerRole.HOME_APPLIANCE + assert parameters["additional_ca_pem"].kind is inspect.Parameter.KEYWORD_ONLY + assert parameters["additional_ca_pem"].default is None + + +def test_server_certificate_auth_is_a_public_authentication_provider(): + profile = SamsungServerProfile.bound_device( + "abababab-abab-abab-abab-abababababab", + role=SamsungServerRole.VD_DEVICE, + ) + provider = ServerCertificateAuth(server_profile=profile) + assert isinstance(provider, AuthenticationProvider) + session = DtlsCoapSession("device.example", 5684, auth=provider) + assert session.auth is provider + assert session.cert_path is None + assert session.key_path is None + assert session.cert_pem is None + assert session.key_pem is None + parameters = inspect.signature(ServerCertificateAuth).parameters + assert list(parameters) == ["server_profile"] + assert parameters["server_profile"].kind is inspect.Parameter.KEYWORD_ONLY + assert parameters["server_profile"].default is inspect.Parameter.empty + def test_psk_auth_is_a_public_authentication_provider(): provider = PskAuth(identity=b"i" * 16, key=b"k" * 16)