feat(protocol): add pure OwnerPSK derivation

2026-08-15 12:24:11 -07:00
parent bc4465b274
commit 79493fbd48
4 changed files with 315 additions and 1 deletions
@@ -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
 ```

@@ -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,
+    )
@@ -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)
@@ -18,6 +18,7 @@ from smartthings_local.protocol.dtls_session import (
    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",