diff --git a/xtest/abac.py b/xtest/abac.py index 1933e0bd9..6b0bc680a 100644 --- a/xtest/abac.py +++ b/xtest/abac.py @@ -170,6 +170,9 @@ class KasGrantValue(BaseModelIgnoreExtra): "ec:secp256r1", "ec:secp384r1", "ec:secp521r1", + "hpqt:xwing", + "hpqt:secp256r1-mlkem768", + "hpqt:secp384r1-mlkem1024", ] KAS_PUBLIC_KEY_ALG_ENUM_RSA_2048 = 1 @@ -179,6 +182,9 @@ class KasGrantValue(BaseModelIgnoreExtra): KAS_PUBLIC_KEY_ALG_ENUM_EC_SECP384R1 = 6 KAS_PUBLIC_KEY_ALG_ENUM_EC_SECP521R1 = 7 +KAS_PUBLIC_KEY_ALG_ENUM_HPQT_XWING = 10 +KAS_PUBLIC_KEY_ALG_ENUM_HPQT_SECP256R1_MLKEM768 = 11 +KAS_PUBLIC_KEY_ALG_ENUM_HPQT_SECP384R1_MLKEM1024 = 12 _KAS_ALG_TO_STR_MAP = { KAS_PUBLIC_KEY_ALG_ENUM_RSA_2048: "rsa:2048", @@ -186,6 +192,9 @@ class KasGrantValue(BaseModelIgnoreExtra): KAS_PUBLIC_KEY_ALG_ENUM_EC_SECP256R1: "ec:secp256r1", KAS_PUBLIC_KEY_ALG_ENUM_EC_SECP384R1: "ec:secp384r1", KAS_PUBLIC_KEY_ALG_ENUM_EC_SECP521R1: "ec:secp521r1", + KAS_PUBLIC_KEY_ALG_ENUM_HPQT_XWING: "hpqt:xwing", + KAS_PUBLIC_KEY_ALG_ENUM_HPQT_SECP256R1_MLKEM768: "hpqt:secp256r1-mlkem768", + KAS_PUBLIC_KEY_ALG_ENUM_HPQT_SECP384R1_MLKEM1024: "hpqt:secp384r1-mlkem1024", } _STR_TO_KAS_ALG_MAP = {v: k for k, v in _KAS_ALG_TO_STR_MAP.items()} diff --git a/xtest/fixtures/keys.py b/xtest/fixtures/keys.py index a85161e3f..a07518895 100644 --- a/xtest/fixtures/keys.py +++ b/xtest/fixtures/keys.py @@ -208,6 +208,54 @@ def key_r4096( return _get_or_create_key(otdfctl, kas_entry_km1, "r4096", "rsa:4096", root_key) +@pytest.fixture(scope="module") +def key_xwing( + otdfctl: OpentdfCommandLineTool, + kas_entry_km1: abac.KasEntry, + root_key: str, +) -> abac.KasKey: + """Get or create X-Wing hybrid PQ/T managed key on km1.""" + return _get_or_create_key( + otdfctl, kas_entry_km1, "xwing", "hpqt:xwing", root_key, "mechanism-xwing" + ) + + +@pytest.fixture(scope="module") +def key_secpmlkem_3( + otdfctl: OpentdfCommandLineTool, + kas_entry_km1: abac.KasEntry, + root_key: str, +) -> abac.KasKey: + """Get or create hpqt:secp256r1-mlkem768 managed key on km1.""" + + return _get_or_create_key( + otdfctl, + kas_entry_km1, + "secp256r1-mlkem768", + "hpqt:secp256r1-mlkem768", + root_key, + "mechanism-secpmlkem", + ) + + +@pytest.fixture(scope="module") +def key_secpmlkem_5( + otdfctl: OpentdfCommandLineTool, + kas_entry_km1: abac.KasEntry, + root_key: str, +) -> abac.KasKey: + """Get or create hpqt:secp384r1-mlkem1024 managed key on km1.""" + + return _get_or_create_key( + otdfctl, + kas_entry_km1, + "secp384r1-mlkem1024", + "hpqt:secp384r1-mlkem1024", + root_key, + "mechanism-secpmlkem", + ) + + # --------------------------------------------------------------------------- # Attribute + key assignment fixtures (value-level) # --------------------------------------------------------------------------- @@ -259,6 +307,79 @@ def attribute_with_different_kids( return attr +@pytest.fixture(scope="module") +def attribute_with_xwing_key( + otdfctl: OpentdfCommandLineTool, + key_xwing: abac.KasKey, + otdf_client_scs: abac.SubjectConditionSet, + temporary_namespace: abac.Namespace, +) -> tuple[abac.Attribute, list[str]]: + """Create an ALL_OF attribute and assign an X-Wing key to it.""" + return _create_keyed_attribute( + otdfctl, + temporary_namespace, + "xwing-test", + [("xw1", key_xwing)], + otdf_client_scs, + "mechanism-xwing", + ) + + +@pytest.fixture(scope="module") +def attribute_with_xwing_and_ec_keys( + otdfctl: OpentdfCommandLineTool, + key_xwing: abac.KasKey, + managed_key_km2_ec: abac.KasKey, + otdf_client_scs: abac.SubjectConditionSet, + temporary_namespace: abac.Namespace, +) -> tuple[abac.Attribute, list[str]]: + """Create an ALL_OF attribute with both X-Wing and EC keys assigned.""" + return _create_keyed_attribute( + otdfctl, + temporary_namespace, + "xwing-hybrid-test", + [("xw1", key_xwing), ("ec1", managed_key_km2_ec)], + otdf_client_scs, + "mechanism-xwing", + ) + + +@pytest.fixture(scope="module") +def attribute_with_secpmlkem_3_key( + otdfctl: OpentdfCommandLineTool, + key_secpmlkem_3: abac.KasKey, + otdf_client_scs: abac.SubjectConditionSet, + temporary_namespace: abac.Namespace, +) -> tuple[abac.Attribute, list[str]]: + """Create an ALL_OF attribute and assign an secp256r1-mlkem768 key to it.""" + return _create_keyed_attribute( + otdfctl, + temporary_namespace, + "secpmlkem3-test", + [("m3", key_secpmlkem_3)], + otdf_client_scs, + "mechanism-secpmlkem", + ) + + +@pytest.fixture(scope="module") +def attribute_with_secpmlkem_5_key( + otdfctl: OpentdfCommandLineTool, + key_secpmlkem_5: abac.KasKey, + otdf_client_scs: abac.SubjectConditionSet, + temporary_namespace: abac.Namespace, +) -> tuple[abac.Attribute, list[str]]: + """Create an ALL_OF attribute and assign an secp384r1-mlkem1024 key to it.""" + return _create_keyed_attribute( + otdfctl, + temporary_namespace, + "secpmlkem5-test", + [("m5", key_secpmlkem_5)], + otdf_client_scs, + "mechanism-secpmlkem", + ) + + # --------------------------------------------------------------------------- # Attribute + key assignment fixture (attribute-level) # --------------------------------------------------------------------------- diff --git a/xtest/sdk/go/cli.sh b/xtest/sdk/go/cli.sh index 75b08728b..2984cbbd5 100755 --- a/xtest/sdk/go/cli.sh +++ b/xtest/sdk/go/cli.sh @@ -100,6 +100,16 @@ if [ "$1" == "supports" ]; then "${cmd[@]}" --version --json | jq -re .sdk_version | awk -F. '{ if ($1 > 0 || ($1 == 0 && $2 > 3) || ($1 == 0 && $2 == 3 && $3 >= 18)) exit 0; else exit 1; }' exit $? ;; + mechanism-xwing) + set -o pipefail + "${cmd[@]}" help policy kas-registry key create | grep -i hpqt:xwing + exit $? + ;; + mechanism-secpmlkem) + set -o pipefail + "${cmd[@]}" help policy kas-registry key create | grep -i hpqt:secp256r1-mlkem768 + exit $? + ;; *) echo "Unknown feature: $2" exit 2 diff --git a/xtest/sdk/java/cli.sh b/xtest/sdk/java/cli.sh index 7410862cf..0f5758c50 100755 --- a/xtest/sdk/java/cli.sh +++ b/xtest/sdk/java/cli.sh @@ -102,6 +102,11 @@ if [ "$1" == "supports" ]; then java -jar "$SCRIPT_DIR"/cmdline.jar --version | jq -re .version | awk -F. '{ if ($1 > 0 || ($1 == 0 && $2 >= 13)) exit 0; else exit 1; }' exit $? ;; + mechanism-xwing) + set -o pipefail + java -jar "$SCRIPT_DIR"/cmdline.jar help encrypt | grep -i xwing + exit $? + ;; *) echo "Unknown feature: $2" exit 2 diff --git a/xtest/sdk/js/cli.sh b/xtest/sdk/js/cli.sh index cf418d208..354596800 100755 --- a/xtest/sdk/js/cli.sh +++ b/xtest/sdk/js/cli.sh @@ -91,6 +91,11 @@ if [ "$1" == "supports" ]; then npx $CTL --version | jq -re '.["@opentdf/sdk"]' | awk -F. '{ if ($1 > 0 || ($1 == 0 && $2 >= 9)) exit 0; else exit 1; }' exit $? ;; + mechanism-xwing) + set -o pipefail + npx $CTL help | grep -i xwing + exit $? + ;; *) echo "Unknown feature: $2" exit 2 diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 56e4c11b1..0c4c9c611 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -51,6 +51,10 @@ "mechanism-rsa-4096", # Support for encrypting with EC curves secp384r1 and secp521r1 managed keys. "mechanism-ec-curves-384-521", + # Support for encrypting with X-Wing hybrid post-quantum/traditional KEM. + "mechanism-xwing", + # Support for encrypting with hybrid post-quantum/traditional KEM with NIST Elliptic Curves. + "mechanism-secpmlkem", "ns_grants", "obligations", ] @@ -127,6 +131,11 @@ def __init__(self, **kwargs: dict[str, Any]): if self.semver >= (0, 13, 0): self.features.add("mechanism-ec-curves-384-521") + # X-Wing hybrid PQ/T KEM support (ML-KEM-768 + X25519) + if self.semver >= (0, 14, 0): + self.features.add("mechanism-xwing") + self.features.add("mechanism-secpmlkem") + print(f"PLATFORM_VERSION '{v}' supports [{', '.join(self.features)}]") def skip_if_unsupported(self, *features: feature_type): diff --git a/xtest/test_pqc.py b/xtest/test_pqc.py new file mode 100644 index 000000000..dc8536e1e --- /dev/null +++ b/xtest/test_pqc.py @@ -0,0 +1,324 @@ +"""Tests for hybrid post-quantum/traditional KEM. + +These tests verify that TDF encryption and decryption work correctly when +X-Wing and NIST approved hybrid managed keys are assigned to attributes via the policy service. +""" + +import base64 +import filecmp +from pathlib import Path + +import pytest + +import tdfs +from abac import Attribute, KasKey +from tdfs import KeyAccessObject + +cipherTexts: dict[str, Path] = {} + +# X-Wing KEM sizes per draft-connolly-cfrg-xwing-kem-10 +XWING_ENCAPSULATION_KEY_SIZE = 1216 # public key, bytes +XWING_CIPHERTEXT_SIZE = 1120 # KEM ciphertext (wrappedKey), bytes + + +def _b64_decoded_len(s: str) -> int: + """Return the byte length of a base64-encoded string.""" + return len(base64.b64decode(s)) + + +def _pem_decoded_len(pem: str) -> int: + """Return the byte length of the DER payload inside a PEM block.""" + lines = [ln for ln in pem.strip().splitlines() if not ln.startswith("-----")] + return len(base64.b64decode("".join(lines))) + + +def assert_xwing_kao_sizes(kao: KeyAccessObject): + """Assert that an X-Wing KAO has correctly sized wrappedKey and ephemeralPublicKey.""" + wrapped_len = _b64_decoded_len(kao.wrappedKey) + assert wrapped_len == XWING_CIPHERTEXT_SIZE, ( + f"X-Wing wrappedKey should be {XWING_CIPHERTEXT_SIZE} bytes, got {wrapped_len}" + ) + assert kao.ephemeralPublicKey is not None, ( + "X-Wing KAO must include an ephemeralPublicKey" + ) + epk_len = _b64_decoded_len(kao.ephemeralPublicKey) + assert epk_len == XWING_ENCAPSULATION_KEY_SIZE, ( + f"X-Wing ephemeralPublicKey should be {XWING_ENCAPSULATION_KEY_SIZE} bytes, got {epk_len}" + ) + + +def assert_xwing_public_key_size(kas_key: KasKey): + """Assert that the KAS registry public key for X-Wing is the expected size.""" + pem = kas_key.key.public_key_ctx.pem + der_len = _pem_decoded_len(pem) + assert der_len >= XWING_ENCAPSULATION_KEY_SIZE, ( + f"X-Wing public key DER should be >= {XWING_ENCAPSULATION_KEY_SIZE} bytes, got {der_len}" + ) + + +def test_xwing_roundtrip( + attribute_with_xwing_key: tuple[Attribute, list[str]], + key_xwing: KasKey, + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + tmp_dir: Path, + pt_file: Path, + kas_url_km1: str, + in_focus: set[tdfs.SDK], +): + """Encrypt and decrypt with an X-Wing managed key.""" + if not in_focus & {encrypt_sdk, decrypt_sdk}: + pytest.skip("Not in focus") + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("key_management", "autoconfigure", "mechanism-xwing") + encrypt_sdk.skip_if_unsupported( + "key_management", "autoconfigure", "mechanism-xwing" + ) + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) + + attr, key_ids = attribute_with_xwing_key + + sample_name = f"xwing-{encrypt_sdk}" + if sample_name in cipherTexts: + ct_file = cipherTexts[sample_name] + else: + ct_file = tmp_dir / f"{sample_name}.tdf" + cipherTexts[sample_name] = ct_file + encrypt_sdk.encrypt( + pt_file, + ct_file, + mime_type="text/plain", + container="ztdf", + attr_values=attr.value_fqns, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + ) + + manifest = tdfs.manifest(ct_file) + assert len(manifest.encryptionInformation.keyAccess) == 1 + + manifest_kids = {kao.kid for kao in manifest.encryptionInformation.keyAccess} + expected_kids = set(key_ids) + assert manifest_kids == expected_kids, ( + f"Expected key IDs {expected_kids} but got {manifest_kids}" + ) + + manifest_urls = {kao.url for kao in manifest.encryptionInformation.keyAccess} + assert kas_url_km1 in manifest_urls + + # Verify X-Wing KEM sizes in the KAO and registered public key + kao = manifest.encryptionInformation.keyAccess[0] + assert_xwing_kao_sizes(kao) + assert_xwing_public_key_size(key_xwing) + + rt_file = tmp_dir / f"xwing-{encrypt_sdk}-{decrypt_sdk}.untdf" + decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") + assert filecmp.cmp(pt_file, rt_file) + + +def test_xwing_with_ec_roundtrip( + attribute_with_xwing_and_ec_keys: tuple[Attribute, list[str]], + key_xwing: KasKey, + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + tmp_dir: Path, + pt_file: Path, + kas_url_km1: str, + kas_url_km2: str, + in_focus: set[tdfs.SDK], +): + """Encrypt and decrypt with both X-Wing and EC keys (multi-mechanism).""" + if not in_focus & {encrypt_sdk, decrypt_sdk}: + pytest.skip("Not in focus") + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("key_management", "autoconfigure", "mechanism-xwing") + encrypt_sdk.skip_if_unsupported( + "key_management", "autoconfigure", "mechanism-xwing" + ) + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) + + attr, key_ids = attribute_with_xwing_and_ec_keys + + sample_name = f"xwing-ec-{encrypt_sdk}" + if sample_name in cipherTexts: + ct_file = cipherTexts[sample_name] + else: + ct_file = tmp_dir / f"{sample_name}.tdf" + cipherTexts[sample_name] = ct_file + encrypt_sdk.encrypt( + pt_file, + ct_file, + mime_type="text/plain", + container="ztdf", + attr_values=attr.value_fqns, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + ) + + manifest = tdfs.manifest(ct_file) + assert len(manifest.encryptionInformation.keyAccess) == 2 + + manifest_kids = {kao.kid for kao in manifest.encryptionInformation.keyAccess} + expected_kids = set(key_ids) + assert manifest_kids == expected_kids, ( + f"Expected key IDs {expected_kids} but got {manifest_kids}" + ) + + manifest_urls = {kao.url for kao in manifest.encryptionInformation.keyAccess} + assert manifest_urls <= {kas_url_km1, kas_url_km2}, ( + f"Expected KAS URLs from km1 or km2, but got {manifest_urls}" + ) + + # Verify X-Wing KEM sizes on the xwing KAO + xwing_kid = key_xwing.key.key_id + xwing_kao = next( + kao for kao in manifest.encryptionInformation.keyAccess if kao.kid == xwing_kid + ) + assert xwing_kao is not None, ( + f"X-Wing KAO with kid={xwing_kid} not found in manifest" + ) + + assert_xwing_kao_sizes(xwing_kao) + assert_xwing_public_key_size(key_xwing) + + if any( + kao.type == "ec-wrapped" for kao in manifest.encryptionInformation.keyAccess + ): + tdfs.skip_if_unsupported(decrypt_sdk, "ecwrap") + rt_file = tmp_dir / f"xwing-ec-{encrypt_sdk}-{decrypt_sdk}.untdf" + decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") + assert filecmp.cmp(pt_file, rt_file) + + +def test_secpmlkem_3_roundtrip( + attribute_with_secpmlkem_3_key: tuple[Attribute, list[str]], + key_secpmlkem_3: KasKey, + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + tmp_dir: Path, + pt_file: Path, + kas_url_km1: str, + in_focus: set[tdfs.SDK], +): + """Encrypt and decrypt with an X-Wing managed key.""" + if not in_focus & {encrypt_sdk, decrypt_sdk}: + pytest.skip("Not in focus") + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("key_management", "autoconfigure", "mechanism-secpmlkem") + encrypt_sdk.skip_if_unsupported( + "key_management", "autoconfigure", "mechanism-secpmlkem" + ) + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) + + attr, key_ids = attribute_with_secpmlkem_3_key + + sample_name = f"secpmlkem_3-{encrypt_sdk}" + if sample_name in cipherTexts: + ct_file = cipherTexts[sample_name] + else: + ct_file = tmp_dir / f"{sample_name}.tdf" + cipherTexts[sample_name] = ct_file + encrypt_sdk.encrypt( + pt_file, + ct_file, + mime_type="text/plain", + container="ztdf", + attr_values=attr.value_fqns, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + ) + + manifest = tdfs.manifest(ct_file) + assert len(manifest.encryptionInformation.keyAccess) == 1 + + manifest_kids = {kao.kid for kao in manifest.encryptionInformation.keyAccess} + expected_kids = set(key_ids) + assert manifest_kids == expected_kids, ( + f"Expected key IDs {expected_kids} but got {manifest_kids}" + ) + + manifest_urls = {kao.url for kao in manifest.encryptionInformation.keyAccess} + assert kas_url_km1 in manifest_urls + + # Verify NIST curve compatible MLKEM hybrid sizes in the KAO and registered public key + kao = manifest.encryptionInformation.keyAccess[0] + wrapped_len = _b64_decoded_len(kao.wrappedKey) + assert wrapped_len > XWING_CIPHERTEXT_SIZE, ( + f"wrappedKey should be larger than {XWING_CIPHERTEXT_SIZE} bytes, got {wrapped_len}" + ) + pem = key_secpmlkem_3.key.public_key_ctx.pem + der_len = _pem_decoded_len(pem) + assert der_len >= XWING_ENCAPSULATION_KEY_SIZE, ( + f"public key DER should be >= {XWING_ENCAPSULATION_KEY_SIZE} bytes, got {der_len}" + ) + + rt_file = tmp_dir / f"secpmlkem_3-{encrypt_sdk}-{decrypt_sdk}.untdf" + decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") + assert filecmp.cmp(pt_file, rt_file) + + +def test_secpmlkem_5_roundtrip( + attribute_with_secpmlkem_5_key: tuple[Attribute, list[str]], + key_secpmlkem_5: KasKey, + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + tmp_dir: Path, + pt_file: Path, + kas_url_km1: str, + in_focus: set[tdfs.SDK], +): + """Encrypt and decrypt with an X-Wing managed key.""" + if not in_focus & {encrypt_sdk, decrypt_sdk}: + pytest.skip("Not in focus") + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("key_management", "autoconfigure", "mechanism-secpmlkem") + encrypt_sdk.skip_if_unsupported( + "key_management", "autoconfigure", "mechanism-secpmlkem" + ) + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) + + attr, key_ids = attribute_with_secpmlkem_5_key + + sample_name = f"secpmlkem_3-{encrypt_sdk}" + if sample_name in cipherTexts: + ct_file = cipherTexts[sample_name] + else: + ct_file = tmp_dir / f"{sample_name}.tdf" + cipherTexts[sample_name] = ct_file + encrypt_sdk.encrypt( + pt_file, + ct_file, + mime_type="text/plain", + container="ztdf", + attr_values=attr.value_fqns, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + ) + + manifest = tdfs.manifest(ct_file) + assert len(manifest.encryptionInformation.keyAccess) == 1 + + manifest_kids = {kao.kid for kao in manifest.encryptionInformation.keyAccess} + expected_kids = set(key_ids) + assert manifest_kids == expected_kids, ( + f"Expected key IDs {expected_kids} but got {manifest_kids}" + ) + + manifest_urls = {kao.url for kao in manifest.encryptionInformation.keyAccess} + assert kas_url_km1 in manifest_urls + + # Verify NIST curve compatible MLKEM hybrid sizes in the KAO and registered public key + kao = manifest.encryptionInformation.keyAccess[0] + wrapped_len = _b64_decoded_len(kao.wrappedKey) + assert wrapped_len > XWING_CIPHERTEXT_SIZE, ( + f"wrappedKey should be larger than {XWING_CIPHERTEXT_SIZE} bytes, got {wrapped_len}" + ) + pem = key_secpmlkem_5.key.public_key_ctx.pem + der_len = _pem_decoded_len(pem) + assert der_len >= XWING_ENCAPSULATION_KEY_SIZE, ( + f"public key DER should be >= {XWING_ENCAPSULATION_KEY_SIZE} bytes, got {der_len}" + ) + + rt_file = tmp_dir / f"secpmlkem_3-{encrypt_sdk}-{decrypt_sdk}.untdf" + decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") + assert filecmp.cmp(pt_file, rt_file)