bhetherman/SmartThings-Local
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #33 from Moballo-LLC/codex/py-07-certificate-profiles

Browse Source 
Add bound Samsung server certificate profile
This commit is contained in:
Quite Yellow
2026-08-15 10:20:14 +01:00
committed by GitHub
parent 9ef5598813 83a5973434
commit e9aee1c235
4 changed files with 1453 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+70
View File
@@ -81,6 +81,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`
smartthings_local/protocol/auth.py
+334 -9
View File
@@ -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<device_identity>[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 @@ class AuthenticationProvider(Protocol):
"""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 @@ class CertificateAuth:
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 @@ class CertificateAuth:
"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 @@ class CertificateAuth:
)
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 @@ class CertificateAuth:
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 @@ class CertificateAuth:
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 @@ class 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)
_configure_certificate_server(context, self._server_profile)
if self._certificate_pem is not None:
_load_pem_chain(
context,
@@ -240,7 +558,14 @@ class PskAuth:
"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",
]
tests/test_certificate_profiles.py
+1005
View File
File diff suppressed because it is too large Load Diff
tests/test_public_api_contract.py
+44
View File
@@ -10,6 +10,9 @@ from smartthings_local.protocol.auth import (
AuthenticationProvider,
CertificateAuth,
PskAuth,
SamsungServerProfile,
SamsungServerRole,
ServerCertificateAuth,
)
from smartthings_local.protocol.dtls_session import (
ConnectCancellation,
@@ -57,6 +60,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)