feat(xtest): adds hybrid PQ/T KEM round trip testing

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,13 +182,19 @@ 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",
     KAS_PUBLIC_KEY_ALG_ENUM_RSA_4096: "rsa:4096",
     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()}
 

xtest/fixtures/keys.py

@@ -2,7 +2,7 @@
 
 This module contains fixtures for:
 - Extra keys loaded from extra-keys.json
-- Managed key creation and assignment
+- Managed key creation and assignment (RSA, EC, X-Wing)
 - Public key registration
 - Legacy key imports
 - Base key configuration
@@ -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 X-Wing hybrid 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 X-Wing hybrid 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 X-Wing 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 X-Wing 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)
 # ---------------------------------------------------------------------------

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

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

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

xtest/tdfs.py

@@ -51,6 +51,9 @@
     "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",
+    "mechanism-secpmlkem",
     "ns_grants",
     "obligations",
 ]
@@ -127,8 +130,21 @@ 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 require(self, *features: feature_type):
+        """Skip the current test if any of the given features are unsupported."""
+        for feature in features:
+            if feature not in self.features:
+                pytest.skip(
+                    f"platform service {self.version} doesn't yet support [{feature}]"
+                )
+
     def skip_if_unsupported(self, *features: feature_type):
         """Skip the current test if any of the given features are unsupported."""
         missing = [f for f in features if f not in self.features]
@@ -506,6 +522,18 @@ def all_versions_of(sdk: sdk_type) -> list[SDK]:
     return versions
 
 
+def parse_sdk_spec(spec: str) -> list[SDK]:
+    """Parse an SDK spec like 'go' or 'go:gemini' into a list of SDK objects.
+
+    Bare names (e.g., 'go') return all discovered versions.
+    Qualified names (e.g., 'go:gemini') return a single specific version.
+    """
+    if ":" in spec:
+        sdk_name, version = spec.split(":", 1)
+        return [SDK(sdk_name, version)]
+    return all_versions_of(spec)
+
+
 def skip_if_unsupported(sdk: SDK, *features: feature_type):
     pfs = get_platform_features()
     pfs.skip_if_unsupported(*features)