feat(protocol): add bound server certificate profile
This commit is contained in:
Jason Morcos
@@ -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",
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user