diff --git a/xtest/abac.py b/xtest/abac.py index ca5a43ef2..1933e0bd9 100644 --- a/xtest/abac.py +++ b/xtest/abac.py @@ -1,4 +1,5 @@ import enum +from typing import Literal from pydantic import BaseModel, ConfigDict, Field, field_validator @@ -163,6 +164,14 @@ class KasGrantValue(BaseModelIgnoreExtra): key_access_server_id: str | None = None +kas_algorithm_type = Literal[ + "rsa:2048", + "rsa:4096", + "ec:secp256r1", + "ec:secp384r1", + "ec:secp521r1", +] + KAS_PUBLIC_KEY_ALG_ENUM_RSA_2048 = 1 KAS_PUBLIC_KEY_ALG_ENUM_RSA_4096 = 2 diff --git a/xtest/fixtures/attributes.py b/xtest/fixtures/attributes.py index 5ae2bea85..bffe89503 100644 --- a/xtest/fixtures/attributes.py +++ b/xtest/fixtures/attributes.py @@ -44,7 +44,7 @@ def attribute_single_kas_grant( temporary_namespace: abac.Namespace, ): """Attribute with single KAS grant on value 'a'.""" - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() anyof = otdfctl.attribute_create( temporary_namespace, "letter", abac.AttributeRule.ANY_OF, ["a"] ) @@ -90,7 +90,7 @@ def attribute_two_kas_grant_or( assert sm.attribute_value.value == "alpha" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_value(kas_entry_alpha, alpha) otdfctl.grant_assign_value(kas_entry_beta, beta) else: @@ -132,7 +132,7 @@ def attribute_two_kas_grant_and( assert sm2.attribute_value.value == "bet" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_value(kas_entry_alpha, alef) otdfctl.grant_assign_value(kas_entry_beta, bet) else: @@ -171,7 +171,7 @@ def one_attribute_attr_kas_grant( assert sm.attribute_value.value == "alpha" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_attr(kas_entry_gamma, anyof) else: kas_key_alpha = otdfctl.kas_registry_create_public_key_only( @@ -292,7 +292,7 @@ def attr_and_value_kas_grants_or( assert sm.attribute_value.value == "alpha" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_attr(kas_entry_gamma, anyof) otdfctl.grant_assign_value(kas_entry_alpha, beta) else: @@ -337,7 +337,7 @@ def attr_and_value_kas_grants_and( assert sm2.attribute_value.value == "beta" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_attr(kas_entry_gamma, allof) otdfctl.grant_assign_value(kas_entry_alpha, beta) else: @@ -375,7 +375,7 @@ def one_attribute_ns_kas_grant( sm = otdfctl.scs_map(otdf_client_scs, alpha) assert sm.attribute_value.value == "alpha" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_ns(kas_entry_delta, temporary_namespace) else: kas_key_ns = otdfctl.kas_registry_create_public_key_only( @@ -395,7 +395,7 @@ def attribute_missing_value_key_mapping( root_key: str, ) -> tuple[str, str]: """Attribute with attribute-level managed key mapping and a missing value FQN.""" - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() if "key_management" not in pfs.features: pytest.skip("Key management not supported by platform") @@ -449,7 +449,7 @@ def ns_and_value_kas_grants_or( assert sm.attribute_value.value == "alpha" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_value(kas_entry_alpha, beta) otdfctl.grant_assign_ns(kas_entry_delta, temp_namespace) else: @@ -494,7 +494,7 @@ def ns_and_value_kas_grants_and( assert sm2.attribute_value.value == "beta" # Now assign it to the current KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_value(kas_entry_alpha, beta) otdfctl.grant_assign_ns(kas_entry_delta, temp_namespace) else: @@ -538,7 +538,7 @@ def attribute_default_rsa( assert sm.attribute_value.value == "wrapped" # Assign RSA key on default KAS - if "key_management" not in tdfs.PlatformFeatureSet().features: + if "key_management" not in tdfs.get_platform_features().features: otdfctl.grant_assign_value(kas_entry_default, wrapped) else: kas_key = otdfctl.kas_registry_create_public_key_only( diff --git a/xtest/fixtures/audit.py b/xtest/fixtures/audit.py index 48557bdfc..2d0783bfc 100644 --- a/xtest/fixtures/audit.py +++ b/xtest/fixtures/audit.py @@ -67,10 +67,10 @@ def audit_log_config(request: pytest.FixtureRequest) -> AuditLogConfig: KAS_ALPHA_LOG_FILE, KAS_BETA_LOG_FILE, etc: Paths to additional KAS log files """ # Import here to avoid circular dependency - from tdfs import PlatformFeatureSet + from tdfs import get_platform_features # Check if platform version supports audit logging - pfs = PlatformFeatureSet() + pfs = get_platform_features() platform_supports_audit = "audit_logging" in pfs.features # Check if disabled via CLI or environment variable diff --git a/xtest/fixtures/keys.py b/xtest/fixtures/keys.py index 9964c7083..a85161e3f 100644 --- a/xtest/fixtures/keys.py +++ b/xtest/fixtures/keys.py @@ -34,6 +34,71 @@ def _key_id_suffix(wrapping_key: str) -> str: return hashlib.sha256(wrapping_key.encode()).hexdigest()[:8] +def _get_or_create_key( + otdfctl: OpentdfCommandLineTool, + kas_entry: abac.KasEntry, + key_id_prefix: str, + algorithm: abac.kas_algorithm_type, + root_key: str, + *required_features: tdfs.feature_type, +) -> abac.KasKey: + """Get or create a managed key, skipping if required platform features are missing. + + Key ID is "{prefix}-{root_key_hash}" to ensure uniqueness across root key changes. + """ + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("key_management", *required_features) + + key_id = f"{key_id_prefix}-{_key_id_suffix(root_key)}" + existing_keys = otdfctl.kas_registry_keys_list(kas_entry) + key = next((k for k in existing_keys if k.key.key_id == key_id), None) + if key is None: + key = otdfctl.kas_registry_create_key( + kas_entry, + key_id=key_id, + mode="local", + algorithm=algorithm, + wrapping_key=root_key, + wrapping_key_id="root", + ) + return key + + +def _create_keyed_attribute( + otdfctl: OpentdfCommandLineTool, + namespace: abac.Namespace, + attr_name: str, + value_key_pairs: list[tuple[str, abac.KasKey]], + scs: abac.SubjectConditionSet, + *required_features: tdfs.feature_type, +) -> tuple[abac.Attribute, list[str]]: + """Create an ALL_OF attribute, SCS-map each value, and assign keys at value level. + + Returns (attribute, [key_id, ...]). + """ + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("key_management", *required_features) + + value_names = [name for name, _ in value_key_pairs] + attr = otdfctl.attribute_create( + namespace, attr_name, abac.AttributeRule.ALL_OF, value_names + ) + assert attr.values and len(attr.values) == len(value_key_pairs) + + for val, (expected_name, key) in zip(attr.values, value_key_pairs, strict=True): + assert val.value == expected_name + sm = otdfctl.scs_map(scs, val) + assert sm.attribute_value.value == val.value + otdfctl.key_assign_value(key, val) + + return (attr, [key.key.key_id for _, key in value_key_pairs]) + + +# --------------------------------------------------------------------------- +# Extra keys (loaded from JSON) +# --------------------------------------------------------------------------- + + class ExtraKey(typing.TypedDict): """TypedDict for extra keys in extra-keys.json""" @@ -66,34 +131,19 @@ def pick_extra_key(extra_keys: dict[str, ExtraKey], kid: str) -> abac.KasPublicK ) +# --------------------------------------------------------------------------- +# Managed key fixtures +# --------------------------------------------------------------------------- + + @pytest.fixture(scope="module") def managed_key_km1_rsa( otdfctl: OpentdfCommandLineTool, kas_entry_km1: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create RSA managed key on km1. - - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - - key_id = f"km1-rsa-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km1) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km1, - key_id=key_id, - mode="local", - algorithm="rsa:2048", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key + """Get or create RSA managed key on km1.""" + return _get_or_create_key(otdfctl, kas_entry_km1, "km1-rsa", "rsa:2048", root_key) @pytest.fixture(scope="module") @@ -102,28 +152,10 @@ def managed_key_km2_ec( kas_entry_km2: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create EC managed key on km2. - - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - - key_id = f"km2-ec-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km2) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km2, - key_id=key_id, - mode="local", - algorithm="ec:secp256r1", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key + """Get or create EC managed key on km2.""" + return _get_or_create_key( + otdfctl, kas_entry_km2, "km2-ec", "ec:secp256r1", root_key + ) @pytest.fixture(scope="module") @@ -132,28 +164,8 @@ def key_e256( kas_entry_km2: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create EC secp256r1 managed key on km2. - - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - - key_id = f"e256-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km2) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km2, - key_id=key_id, - mode="local", - algorithm="ec:secp256r1", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key + """Get or create EC secp256r1 managed key on km2.""" + return _get_or_create_key(otdfctl, kas_entry_km2, "e256", "ec:secp256r1", root_key) @pytest.fixture(scope="module") @@ -162,28 +174,8 @@ def key_e384( kas_entry_km2: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create EC secp384r1 managed key on km2 - - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - - key_id = f"e384-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km2) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km2, - key_id=key_id, - mode="local", - algorithm="ec:secp384r1", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key + """Get or create EC secp384r1 managed key on km2.""" + return _get_or_create_key(otdfctl, kas_entry_km2, "e384", "ec:secp384r1", root_key) @pytest.fixture(scope="module") @@ -192,28 +184,8 @@ def key_e521( kas_entry_km2: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create EC secp521r1 managed key on km2. - - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - - key_id = f"e521-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km2) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km2, - key_id=key_id, - mode="local", - algorithm="ec:secp521r1", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key + """Get or create EC secp521r1 managed key on km2.""" + return _get_or_create_key(otdfctl, kas_entry_km2, "e521", "ec:secp521r1", root_key) @pytest.fixture(scope="module") @@ -222,28 +194,8 @@ def key_r2048( kas_entry_km1: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create RSA 2048 managed key on km1. - - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - - key_id = f"r2048-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km1) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km1, - key_id=key_id, - mode="local", - algorithm="rsa:2048", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key + """Get or create RSA 2048 managed key on km1.""" + return _get_or_create_key(otdfctl, kas_entry_km1, "r2048", "rsa:2048", root_key) @pytest.fixture(scope="module") @@ -252,28 +204,13 @@ def key_r4096( kas_entry_km1: abac.KasEntry, root_key: str, ) -> abac.KasKey: - """Get or create RSA 4096 managed key on km1. + """Get or create RSA 4096 managed key on km1.""" + return _get_or_create_key(otdfctl, kas_entry_km1, "r4096", "rsa:4096", root_key) - Key ID includes a hash of the root key to ensure that if the root key changes, - a new key will be created instead of reusing an incompatible one. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled") - key_id = f"r4096-{_key_id_suffix(root_key)}" - existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km1) - key = next((k for k in existing_keys if k.key.key_id == key_id), None) - if key is None: - key = otdfctl.kas_registry_create_key( - kas_entry_km1, - key_id=key_id, - mode="local", - algorithm="rsa:4096", - wrapping_key=root_key, - wrapping_key_id="root", - ) - return key +# --------------------------------------------------------------------------- +# Attribute + key assignment fixtures (value-level) +# --------------------------------------------------------------------------- @pytest.fixture(scope="module") @@ -287,63 +224,44 @@ def attribute_allof_with_extended_mechanisms( otdf_client_scs: abac.SubjectConditionSet, temporary_namespace: abac.Namespace, ) -> tuple[abac.Attribute, list[str]]: - """Create an ALL_OF attribute and assign extended mechanism keys to it. - - - Uses ec:secp256r1, ec:secp384r1, ec:secp521r1, and rsa:2048, rsa:4096 keys - - Reuses existing managed keys - - Assigns all keys to attribute values (value-level assignment) - - Maps all attribute values to the client SCS - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip( - "Key management feature is not enabled; skipping key assignment fixture" - ) - - # Create attribute with three values under ALL_OF - attr = otdfctl.attribute_create( + """Create an ALL_OF attribute and assign extended mechanism keys to it.""" + return _create_keyed_attribute( + otdfctl, temporary_namespace, "mechanism-select", - abac.AttributeRule.ALL_OF, - ["ec-secp256r1", "ec-secp384r1", "ec-secp521r1", "rsa-2048", "rsa-4096"], - ) - assert attr.values and len(attr.values) == 5 - v_e256, v_e384, v_e521, v_r2048, v_r4096 = attr.values - assert v_e256.value == "ec-secp256r1" - assert v_e384.value == "ec-secp384r1" - assert v_e521.value == "ec-secp521r1" - assert v_r2048.value == "rsa-2048" - assert v_r4096.value == "rsa-4096" - - # Ensure client has access to all values - sm1 = otdfctl.scs_map(otdf_client_scs, v_e256) - assert sm1.attribute_value.value == v_e256.value - sm2 = otdfctl.scs_map(otdf_client_scs, v_e384) - assert sm2.attribute_value.value == v_e384.value - sm3 = otdfctl.scs_map(otdf_client_scs, v_e521) - assert sm3.attribute_value.value == v_e521.value - sm4 = otdfctl.scs_map(otdf_client_scs, v_r2048) - assert sm4.attribute_value.value == v_r2048.value - sm5 = otdfctl.scs_map(otdf_client_scs, v_r4096) - assert sm5.attribute_value.value == v_r4096.value - - # Assign keys to corresponding attribute values - otdfctl.key_assign_value(key_e256, v_e256) - otdfctl.key_assign_value(key_e384, v_e384) - otdfctl.key_assign_value(key_e521, v_e521) - otdfctl.key_assign_value(key_r2048, v_r2048) - otdfctl.key_assign_value(key_r4096, v_r4096) - - return ( - attr, [ - key_e256.key.key_id, - key_e384.key.key_id, - key_e521.key.key_id, - key_r2048.key.key_id, - key_r4096.key.key_id, + ("ec-secp256r1", key_e256), + ("ec-secp384r1", key_e384), + ("ec-secp521r1", key_e521), + ("rsa-2048", key_r2048), + ("rsa-4096", key_r4096), ], + otdf_client_scs, + ) + + +@pytest.fixture(scope="module") +def attribute_with_different_kids( + otdfctl: OpentdfCommandLineTool, + temporary_namespace: abac.Namespace, + public_key_kas_default_kid_r1: abac.KasKey, + public_key_kas_default_kid_e1: abac.KasKey, + otdf_client_scs: abac.SubjectConditionSet, +) -> abac.Attribute: + """Create an attribute with different KAS public keys (value-level assignment).""" + attr, _ = _create_keyed_attribute( + otdfctl, + temporary_namespace, + "multikeys", + [("r1", public_key_kas_default_kid_r1), ("e1", public_key_kas_default_kid_e1)], + otdf_client_scs, ) + return attr + + +# --------------------------------------------------------------------------- +# Attribute + key assignment fixture (attribute-level) +# --------------------------------------------------------------------------- @pytest.fixture(scope="module") @@ -354,20 +272,9 @@ def attribute_allof_with_two_managed_keys( otdf_client_scs: abac.SubjectConditionSet, temporary_namespace: abac.Namespace, ) -> tuple[abac.Attribute, list[str]]: - """Create an ALL_OF attribute and assign two managed keys (RSA and EC) to it. + """Create an ALL_OF attribute and assign two managed keys at attribute level.""" + tdfs.get_platform_features().skip_if_unsupported("key_management") - - Uses km1 (rsa:2048) and km2 (ec:secp256r1) - - Reuses existing managed keys - - Assigns both keys to the same attribute (attribute-level assignment) - - Maps both attribute values to the client SCS - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip( - "Key management feature is not enabled; skipping key assignment fixture" - ) - - # Create attribute with two values under ALL_OF attr = otdfctl.attribute_create( temporary_namespace, "kmallof", abac.AttributeRule.ALL_OF, ["r1", "e1"] ) @@ -376,19 +283,21 @@ def attribute_allof_with_two_managed_keys( assert r1.value == "r1" assert e1.value == "e1" - # Ensure client has access to both values - sm1 = otdfctl.scs_map(otdf_client_scs, r1) - assert sm1.attribute_value.value == r1.value - sm2 = otdfctl.scs_map(otdf_client_scs, e1) - assert sm2.attribute_value.value == e1.value + for val in [r1, e1]: + sm = otdfctl.scs_map(otdf_client_scs, val) + assert sm.attribute_value.value == val.value - # Assign both keys to the attribute otdfctl.key_assign_attr(managed_key_km1_rsa, attr) otdfctl.key_assign_attr(managed_key_km2_ec, attr) return (attr, [managed_key_km1_rsa.key.key_id, managed_key_km2_ec.key.key_id]) +# --------------------------------------------------------------------------- +# Public key registration fixtures +# --------------------------------------------------------------------------- + + @pytest.fixture(scope="module") def public_key_kas_default_kid_r1( otdfctl: OpentdfCommandLineTool, @@ -413,44 +322,9 @@ def public_key_kas_default_kid_e1( ) -@pytest.fixture(scope="module") -def attribute_with_different_kids( - otdfctl: OpentdfCommandLineTool, - temporary_namespace: abac.Namespace, - public_key_kas_default_kid_r1: abac.KasKey, - public_key_kas_default_kid_e1: abac.KasKey, - otdf_client_scs: abac.SubjectConditionSet, -): - """ - Create an attribute with different KAS public keys. - This is used to test the handling of multiple KAS public keys with different mechanisms. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip( - "Key management feature is not enabled, skipping test for multiple KAS keys" - ) - allof = otdfctl.attribute_create( - temporary_namespace, - "multikeys", - abac.AttributeRule.ALL_OF, - ["r1", "e1"], - ) - assert allof.values - (ar1, ae1) = allof.values - assert ar1.value == "r1" - assert ae1.value == "e1" - - for attr in [ar1, ae1]: - # Then assign it to all clientIds = opentdf-sdk - sm = otdfctl.scs_map(otdf_client_scs, attr) - assert sm.attribute_value.value == attr.value - - # Assign kas key to the attribute values - otdfctl.key_assign_value(public_key_kas_default_kid_e1, ae1) - otdfctl.key_assign_value(public_key_kas_default_kid_r1, ar1) - - return allof +# --------------------------------------------------------------------------- +# Legacy and base key fixtures +# --------------------------------------------------------------------------- @pytest.fixture(scope="module") @@ -460,14 +334,8 @@ def legacy_imported_golden_r1_key( extra_keys: dict[str, ExtraKey], root_key: str, ) -> abac.KasKey: - """ - Import (or reuse) the legacy 'golden-r1' key for decrypting golden TDFs. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip( - "Key management feature is not enabled; skipping legacy key import fixture" - ) + """Import (or reuse) the legacy 'golden-r1' key for decrypting golden TDFs.""" + tdfs.get_platform_features().skip_if_unsupported("key_management") golden_key = extra_keys["golden-r1"] existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km2) @@ -493,13 +361,8 @@ def base_key_e1( kas_entry_km1: abac.KasEntry, root_key: str, ) -> None: - """ - Ensure a managed key with key_id 'e1' exists on the default KAS - and is configured as the base key. - """ - pfs = tdfs.PlatformFeatureSet() - if "key_management" not in pfs.features: - pytest.skip("Key management feature is not enabled; skipping base key fixture") + """Ensure a managed key 'e1' exists on km1 and is configured as the base key.""" + tdfs.get_platform_features().skip_if_unsupported("key_management") existing_keys = otdfctl.kas_registry_keys_list(kas_entry_km1) key_id = "e1" diff --git a/xtest/fixtures/obligations.py b/xtest/fixtures/obligations.py index 370995ee3..953aee052 100644 --- a/xtest/fixtures/obligations.py +++ b/xtest/fixtures/obligations.py @@ -59,7 +59,7 @@ def _obligation_setup_helper( Creates attribute and optional SCS mapping, obligation definition with a single value, and an obligation trigger (optionally scoped to a client id). """ - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() if "obligations" not in pfs.features: pytest.skip("Obligations feature is not enabled") diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 0f2334bbe..56e4c11b1 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -130,11 +130,23 @@ def __init__(self, **kwargs: dict[str, Any]): print(f"PLATFORM_VERSION '{v}' supports [{', '.join(self.features)}]") def skip_if_unsupported(self, *features: feature_type): - for feature in features: - if feature not in self.features: - pytest.skip( - f"platform service {self.version} doesn't yet support [{feature}]" - ) + """Skip the current test if any of the given features are unsupported.""" + missing = [f for f in features if f not in self.features] + if missing: + pytest.skip( + f"platform service {self.version} doesn't yet support {missing}" + ) + + +_cached_pfs: PlatformFeatureSet | None = None + + +def get_platform_features() -> PlatformFeatureSet: + """Return a cached PlatformFeatureSet singleton.""" + global _cached_pfs + if _cached_pfs is None: + _cached_pfs = PlatformFeatureSet() + return _cached_pfs class DataAttribute(BaseModel): @@ -495,7 +507,7 @@ def all_versions_of(sdk: sdk_type) -> list[SDK]: def skip_if_unsupported(sdk: SDK, *features: feature_type): - pfs = PlatformFeatureSet() + pfs = get_platform_features() pfs.skip_if_unsupported(*features) sdk.skip_if_unsupported(*features) diff --git a/xtest/test_abac.py b/xtest/test_abac.py index 263bc205f..f32e7b35a 100644 --- a/xtest/test_abac.py +++ b/xtest/test_abac.py @@ -83,7 +83,7 @@ def test_key_mapping_multiple_mechanisms( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -134,7 +134,7 @@ def test_key_mapping_extended_mechanisms( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() pfs.skip_if_unsupported( "key_management", "autoconfigure", "mechanism-ec-curves-384-521" ) @@ -144,7 +144,7 @@ def test_key_mapping_extended_mechanisms( "mechanism-rsa-4096", "mechanism-ec-curves-384-521", ) - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) skip_dspx1153(encrypt_sdk, decrypt_sdk) @@ -210,7 +210,7 @@ def test_key_mapping_extended_ec_mechanisms( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() pfs.skip_if_unsupported( "key_management", "autoconfigure", "mechanism-ec-curves-384-521" ) @@ -283,7 +283,7 @@ def test_key_mapping_extended_rsa_mechanisms( tdfs.skip_if_unsupported(encrypt_sdk, "key_management") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") encrypt_sdk.skip_if_unsupported("mechanism-rsa-4096") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) skip_dspx1153(encrypt_sdk, decrypt_sdk) @@ -349,7 +349,7 @@ def test_autoconfigure_one_attribute_standard( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -405,7 +405,7 @@ def test_autoconfigure_two_kas_or_standard( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -467,7 +467,7 @@ def test_autoconfigure_double_kas_and( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -528,7 +528,7 @@ def test_autoconfigure_one_attribute_attr_grant( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -575,7 +575,7 @@ def test_autoconfigure_two_kas_or_attr_and_value_grant( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -629,7 +629,7 @@ def test_autoconfigure_two_kas_and_attr_and_value_grant( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -682,7 +682,7 @@ def test_autoconfigure_one_attribute_ns_grant( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure", "ns_grants") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -729,7 +729,7 @@ def test_autoconfigure_two_kas_or_ns_and_value_grant( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure", "ns_grants") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -783,7 +783,7 @@ def test_autoconfigure_two_kas_and_ns_and_value_grant( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure", "ns_grants") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -1040,7 +1040,7 @@ def test_autoconfigure_key_management_two_kas_two_keys( tdfs.skip_if_unsupported(encrypt_sdk, "key_management") tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") skip_dspx2457(encrypt_sdk) - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -1157,7 +1157,7 @@ def test_encrypt_decrypt_all_containers_with_base_key_e1( pytest.skip("Not in focus") tdfs.skip_if_unsupported(encrypt_sdk, "key_management") tdfs.skip_if_unsupported(decrypt_sdk, "key_management") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) diff --git a/xtest/test_audit_logs_integration.py b/xtest/test_audit_logs_integration.py index cbe220712..06ddbccdb 100644 --- a/xtest/test_audit_logs_integration.py +++ b/xtest/test_audit_logs_integration.py @@ -60,7 +60,7 @@ def test_rewrap_success_fields( """Verify all expected fields in successful rewrap audit.""" if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -108,7 +108,7 @@ def test_rewrap_success_with_attributes( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") @@ -154,7 +154,7 @@ def test_multiple_kao_rewrap_audit( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") @@ -308,7 +308,7 @@ def test_decision_on_successful_access( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) tdfs.skip_if_unsupported(encrypt_sdk, "autoconfigure") @@ -370,7 +370,7 @@ def test_audit_logs_on_tampered_file( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -429,7 +429,7 @@ def test_audit_under_sequential_load( """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) diff --git a/xtest/test_policytypes.py b/xtest/test_policytypes.py index d677c8748..a7bf62c82 100644 --- a/xtest/test_policytypes.py +++ b/xtest/test_policytypes.py @@ -20,7 +20,7 @@ def skip_rts_as_needed( if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) if container == "ztdf-ecwrap": diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index 882d7e279..17e89abf2 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -103,7 +103,7 @@ def test_tdf_roundtrip( ): if container == "ztdf" and decrypt_sdk in dspx1153Fails: pytest.skip(f"DSPX-1153 SDK [{decrypt_sdk}] has a bug with payload tampering") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -162,7 +162,7 @@ def test_tdf_spec_target_422( tmp_dir: Path, in_focus: set[tdfs.SDK], ): - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) if "hexaflexible" not in pfs.features: pytest.skip(f"Hexaflexible is not supported in platform {pfs.version}") @@ -306,7 +306,7 @@ def test_tdf_assertions_unkeyed( assertion_file_no_keys: str, in_focus: set[tdfs.SDK], ): - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -339,7 +339,7 @@ def test_tdf_assertions_with_keys( assertion_verification_file_rs_and_hs_keys: str, in_focus: set[tdfs.SDK], ): - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) @@ -380,7 +380,7 @@ def test_tdf_assertions_422_format( ): if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) if not encrypt_sdk.supports("hexaflexible"): pytest.skip( @@ -576,7 +576,7 @@ def test_tdf_with_unbound_policy( ) -> None: if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = do_encrypt_with( @@ -614,7 +614,7 @@ def test_tdf_with_altered_policy_binding( ) -> None: if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = do_encrypt_with(pt_file, encrypt_sdk, "ztdf", tmp_dir) @@ -650,7 +650,7 @@ def test_tdf_with_altered_root_sig( ): if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = do_encrypt_with( @@ -679,7 +679,7 @@ def test_tdf_with_altered_seg_sig_wrong( ): if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = do_encrypt_with( @@ -713,7 +713,7 @@ def test_tdf_with_altered_enc_seg_size( ): if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = do_encrypt_with( @@ -748,7 +748,7 @@ def test_tdf_with_altered_assertion_statement( ): if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) if not encrypt_sdk.supports("assertions"): @@ -787,7 +787,7 @@ def test_tdf_with_altered_assertion_with_keys( ): if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) if not encrypt_sdk.supports("assertions"): @@ -835,7 +835,7 @@ def test_tdf_altered_payload_end( pytest.skip("Not in focus") if decrypt_sdk in dspx1153Fails: pytest.skip(f"DSPX-1153 SDK [{decrypt_sdk}] has a bug with payload tampering") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = do_encrypt_with( @@ -868,7 +868,7 @@ def test_tdf_with_malicious_kao( ) -> None: if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") - pfs = tdfs.PlatformFeatureSet() + pfs = tdfs.get_platform_features() tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) if not decrypt_sdk.supports("kasallowlist"):