From 8fb37ca2ed77808d5833ce1ef54ee16cef7858b1 Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Sat, 8 Aug 2026 11:23:15 -0700 Subject: [PATCH] refactor(protocol): add certificate authentication provider --- README.md | 20 +- smartthings_local/protocol/auth.py | 167 ++++++++++++ smartthings_local/protocol/dtls_session.py | 84 +++--- tests/test_dtls_session_cert_loading.py | 290 +++++++++++++++++++-- tests/test_public_api_contract.py | 9 + 5 files changed, 493 insertions(+), 77 deletions(-) create mode 100644 smartthings_local/protocol/auth.py diff --git a/README.md b/README.md index 54f69f3..33c9001 100644 --- a/README.md +++ b/README.md @@ -27,12 +27,16 @@ session directly: ```python import cbor2 +from smartthings_local.protocol.auth import CertificateAuth from smartthings_local.protocol.dtls_session import DtlsCoapSession +auth = CertificateAuth.from_files( + "certs/client_fullchain.pem", + "certs/client.key", +) sess = DtlsCoapSession( "192.0.2.100", 49154, - cert_path="certs/client_fullchain.pem", - key_path="certs/client.key", + auth=auth, ) sess.connect() sess.start_reader() @@ -44,12 +48,19 @@ sess.subscribe(["operational", "state", "vs", "0"], # OBSERVE sess.close() ``` -If the cert/key are minted at runtime and never written to disk (e.g. inside an HA config flow), pass them in memory instead of by path: +If the cert/key are minted at runtime and never written to disk (e.g. inside +an HA config flow), create the provider from memory instead: ```python -sess = DtlsCoapSession("192.0.2.100", 49154, cert_pem=cert_pem, key_pem=key_pem) +auth = CertificateAuth.from_memory(cert_pem, key_pem) +sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth) ``` +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` +with those legacy arguments. + ### Classified errors Runtime transport failures use the public types in @@ -479,6 +490,7 @@ smartthings_local/ The installable library — `pip install sm __init__.py protocol/ DTLS-CoAP transport (reusable by any consumer, not just MQTT) __init__.py + auth.py Immutable DTLS authentication providers coap.py CoAP wire protocol: message encode/decode, token handling dtls_session.py DTLS session: handshake, client-cert auth (file or in-memory PEM), Block2, liveness dtls_probe.py Stateless DTLS liveness + opt-in stateful diagnostic diff --git a/smartthings_local/protocol/auth.py b/smartthings_local/protocol/auth.py new file mode 100644 index 0000000..8291e09 --- /dev/null +++ b/smartthings_local/protocol/auth.py @@ -0,0 +1,167 @@ +"""Immutable authentication providers for DTLS sessions.""" + +from __future__ import annotations + +import re +from os import PathLike +from pathlib import Path +from typing import Protocol, runtime_checkable + +from OpenSSL import SSL, crypto + +_OCF_ROOT_CA = str(Path(__file__).with_name("ocf_root_ca.pem")) +_DTLS_CIPHERS = b"ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0" +_PEM_CERT_RE = re.compile( + rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", + re.DOTALL, +) + + +def _verify_peer(_connection, _certificate, _error, _depth, ok): + """Keep pyOpenSSL's existing verification result unchanged.""" + return ok + + +def _load_pem_chain(ctx: SSL.Context, cert_pem: str, key_pem: str) -> None: + """Load a PEM certificate chain and private key into a context in memory.""" + certificates = _PEM_CERT_RE.findall(cert_pem.encode()) + if not certificates: + raise ValueError("No certificates found in cert_pem") + ctx.use_certificate( + crypto.load_certificate(crypto.FILETYPE_PEM, certificates[0]) + ) + for extra in certificates[1:]: + ctx.add_extra_chain_cert( + crypto.load_certificate(crypto.FILETYPE_PEM, extra) + ) + ctx.use_privatekey( + crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem.encode()) + ) + ctx.check_privatekey() + + +@runtime_checkable +class AuthenticationProvider(Protocol): + """Configure authentication for a newly created DTLS context.""" + + def configure_context(self, context: SSL.Context) -> None: + """Configure trust, verification, ciphers, and client credentials.""" + + +class CertificateAuth: + """Certificate authentication loaded from files or in-memory PEM data. + + Use :meth:`from_files` or :meth:`from_memory` to create an instance. + Credential sources are intentionally not exposed as public attributes. + """ + + __slots__ = ( + "_certificate_path", + "_certificate_pem", + "_private_key_path", + "_private_key_pem", + ) + + def __init__( + self, + *, + certificate_path: str | PathLike[str] | None = None, + private_key_path: str | PathLike[str] | None = None, + certificate_pem: str | None = None, + private_key_pem: str | None = None, + ) -> None: + file_supplied = ( + certificate_path is not None or private_key_path is not None + ) + memory_supplied = ( + certificate_pem is not None or private_key_pem is not None + ) + if file_supplied and memory_supplied: + raise ValueError( + "pass either certificate_path/private_key_path or " + "certificate_pem/private_key_pem, not both" + ) + if file_supplied: + if certificate_path is None or private_key_path is None: + raise ValueError( + "certificate_path and private_key_path must be passed together" + ) + elif memory_supplied: + if certificate_pem is None or private_key_pem is None: + raise ValueError( + "certificate_pem and private_key_pem must be passed together" + ) + else: + raise ValueError( + "must pass either certificate_path/private_key_path or " + "certificate_pem/private_key_pem" + ) + object.__setattr__( + self, + "_certificate_path", + str(certificate_path) if certificate_path is not None else None, + ) + object.__setattr__( + self, + "_private_key_path", + str(private_key_path) if private_key_path is not None else None, + ) + object.__setattr__(self, "_certificate_pem", certificate_pem) + object.__setattr__(self, "_private_key_pem", private_key_pem) + + def __setattr__(self, _name: str, _value: object) -> None: + raise AttributeError("CertificateAuth is immutable") + + def __delattr__(self, _name: str) -> None: + raise AttributeError("CertificateAuth is immutable") + + @classmethod + def from_files( + cls, + certificate_path: str | PathLike[str], + private_key_path: str | PathLike[str], + ) -> CertificateAuth: + """Create a provider backed by certificate-chain and key files.""" + return cls( + certificate_path=certificate_path, + private_key_path=private_key_path, + ) + + @classmethod + def from_memory( + cls, + certificate_pem: str, + private_key_pem: str, + ) -> 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, + ) + + def __repr__(self) -> str: + """Return a representation that never includes credential material.""" + return "CertificateAuth()" + + 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) + if self._certificate_pem is not None: + _load_pem_chain( + context, + self._certificate_pem, + self._private_key_pem, + ) + else: + context.use_certificate_chain_file(self._certificate_path) + context.use_privatekey_file(self._private_key_path) + context.check_privatekey() + + +__all__ = ["AuthenticationProvider", "CertificateAuth"] diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index cfee7a2..aeeb305 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -19,11 +19,9 @@ delivered via the on_notification callback. """ import os -import re as _re import socket import threading import time -from pathlib import Path from OpenSSL import SSL @@ -42,15 +40,18 @@ encode_options, parse_coap, build_coap, block_value, fmt_code, split_dtls as _split_dtls, ) +from .auth import ( + AuthenticationProvider, + CertificateAuth, + _DTLS_CIPHERS, + _OCF_ROOT_CA, + _load_pem_chain, +) from .endpoint import open_connected_udp_socket import logging logger = logging.getLogger(__name__) -_OCF_ROOT_CA = str(Path(__file__).parent / 'ocf_root_ca.pem') -_DTLS_CIPHERS = b'ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0' - - # Diagnostic logging — when DEBUG_BRIDGE=1 in env, the bridge dumps # every received CoAP frame, every /operational/state/vs/0 + /oven/vs/0 # + /power/vs/0 + /mode/vs/0-options rep change, the full link tree at @@ -71,32 +72,6 @@ # once the ceiling is measured empirically. _DEFAULT_RATE_LIMIT_RPS = 5.0 -_PEM_CERT_RE = _re.compile( - rb'-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----', - _re.DOTALL, -) - - -def _load_pem_chain(ctx: SSL.Context, cert_pem: str, key_pem: str) -> None: - """Load a PEM cert chain and private key into an SSL context in memory. - - Parses all certificate blocks from cert_pem: the first is the leaf - (use_certificate), the rest are intermediates (add_extra_chain_cert). - No temp files are written. - """ - from OpenSSL import crypto - certs = _PEM_CERT_RE.findall(cert_pem.encode()) - if not certs: - raise ValueError("No certificates found in cert_pem") - ctx.use_certificate(crypto.load_certificate(crypto.FILETYPE_PEM, certs[0])) - for extra in certs[1:]: - ctx.add_extra_chain_cert( - crypto.load_certificate(crypto.FILETYPE_PEM, extra) - ) - ctx.use_privatekey(crypto.load_privatekey(crypto.FILETYPE_PEM, key_pem.encode())) - ctx.check_privatekey() - - class DtlsCoapSession: """Single sustained DTLS-CoAP session. @@ -109,10 +84,9 @@ class DtlsCoapSession: code, _ = sess.post(['mode','vs','0'], cbor) sess.close() - Cert material comes from either a file pair (cert_path, key_path) or - an in-memory PEM pair (cert_pem, key_pem) — exactly one pair required. - The in-memory path exists for callers (e.g. an HA config flow) that - mint a client cert at runtime and never write it to disk. + Authentication comes from an immutable provider. For compatibility, + cert_path/key_path and cert_pem/key_pem create a CertificateAuth provider + internally — exactly one legacy pair is required when auth is omitted. """ HANDSHAKE_TIMEOUT_S = 12.0 @@ -123,17 +97,24 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, cert_pem=None, key_pem=None, on_notification=None, mtu=1200, rate_limit_rps: float = _DEFAULT_RATE_LIMIT_RPS, - local_port=None, family=socket.AF_UNSPEC): - if (cert_path is not None or key_path is not None) and \ - (cert_pem is not None or key_pem is not None): + local_port=None, family=socket.AF_UNSPEC, + auth: AuthenticationProvider | None = None): + file_supplied = cert_path is not None or key_path is not None + memory_supplied = cert_pem is not None or key_pem is not None + if auth is not None and (file_supplied or memory_supplied): + raise ValueError( + "pass auth or legacy certificate arguments, not both") + if auth is None and file_supplied and memory_supplied: raise ValueError( "pass either cert_path/key_path or cert_pem/key_pem, not both") - if cert_pem is not None or key_pem is not None: + if auth is None and memory_supplied: if cert_pem is None or key_pem is None: raise ValueError("cert_pem and key_pem must be passed together") - elif cert_path is None or key_path is None: + elif auth is None and (cert_path is None or key_path is None): raise ValueError( "must pass either cert_path/key_path or cert_pem/key_pem") + if auth is not None and not isinstance(auth, AuthenticationProvider): + raise TypeError("auth must implement AuthenticationProvider") self.host = host self.port = port @@ -141,6 +122,12 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, self.key_path = str(key_path) if key_path is not None else None self.cert_pem = cert_pem self.key_pem = key_pem + if auth is None: + if cert_pem is not None: + auth = CertificateAuth.from_memory(cert_pem, key_pem) + else: + auth = CertificateAuth.from_files(self.cert_path, self.key_path) + self.auth = auth self.on_notification = on_notification # fn(href, payload_bytes) self.mtu = mtu self._min_req_interval = 1.0 / rate_limit_rps @@ -196,20 +183,7 @@ def connect(self): """DTLS handshake. Blocks up to HANDSHAKE_TIMEOUT_S. Raises ConnectionError / TimeoutError on failure.""" ctx = SSL.Context(SSL.DTLS_METHOD) - - ctx.load_verify_locations(_OCF_ROOT_CA) - ctx.set_verify(SSL.VERIFY_PEER, lambda conn, cert, err, depth, ok: ok) - # @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. - ctx.set_cipher_list(_DTLS_CIPHERS) - if self.cert_pem is not None: - _load_pem_chain(ctx, self.cert_pem, self.key_pem) - else: - ctx.use_certificate_chain_file(self.cert_path) - ctx.use_privatekey_file(self.key_path) - ctx.check_privatekey() + self.auth.configure_context(ctx) conn = SSL.Connection(ctx, None) conn.set_connect_state() diff --git a/tests/test_dtls_session_cert_loading.py b/tests/test_dtls_session_cert_loading.py index 2891e1b..b29102d 100644 --- a/tests/test_dtls_session_cert_loading.py +++ b/tests/test_dtls_session_cert_loading.py @@ -1,31 +1,69 @@ +import gc +import traceback +import weakref +from dataclasses import asdict +from datetime import datetime, timedelta, timezone + import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID from OpenSSL import SSL, crypto +from smartthings_local.protocol.auth import CertificateAuth from smartthings_local.protocol.dtls_session import DtlsCoapSession, _load_pem_chain -def _make_self_signed_pem_pair(): - """A throwaway self-signed cert + key, just to exercise PEM loading — - not meant to resemble a real Samsung client cert.""" - key = crypto.PKey() - key.generate_key(crypto.TYPE_RSA, 2048) +def _make_generated_pem_chain(): + """Create a throwaway leaf + root chain unrelated to Samsung devices.""" + now = datetime.now(timezone.utc) + root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + root_name = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, "Synthetic test root")] + ) + root_cert = ( + x509.CertificateBuilder() + .subject_name(root_name) + .issuer_name(root_name) + .public_key(root_key.public_key()) + .serial_number(1) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(hours=1)) + .add_extension(x509.BasicConstraints(ca=True, path_length=None), True) + .sign(root_key, hashes.SHA256()) + ) - cert = crypto.X509() - cert.get_subject().CN = "test" - cert.set_serial_number(1) - cert.gmtime_adj_notBefore(0) - cert.gmtime_adj_notAfter(3600) - cert.set_issuer(cert.get_subject()) - cert.set_pubkey(key) - cert.sign(key, "sha256") + leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + leaf_name = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, "Synthetic test client")] + ) + leaf_cert = ( + x509.CertificateBuilder() + .subject_name(leaf_name) + .issuer_name(root_name) + .public_key(leaf_key.public_key()) + .serial_number(2) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(hours=1)) + .add_extension(x509.BasicConstraints(ca=False, path_length=None), True) + .sign(root_key, hashes.SHA256()) + ) - cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode() - key_pem = crypto.dump_privatekey(crypto.FILETYPE_PEM, key).decode() + cert_pem = ( + leaf_cert.public_bytes(serialization.Encoding.PEM) + + root_cert.public_bytes(serialization.Encoding.PEM) + ).decode() + key_pem = leaf_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() return cert_pem, key_pem def test_load_pem_chain_loads_cert_and_key_in_memory(): - cert_pem, key_pem = _make_self_signed_pem_pair() + cert_pem, key_pem = _make_generated_pem_chain() ctx = SSL.Context(SSL.DTLS_METHOD) _load_pem_chain(ctx, cert_pem, key_pem) ctx.check_privatekey() # raises if cert/key don't match @@ -37,7 +75,7 @@ def test_load_pem_chain_rejects_cert_pem_with_no_certificates(): def test_session_requires_exactly_one_cert_source(): - cert_pem, key_pem = _make_self_signed_pem_pair() + cert_pem, key_pem = _make_generated_pem_chain() with pytest.raises(ValueError): DtlsCoapSession("host", 1234) # neither pair given @@ -49,10 +87,226 @@ def test_session_requires_exactly_one_cert_source(): with pytest.raises(ValueError): DtlsCoapSession("host", 1234, cert_pem=cert_pem) # key_pem missing + with pytest.raises(ValueError): + DtlsCoapSession("host", 1234, cert_path="/a") # key_path missing + + +def test_session_rejects_provider_with_legacy_certificate_arguments(): + cert_pem, key_pem = _make_generated_pem_chain() + auth = CertificateAuth.from_memory(cert_pem, key_pem) + + with pytest.raises(ValueError, match="auth or legacy certificate"): + DtlsCoapSession( + "host", + 1234, + cert_pem=cert_pem, + key_pem=key_pem, + auth=auth, + ) + + +def test_session_rejects_object_that_is_not_an_authentication_provider(): + with pytest.raises(TypeError, match="AuthenticationProvider"): + DtlsCoapSession("host", 1234, auth=object()) + + +def test_session_accepts_explicit_certificate_provider(): + cert_pem, key_pem = _make_generated_pem_chain() + auth = CertificateAuth.from_memory(cert_pem, key_pem) + session = DtlsCoapSession("host", 1234, auth=auth) + + assert session.auth is auth + assert session.cert_path is None + assert session.key_path is None + assert session.cert_pem is None + assert session.key_pem is None + + +def test_session_retains_authentication_provider_for_its_lifetime(): + class RetainedProvider: + def configure_context(self, _context): + return None + + auth = RetainedProvider() + reference = weakref.ref(auth) + session = DtlsCoapSession("host", 1234, auth=auth) + + del auth + gc.collect() + assert reference() is session.auth + def test_session_accepts_pem_pair(): - cert_pem, key_pem = _make_self_signed_pem_pair() + cert_pem, key_pem = _make_generated_pem_chain() sess = DtlsCoapSession("host", 1234, cert_pem=cert_pem, key_pem=key_pem) assert sess.cert_path is None assert sess.key_path is None assert sess.cert_pem == cert_pem + assert isinstance(sess.auth, CertificateAuth) + + +def test_session_routes_legacy_file_pair_through_certificate_auth(tmp_path): + cert_path = tmp_path / "client.pem" + key_path = tmp_path / "client-key.pem" + session = DtlsCoapSession( + "host", + 1234, + cert_path=cert_path, + key_path=key_path, + ) + + assert isinstance(session.auth, CertificateAuth) + assert session.cert_path == str(cert_path) + assert session.key_path == str(key_path) + + +def test_certificate_auth_loads_generated_chain_from_memory_and_files(tmp_path): + cert_pem, key_pem = _make_generated_pem_chain() + + memory_context = SSL.Context(SSL.DTLS_METHOD) + CertificateAuth.from_memory(cert_pem, key_pem).configure_context( + memory_context + ) + memory_context.check_privatekey() + + cert_path = tmp_path / "client.pem" + key_path = tmp_path / "client-key.pem" + cert_path.write_text(cert_pem) + key_path.write_text(key_pem) + file_context = SSL.Context(SSL.DTLS_METHOD) + CertificateAuth.from_files(cert_path, key_path).configure_context(file_context) + file_context.check_privatekey() + + +def test_certificate_auth_rejects_invalid_memory_material(): + auth = CertificateAuth.from_memory("not a certificate", "not a key") + with pytest.raises(ValueError, match="No certificates found"): + auth.configure_context(SSL.Context(SSL.DTLS_METHOD)) + + +def test_invalid_certificate_error_does_not_include_credential_material(): + marker = "credential" + "-marker" + certificate_blob = ( + "-----BEGIN CERTIFICATE-----\n" + f"{marker}\n" + "-----END CERTIFICATE-----\n" + ) + key_blob = "invalid-" + marker + auth = CertificateAuth.from_memory(certificate_blob, key_blob) + + with pytest.raises(crypto.Error) as captured: + auth.configure_context(SSL.Context(SSL.DTLS_METHOD)) + rendered = ( + str(captured.value) + + repr(captured.value) + + "".join(traceback.format_exception(captured.value)) + ) + assert marker not in rendered + + +def test_certificate_auth_rejects_invalid_file_material(tmp_path): + cert_path = tmp_path / "invalid.pem" + key_path = tmp_path / "invalid-key.pem" + cert_path.write_text("invalid") + key_path.write_text("invalid") + + with pytest.raises(SSL.Error): + CertificateAuth.from_files(cert_path, key_path).configure_context( + SSL.Context(SSL.DTLS_METHOD) + ) + + +def test_certificate_auth_rejects_incomplete_or_mixed_sources(): + with pytest.raises(ValueError): + CertificateAuth() + with pytest.raises(ValueError): + CertificateAuth(certificate_path="/synthetic/client.pem") + with pytest.raises(ValueError): + CertificateAuth(certificate_pem="certificate") + + certificate_path = "/synthetic/client.pem" + key_path = "/synthetic/client-key.pem" + certificate_data = "certificate" + key_data = "key" + with pytest.raises(ValueError): + CertificateAuth( + certificate_path=certificate_path, + private_key_path=key_path, + certificate_pem=certificate_data, + private_key_pem=key_data, + ) + + +def test_certificate_auth_is_immutable_and_has_secret_safe_repr(): + cert_pem, key_pem = _make_generated_pem_chain() + auth = CertificateAuth.from_memory(cert_pem, key_pem) + + rendered = repr(auth) + assert rendered == "CertificateAuth()" + assert cert_pem not in rendered + assert key_pem not in rendered + with pytest.raises(AttributeError, match="immutable"): + auth.certificate_pem = None + with pytest.raises(AttributeError, match="immutable"): + del auth._certificate_pem + + +def test_certificate_auth_has_no_public_or_dataclass_credential_surface(): + cert_pem, key_pem = _make_generated_pem_chain() + auth = CertificateAuth.from_memory(cert_pem, key_pem) + + assert not hasattr(auth, "certificate_pem") + assert not hasattr(auth, "private_key_pem") + with pytest.raises(TypeError): + vars(auth) + with pytest.raises(TypeError): + asdict(auth) + + +def test_certificate_auth_context_setup_matches_legacy_happy_path(): + class RecordingContext: + def __init__(self): + 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",)) + + context = RecordingContext() + CertificateAuth.from_files( + "/synthetic/client.pem", + "/synthetic/client-key.pem", + ).configure_context(context) + + assert [call[0] for call in context.calls] == [ + "load_verify_locations", + "set_verify", + "set_cipher_list", + "use_certificate_chain_file", + "use_privatekey_file", + "check_privatekey", + ] + assert context.calls[1] == ("set_verify", SSL.VERIFY_PEER) + assert context.calls[2] == ( + "set_cipher_list", + b"ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0", + ) + callback = context.verify_callback + assert callback(None, None, 0, 0, True) is True + assert callback(None, None, 1, 0, False) is False diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index 8467f6a..5125819 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -6,6 +6,7 @@ from smartthings_local.ocf.observe_refresh import ObserveRefreshTask from smartthings_local.ocf.state_cache import StateCache +from smartthings_local.protocol.auth import AuthenticationProvider, CertificateAuth from smartthings_local.protocol.dtls_session import DtlsCoapSession @@ -40,6 +41,14 @@ def test_dtls_session_constructor_keeps_file_memory_and_local_port_inputs(): "local_port", ], ) + auth_parameter = inspect.signature(DtlsCoapSession).parameters["auth"] + assert auth_parameter.kind is inspect.Parameter.KEYWORD_ONLY + assert auth_parameter.default is None + + +def test_certificate_auth_is_a_public_authentication_provider(): + provider = CertificateAuth.from_files("/synthetic/cert.pem", "/synthetic/key") + assert isinstance(provider, AuthenticationProvider) def test_dtls_session_keeps_current_consumer_methods():