From 79493fbd4899f4fde4ea463e0f088518f5384859 Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Sat, 15 Aug 2026 12:16:28 -0700 Subject: [PATCH] feat(protocol): add pure OwnerPSK derivation --- README.md | 29 ++++- smartthings_local/protocol/owner_psk.py | 126 +++++++++++++++++++++ tests/test_owner_psk.py | 140 ++++++++++++++++++++++++ tests/test_public_api_contract.py | 21 ++++ 4 files changed, 315 insertions(+), 1 deletion(-) create mode 100644 smartthings_local/protocol/owner_psk.py create mode 100644 tests/test_owner_psk.py diff --git a/README.md b/README.md index 77ba543..d686ced 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,32 @@ the key must be exactly 16 or 32 bytes. `PskAuth` selects only or persist credentials. Ownership transfer and credential discovery are outside this package. +Code that has already completed an authenticated manufacturer-certificate +session can derive IoTivity's 128-bit OwnerPSK from the resulting TLS state: + +```python +from smartthings_local.protocol.owner_psk import derive_mfg_certificate_owner_psk + +owner_psk = derive_mfg_certificate_owner_psk( + master_secret=master_secret, + client_random=client_random, + server_random=server_random, + owner_uuid=owner_uuid, + device_uuid=device_uuid, + cipher_name=cipher_name, + oxm_label=selected_oxm_label, +) +``` + +The caller must supply the exact authenticated TLS values, non-nil raw OCF +UUIDs, negotiated cipher name, and label for the selected OXM. Use +`STANDARD_MFG_CERTIFICATE_OXM_LABEL` for `oic.sec.doxm.mfgcert` and +`CONFIRMED_MFG_CERTIFICATE_OXM_LABEL` for +`x.org.iotivity.conmfgcert`; do not infer the label from the appliance model. +The helper performs deterministic key derivation only: it does not access a +session, discover credentials, choose an ownership method, write security +resources, run OTM, or persist the result. + ### Classified errors Runtime transport failures use the public types in @@ -605,6 +631,7 @@ smartthings_local/ The installable library — `pip install sm 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 dtls_handshake.py Shared memory-BIO handshake driver, bounded by a monotonic deadline (used by session + probe) + owner_psk.py Pure manufacturer-certificate OwnerPSK derivation ocf_root_ca.pem Samsung OCF root CA, bundled for handshake verification ocf/ OCF resource + state layer (reusable) __init__.py @@ -631,7 +658,7 @@ mqtt_demo/ MQTT bridge demo (consumes smartthings_loca .env.example Template — copy to .env, fill in setup_cert.py One-shot cert minting script (live-fetches AC14K_M + UUID) pyproject.toml Packaging — PyPI dist `smartthings-local`, hatch-vcs versioning -tests/ pytest suite (CoAP wire, state cache, import isolation, cert loading, DTLS probe, bridge port resolution, cert signing, certificate profiles, connect deadline, session interruption) +tests/ pytest suite (CoAP wire, state cache, import isolation, cert loading, DTLS probe, bridge port resolution, cert signing, certificate profiles, OwnerPSK derivation, connect deadline, session interruption) .github/workflows/publish.yml Build + PyPI Trusted Publishing on `v*` tags ``` diff --git a/smartthings_local/protocol/owner_psk.py b/smartthings_local/protocol/owner_psk.py new file mode 100644 index 0000000..5dfecde --- /dev/null +++ b/smartthings_local/protocol/owner_psk.py @@ -0,0 +1,126 @@ +"""Pure IoTivity manufacturer-certificate OwnerPSK derivation.""" + +from __future__ import annotations + +from collections.abc import Mapping +import hashlib +import hmac +from types import MappingProxyType +from typing import Final + + +CONFIRMED_MFG_CERTIFICATE_OXM_LABEL: Final = b"x.org.iotivity.conmfgcert" +STANDARD_MFG_CERTIFICATE_OXM_LABEL: Final = b"oic.sec.doxm.mfgcert" + +# OpenSSL cipher names mapped to the key-block lengths used by IoTivity's +# CAGenerateOwnerPSK implementation. +MFG_CERTIFICATE_KEY_BLOCK_LENGTHS: Final[Mapping[str, int]] = MappingProxyType( + { + "ECDHE-ECDSA-AES128-SHA256": 96, + "ECDHE-ECDSA-AES128-CCM": 40, + "ECDHE-ECDSA-AES128-CCM8": 40, + "ECDHE-ECDSA-AES128-GCM-SHA256": 120, + "AES256-SHA256": 128, + "ECDHE-ECDSA-AES256-SHA384": 160, + "ECDHE-ECDSA-AES256-GCM-SHA384": 184, + "AES128-GCM-SHA256": 120, + } +) + +_TLS_MASTER_SECRET_BYTES: Final = 48 +_TLS_RANDOM_BYTES: Final = 32 +_OCF_UUID_BYTES: Final = 16 +_OWNER_PSK_BYTES: Final = 16 + + +def _require_bytes(name: str, value: bytes, length: int) -> bytes: + if not isinstance(value, bytes): + raise TypeError(f"{name} must be bytes") + if len(value) != length: + raise ValueError(f"{name} must be exactly {length} bytes") + return value + + +def _require_uuid(name: str, value: bytes) -> bytes: + value = _require_bytes(name, value, _OCF_UUID_BYTES) + if not any(value): + raise ValueError(f"{name} must not be the nil UUID") + return value + + +def _tls12_p_hash_sha256( + key: bytes, + label: bytes, + random1: bytes, + random2: bytes, + length: int, +) -> bytes: + seed = label + random1 + random2 + a_value = hmac.new(key, seed, hashlib.sha256).digest() + output = bytearray() + while len(output) < length: + output.extend(hmac.new(key, a_value + seed, hashlib.sha256).digest()) + a_value = hmac.new(key, a_value, hashlib.sha256).digest() + return bytes(output[:length]) + + +def derive_mfg_certificate_owner_psk( + *, + master_secret: bytes, + client_random: bytes, + server_random: bytes, + owner_uuid: bytes, + device_uuid: bytes, + cipher_name: str, + oxm_label: bytes, +) -> bytes: + """Derive a 128-bit OwnerPSK from caller-supplied DTLS state. + + This implements IoTivity's two-stage TLS 1.2 SHA-256 P_hash operation. + It performs no session access, network I/O, ownership writes, or storage. + The caller must supply state from an authenticated manufacturer-certificate + session and explicitly select the OXM label used by that transaction. + """ + + if not isinstance(cipher_name, str): + raise TypeError("cipher_name must be a string") + key_block_bytes = MFG_CERTIFICATE_KEY_BLOCK_LENGTHS.get(cipher_name) + if key_block_bytes is None: + raise ValueError("unexpected manufacturer-certificate DTLS cipher") + + master_secret = _require_bytes( + "master_secret", master_secret, _TLS_MASTER_SECRET_BYTES + ) + client_random = _require_bytes( + "client_random", client_random, _TLS_RANDOM_BYTES + ) + server_random = _require_bytes( + "server_random", server_random, _TLS_RANDOM_BYTES + ) + owner_uuid = _require_uuid("owner_uuid", owner_uuid) + device_uuid = _require_uuid("device_uuid", device_uuid) + if not isinstance(oxm_label, bytes): + raise TypeError("oxm_label must be bytes") + if oxm_label not in { + CONFIRMED_MFG_CERTIFICATE_OXM_LABEL, + STANDARD_MFG_CERTIFICATE_OXM_LABEL, + }: + raise ValueError("unexpected manufacturer-certificate OXM label") + + key_block = _tls12_p_hash_sha256( + master_secret, + b"key expansion", + server_random, + client_random, + key_block_bytes, + ) + # IoTivity's OTM callers pass the owner UUID first and the target device + # UUID second. The lower adapter's historical rsrc/prov parameter names + # describe those arguments inconsistently, so preserve the caller order. + return _tls12_p_hash_sha256( + key_block, + oxm_label, + owner_uuid, + device_uuid, + _OWNER_PSK_BYTES, + ) diff --git a/tests/test_owner_psk.py b/tests/test_owner_psk.py new file mode 100644 index 0000000..4777ba9 --- /dev/null +++ b/tests/test_owner_psk.py @@ -0,0 +1,140 @@ +"""IoTivity manufacturer-certificate OwnerPSK derivation contracts.""" + +from __future__ import annotations + +import pytest + +from smartthings_local.protocol.owner_psk import ( + CONFIRMED_MFG_CERTIFICATE_OXM_LABEL, + MFG_CERTIFICATE_KEY_BLOCK_LENGTHS, + STANDARD_MFG_CERTIFICATE_OXM_LABEL, + derive_mfg_certificate_owner_psk, +) + + +_VALID_INPUTS = { + "master_secret": bytes(range(48)), + "client_random": bytes(range(32)), + "server_random": bytes(range(32, 64)), + "owner_uuid": bytes.fromhex("00112233445566778899aabbccddeeff"), + "device_uuid": bytes.fromhex("ffeeddccbbaa99887766554433221100"), + "cipher_name": "ECDHE-ECDSA-AES128-GCM-SHA256", + "oxm_label": CONFIRMED_MFG_CERTIFICATE_OXM_LABEL, +} + + +def test_fixed_iotivity_gcm_vector(): + assert derive_mfg_certificate_owner_psk(**_VALID_INPUTS).hex() == ( + "ccd6c618a91290dee8c106544ed79a33" + ) + + +def test_owner_then_device_uuid_order_matches_iotivity_callers(): + reversed_context = derive_mfg_certificate_owner_psk( + **{ + **_VALID_INPUTS, + "owner_uuid": _VALID_INPUTS["device_uuid"], + "device_uuid": _VALID_INPUTS["owner_uuid"], + } + ) + + assert reversed_context.hex() == "8f0f2416c483546dc1806db769b21b68" + assert reversed_context != derive_mfg_certificate_owner_psk(**_VALID_INPUTS) + + +def test_fixed_iotivity_ccm8_vector(): + inputs = { + **_VALID_INPUTS, + "cipher_name": "ECDHE-ECDSA-AES128-CCM8", + } + assert derive_mfg_certificate_owner_psk(**inputs).hex() == ( + "ddd3d945e266ee3dc27ff3a2c4321d32" + ) + + +def test_standard_and_confirmed_labels_derive_distinct_keys(): + confirmed = derive_mfg_certificate_owner_psk(**_VALID_INPUTS) + standard = derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, "oxm_label": STANDARD_MFG_CERTIFICATE_OXM_LABEL} + ) + + assert standard.hex() == "26ee1fe4c3e74509a2f5db5ab41b1e47" + assert standard != confirmed + + +def test_iotivity_cipher_key_block_lengths_are_immutable(): + assert dict(MFG_CERTIFICATE_KEY_BLOCK_LENGTHS) == { + "ECDHE-ECDSA-AES128-SHA256": 96, + "ECDHE-ECDSA-AES128-CCM": 40, + "ECDHE-ECDSA-AES128-CCM8": 40, + "ECDHE-ECDSA-AES128-GCM-SHA256": 120, + "AES256-SHA256": 128, + "ECDHE-ECDSA-AES256-SHA384": 160, + "ECDHE-ECDSA-AES256-GCM-SHA384": 184, + "AES128-GCM-SHA256": 120, + } + with pytest.raises(TypeError): + MFG_CERTIFICATE_KEY_BLOCK_LENGTHS["new-cipher"] = 1 + + +@pytest.mark.parametrize( + ("field", "length"), + [ + ("master_secret", 48), + ("client_random", 32), + ("server_random", 32), + ("owner_uuid", 16), + ("device_uuid", 16), + ], +) +def test_binary_inputs_require_exact_bytes_and_lengths(field, length): + with pytest.raises(TypeError, match=f"{field} must be bytes"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, field: bytearray(length)} + ) + for invalid_length in (length - 1, length + 1): + with pytest.raises(ValueError, match=f"exactly {length} bytes"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, field: b"x" * invalid_length} + ) + + +@pytest.mark.parametrize("field", ["owner_uuid", "device_uuid"]) +def test_nil_uuid_is_rejected(field): + with pytest.raises(ValueError, match="must not be the nil UUID"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, field: bytes(16)} + ) + + +def test_cipher_and_label_must_be_explicit_supported_values(): + with pytest.raises(TypeError, match="cipher_name must be a string"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, "cipher_name": b"cipher"} + ) + with pytest.raises(ValueError, match="unexpected.*cipher"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, "cipher_name": "ECDHE-RSA-AES128-GCM-SHA256"} + ) + with pytest.raises(TypeError, match="oxm_label must be bytes"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, "oxm_label": "oic.sec.doxm.mfgcert"} + ) + with pytest.raises(ValueError, match="unexpected.*label"): + derive_mfg_certificate_owner_psk( + **{**_VALID_INPUTS, "oxm_label": b"unsupported"} + ) + + +def test_failures_do_not_include_key_material(): + key_material = b"private-master-secret" + with pytest.raises(ValueError) as raised: + derive_mfg_certificate_owner_psk( + **{ + **_VALID_INPUTS, + "master_secret": key_material, + "cipher_name": "unsupported", + } + ) + assert key_material.hex() not in str(raised.value) + assert "private-master-secret" not in str(raised.value) diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index c244b91..712d393 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -18,6 +18,7 @@ ConnectCancellation, DtlsCoapSession, ) +from smartthings_local.protocol.owner_psk import derive_mfg_certificate_owner_psk def _assert_compatible_signature(callable_object, expected: list[str]) -> None: @@ -114,6 +115,26 @@ def test_psk_auth_is_a_public_authentication_provider(): ) +def test_owner_psk_derivation_keeps_every_security_input_explicit(): + parameters = inspect.signature( + derive_mfg_certificate_owner_psk + ).parameters + assert list(parameters) == [ + "master_secret", + "client_random", + "server_random", + "owner_uuid", + "device_uuid", + "cipher_name", + "oxm_label", + ] + assert all( + parameter.kind is inspect.Parameter.KEYWORD_ONLY + and parameter.default is inspect.Parameter.empty + for parameter in parameters.values() + ) + + def test_dtls_session_keeps_current_consumer_methods(): expected = { "close",