From bf3adcb8527170f500be6f4447b7017f5152d5e8 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Wed, 14 Oct 2026 15:10:35 -0400 Subject: [PATCH 1/7] feat(xtest): cover spec-compliant policy bindings KAS must accept a policy binding in the spec's Base64(HMAC) form as well as the legacy Base64(hex(HMAC)) every SDK writes. No released KAS did (opentdf/platform#2095 compared an untrimmed buffer); fixed in opentdf/platform#4223. test_tdf_with_spec_policy_binding re-encodes the binding to the spec form, which needs no key, and expects decrypt to succeed. It is gated on the new platform-only feature spec_policy_binding, force-only until a release carries the fix. test_tdf_with_altered_spec_policy_binding expects a well-formed spec binding with a wrong HMAC to be rejected, so the new path cannot fail open. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Paul Flynn --- xtest/tdfs.py | 10 ++++- xtest/test_tdfs.py | 93 ++++++++++++++++++++++++++++++++++++++++ xtest/test_tdfs_units.py | 77 +++++++++++++++++++++++++++++++++ 3 files changed, 179 insertions(+), 1 deletion(-) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 60bf26e9..2dc0ae27 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -194,6 +194,12 @@ def is_sdk_type(val: str) -> TypeIs[sdk_type]: "multikao", "ns_grants", "obligations", + # Platform: KAS verifies a policy binding in the spec's Base64(HMAC) form as + # well as the legacy Base64(hex(HMAC)) every SDK writes. Released KAS never + # accepted the spec form (opentdf/platform#2095 compared an untrimmed + # buffer); fixed in opentdf/platform#4223. Force-only via + # XT_FORCE_PLATFORM_SUPPORTS until the first release carrying it is known. + "spec_policy_binding", # Writer-side: switch to the ZIP64 sentinel plus extra field at 2 GiB # rather than at 4 GiB. # @@ -227,7 +233,9 @@ def is_sdk_type(val: str) -> TypeIs[sdk_type]: #: ``autoconfigure`` and ``ecwrap`` need the SDK to emit them *and* the platform #: to honour them, so forcing either side alone is a reasonable thing to ask for. #: Only add a name here when no ``sdk/*/cli.sh`` has a ``supports`` case for it. -PLATFORM_ONLY_FEATURES: frozenset[feature_type] = frozenset({"kas_uri_from_kao"}) +PLATFORM_ONLY_FEATURES: frozenset[feature_type] = frozenset( + {"kas_uri_from_kao", "spec_policy_binding"} +) SDK_FORCEABLE_FEATURES = ALL_FEATURES - PLATFORM_ONLY_FEATURES diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index eb8b7fb2..fb27d6fc 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -487,6 +487,42 @@ def change_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: return manifest +def _set_policy_binding_hash(kao: tdfs.KeyAccessObject, hash: str) -> None: + if isinstance(kao.policyBinding, tdfs.PolicyBinding): + kao.policyBinding.hash = hash + else: + kao.policyBinding = hash + + +def spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: + """Re-encode every policy binding as the spec's Base64(HMAC). + + SDKs write the legacy Base64(hex(HMAC)). Both spell the same HMAC, so this + needs no key, and a KAS that accepts the spec form must still decrypt. + """ + for kao in manifest.encryptionInformation.keyAccess: + pb = kao.policyBinding + mac = base64.b64decode(pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb) + if len(mac) == 64: + mac = bytes.fromhex(mac.decode()) + assert len(mac) == 32, f"unexpected policy binding length {len(mac)}" + _set_policy_binding_hash(kao, base64.b64encode(mac).decode()) + return manifest + + +def change_spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: + """A spec-form binding with its HMAC altered: well-formed, but wrong.""" + manifest = spec_policy_binding(manifest) + kao = manifest.encryptionInformation.keyAccess[0] + pb = kao.policyBinding + mac = bytearray( + base64.b64decode(pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb) + ) + mac[-1] ^= 0x01 + _set_policy_binding_hash(kao, base64.b64encode(bytes(mac)).decode()) + return manifest + + def change_root_signature(manifest: tdfs.Manifest) -> tdfs.Manifest: root_sig = manifest.encryptionInformation.integrityInformation.rootSignature.sig altered_sig = base64.b64encode(change_last_three(base64.b64decode(root_sig))) @@ -670,6 +706,63 @@ def test_tdf_with_altered_policy_binding( assert_kas_request_error(exc, decrypt_sdk) +def test_tdf_with_spec_policy_binding( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + pt_file: Path, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +) -> None: + """KAS accepts the spec's Base64(HMAC) policy binding, not only legacy hex.""" + if not in_focus & {encrypt_sdk, decrypt_sdk}: + pytest.skip("Not in focus") + pfs = tdfs.get_platform_features() + pfs.skip_if_unsupported("spec_policy_binding") + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) + ct_file = encrypted_tdf( + encrypt_sdk, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + attr_values=attribute_default_rsa.value_fqns, + ) + b_file = tdfs.update_manifest("spec_policy_binding", ct_file, spec_policy_binding) + rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) + + decrypt_sdk.decrypt(b_file, rt_file, "ztdf") + assert filecmp.cmp(pt_file, rt_file) + + +def test_tdf_with_altered_spec_policy_binding( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +) -> None: + """Accepting the spec form must not fail open on a wrong HMAC.""" + if not in_focus & {encrypt_sdk, decrypt_sdk}: + pytest.skip("Not in focus") + pfs = tdfs.get_platform_features() + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) + ct_file = encrypted_tdf( + encrypt_sdk, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + attr_values=attribute_default_rsa.value_fqns, + ) + b_file = tdfs.update_manifest( + "altered_spec_policy_binding", ct_file, change_spec_policy_binding + ) + rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) + + try: + decrypt_sdk.decrypt(b_file, rt_file, "ztdf", expect_error=True) + assert False, "decrypt succeeded unexpectedly" + except subprocess.CalledProcessError as exc: + assert_kas_request_error(exc, decrypt_sdk) + + ## INTEGRITY TAMPER TESTS diff --git a/xtest/test_tdfs_units.py b/xtest/test_tdfs_units.py index 2a9c17b4..eaf6c57e 100644 --- a/xtest/test_tdfs_units.py +++ b/xtest/test_tdfs_units.py @@ -375,3 +375,80 @@ def test_falls_back_to_the_manifest_default(self, tmp_path: Path): def test_uses_the_per_segment_override(self, tmp_path: Path): ct_file = _manifest_zip(tmp_path, "full.tdf", elides=False) assert tdfs.encrypted_segment_sizes(tdfs.manifest(ct_file)) == [1028, 1028] + + +# --- spec_policy_binding / change_spec_policy_binding ------------------------ +# +# The xtest for spec-form policy bindings rewrites a real manifest. If the +# rewrite produced the wrong bytes, the positive test would fail for the wrong +# reason and the negative one would pass for it. + + +def _binding_manifest(binding: object) -> tdfs.Manifest: + return tdfs.Manifest.model_validate( + { + "encryptionInformation": { + "type": "split", + "policy": "", + "keyAccess": [ + { + "type": "wrapped", + "url": "http://kas", + "protocol": "kas", + "wrappedKey": "", + "policyBinding": binding, + } + ], + "method": {"algorithm": "AES-256-GCM"}, + "integrityInformation": { + "rootSignature": {"alg": "HS256", "sig": ""}, + "segmentHashAlg": "GMAC", + "segments": [], + }, + }, + "payload": { + "type": "reference", + "url": "0.payload", + "protocol": "zip", + "isEncrypted": True, + }, + } + ) + + +_MAC = bytes(range(32)) +_LEGACY = base64.b64encode(_MAC.hex().encode()).decode() +_SPEC = base64.b64encode(_MAC).decode() + + +def _binding_hash(m: tdfs.Manifest) -> str: + pb = m.encryptionInformation.keyAccess[0].policyBinding + return pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb + + +class TestSpecPolicyBinding: + @pytest.mark.parametrize( + "binding", + [{"alg": "HS256", "hash": _LEGACY}, _LEGACY], + ids=["object", "string"], + ) + def test_legacy_hex_becomes_spec_form_of_the_same_hmac(self, binding: object): + from test_tdfs import spec_policy_binding + + assert _binding_hash(spec_policy_binding(_binding_manifest(binding))) == _SPEC + + def test_spec_form_is_left_as_is(self): + from test_tdfs import spec_policy_binding + + m = _binding_manifest({"alg": "HS256", "hash": _SPEC}) + assert _binding_hash(spec_policy_binding(m)) == _SPEC + + def test_altered_spec_form_is_well_formed_but_wrong(self): + from test_tdfs import change_spec_policy_binding + + m = change_spec_policy_binding( + _binding_manifest({"alg": "HS256", "hash": _LEGACY}) + ) + altered = base64.b64decode(_binding_hash(m)) + assert len(altered) == len(_MAC) + assert altered != _MAC From 747848c365d35e38f129e743f16a9b3ce0813022 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Wed, 14 Oct 2026 15:19:55 -0400 Subject: [PATCH 2/7] fix(xtest): run the spec policy binding test against every platform The point is to show that a real KAS -- the current release and main -- rejects an on-spec policy binding. A force-only gate skipped exactly those cells, so drop it. The test fails until a platform carrying opentdf/platform#4223 is under test. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Paul Flynn --- xtest/tdfs.py | 10 +--------- xtest/test_tdfs.py | 8 ++++++-- 2 files changed, 7 insertions(+), 11 deletions(-) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 2dc0ae27..60bf26e9 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -194,12 +194,6 @@ def is_sdk_type(val: str) -> TypeIs[sdk_type]: "multikao", "ns_grants", "obligations", - # Platform: KAS verifies a policy binding in the spec's Base64(HMAC) form as - # well as the legacy Base64(hex(HMAC)) every SDK writes. Released KAS never - # accepted the spec form (opentdf/platform#2095 compared an untrimmed - # buffer); fixed in opentdf/platform#4223. Force-only via - # XT_FORCE_PLATFORM_SUPPORTS until the first release carrying it is known. - "spec_policy_binding", # Writer-side: switch to the ZIP64 sentinel plus extra field at 2 GiB # rather than at 4 GiB. # @@ -233,9 +227,7 @@ def is_sdk_type(val: str) -> TypeIs[sdk_type]: #: ``autoconfigure`` and ``ecwrap`` need the SDK to emit them *and* the platform #: to honour them, so forcing either side alone is a reasonable thing to ask for. #: Only add a name here when no ``sdk/*/cli.sh`` has a ``supports`` case for it. -PLATFORM_ONLY_FEATURES: frozenset[feature_type] = frozenset( - {"kas_uri_from_kao", "spec_policy_binding"} -) +PLATFORM_ONLY_FEATURES: frozenset[feature_type] = frozenset({"kas_uri_from_kao"}) SDK_FORCEABLE_FEATURES = ALL_FEATURES - PLATFORM_ONLY_FEATURES diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index fb27d6fc..966f0aea 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -714,11 +714,15 @@ def test_tdf_with_spec_policy_binding( attribute_default_rsa: Attribute, encrypted_tdf: EncryptFactory, ) -> None: - """KAS accepts the spec's Base64(HMAC) policy binding, not only legacy hex.""" + """KAS accepts the spec's Base64(HMAC) policy binding, not only legacy hex. + + Deliberately ungated: no released KAS accepts the spec form + (opentdf/platform#2095 compared an untrimmed buffer), so this fails until a + platform carrying opentdf/platform#4223 is under test. + """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") pfs = tdfs.get_platform_features() - pfs.skip_if_unsupported("spec_policy_binding") tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) tdfs.skip_hexless_skew(encrypt_sdk, decrypt_sdk) ct_file = encrypted_tdf( From c7e8331108c8f63d0f143bd21fd3f1a6eadd5143 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Wed, 14 Oct 2026 15:23:11 -0400 Subject: [PATCH 3/7] fix(xtest): alter the policy binding HMAC, not its encoding change_policy_binding stored str(bytes) -- the literal "b'...'" -- so test_tdf_with_altered_policy_binding sent invalid base64 and only showed that KAS rejects a malformed binding, never that it checks the HMAC. Flip a bit of the HMAC instead and re-encode in the binding's own form, so a legacy hex binding stays well-formed hex. The spec-form tamper helper now reuses it. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Paul Flynn --- xtest/test_tdfs.py | 45 +++++++++++++++++++--------------------- xtest/test_tdfs_units.py | 16 ++++++++++++++ 2 files changed, 37 insertions(+), 24 deletions(-) diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index 966f0aea..daa2a47c 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -472,19 +472,9 @@ def change_policy(manifest: tdfs.Manifest) -> tdfs.Manifest: return manifest -def change_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: - pb = manifest.encryptionInformation.keyAccess[0].policyBinding - ## if the pb is str then json decode to tdfs.PolicyBinding - if isinstance(pb, tdfs.PolicyBinding): - hash = pb.hash - altered_hash = base64.b64encode(change_last_three(base64.b64decode(hash))) - pb.hash = str(altered_hash) - manifest.encryptionInformation.keyAccess[0].policyBinding = pb - else: - altered_hash = base64.b64encode(change_last_three(base64.b64decode(pb))) - manifest.encryptionInformation.keyAccess[0].policyBinding = str(altered_hash) - - return manifest +def _policy_binding_hash(kao: tdfs.KeyAccessObject) -> str: + pb = kao.policyBinding + return pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb def _set_policy_binding_hash(kao: tdfs.KeyAccessObject, hash: str) -> None: @@ -494,6 +484,22 @@ def _set_policy_binding_hash(kao: tdfs.KeyAccessObject, hash: str) -> None: kao.policyBinding = hash +def change_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: + """Alter the first policy binding's HMAC, keeping its encoding well-formed. + + A malformed binding is rejected before the HMAC is compared, so only a + well-formed one shows the KAS checks the HMAC itself. + """ + kao = manifest.encryptionInformation.keyAccess[0] + decoded = base64.b64decode(_policy_binding_hash(kao)) + legacy = len(decoded) == 64 + mac = bytearray(bytes.fromhex(decoded.decode()) if legacy else decoded) + mac[-1] ^= 0x01 + altered = mac.hex().encode() if legacy else bytes(mac) + _set_policy_binding_hash(kao, base64.b64encode(altered).decode()) + return manifest + + def spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: """Re-encode every policy binding as the spec's Base64(HMAC). @@ -501,8 +507,7 @@ def spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: needs no key, and a KAS that accepts the spec form must still decrypt. """ for kao in manifest.encryptionInformation.keyAccess: - pb = kao.policyBinding - mac = base64.b64decode(pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb) + mac = base64.b64decode(_policy_binding_hash(kao)) if len(mac) == 64: mac = bytes.fromhex(mac.decode()) assert len(mac) == 32, f"unexpected policy binding length {len(mac)}" @@ -512,15 +517,7 @@ def spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: def change_spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: """A spec-form binding with its HMAC altered: well-formed, but wrong.""" - manifest = spec_policy_binding(manifest) - kao = manifest.encryptionInformation.keyAccess[0] - pb = kao.policyBinding - mac = bytearray( - base64.b64decode(pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb) - ) - mac[-1] ^= 0x01 - _set_policy_binding_hash(kao, base64.b64encode(bytes(mac)).decode()) - return manifest + return change_policy_binding(spec_policy_binding(manifest)) def change_root_signature(manifest: tdfs.Manifest) -> tdfs.Manifest: diff --git a/xtest/test_tdfs_units.py b/xtest/test_tdfs_units.py index eaf6c57e..4c481272 100644 --- a/xtest/test_tdfs_units.py +++ b/xtest/test_tdfs_units.py @@ -452,3 +452,19 @@ def test_altered_spec_form_is_well_formed_but_wrong(self): altered = base64.b64decode(_binding_hash(m)) assert len(altered) == len(_MAC) assert altered != _MAC + + +class TestChangePolicyBinding: + @pytest.mark.parametrize( + "binding", + [{"alg": "HS256", "hash": _LEGACY}, _LEGACY], + ids=["object", "string"], + ) + def test_legacy_hex_stays_well_formed_hex_with_a_wrong_hmac(self, binding: object): + """A malformed binding is rejected before the HMAC is ever compared.""" + from test_tdfs import change_policy_binding + + m = change_policy_binding(_binding_manifest(binding)) + altered = bytes.fromhex(base64.b64decode(_binding_hash(m)).decode()) + assert len(altered) == len(_MAC) + assert altered != _MAC From 46cd8fc92ff3bb81e6d87367eada2a59a8bd0fa1 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Wed, 14 Oct 2026 15:37:55 -0400 Subject: [PATCH 4/7] fix(xtest): compare spec policy binding plaintext by content Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Paul Flynn --- xtest/test_tdfs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index daa2a47c..f3721d85 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -731,7 +731,7 @@ def test_tdf_with_spec_policy_binding( rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) decrypt_sdk.decrypt(b_file, rt_file, "ztdf") - assert filecmp.cmp(pt_file, rt_file) + assert filecmp.cmp(pt_file, rt_file, shallow=False) def test_tdf_with_altered_spec_policy_binding( From 6942a5fd93cfb3881826b316f2bbd35c6f5a2c47 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Thu, 15 Oct 2026 13:30:59 -0400 Subject: [PATCH 5/7] fix(xtest): expect spec policy binding rejection on affected platforms Platform releases from opentdf/platform#2095 (v0.5.3) until #4223 reject a Base64(HMAC) policy binding. Against those, assert the decrypt fails with a KAS error instead of failing the cell, keyed on PLATFORM_TAG since a source build reports the last release's PLATFORM_VERSION. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Paul Flynn --- xtest/tdfs.py | 31 +++++++++++++++++++++++++++++ xtest/test_tdfs.py | 14 ++++++++++--- xtest/test_tdfs_units.py | 43 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 85 insertions(+), 3 deletions(-) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 60bf26e9..5acb4034 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -1130,6 +1130,37 @@ def zip64_reader_is_broken(decrypt_sdk: SDK) -> bool: return decrypt_sdk.sdk == "java" and sv is not None and sv < JAVA_ZIP64_READER_FIX +#: First platform release whose KAS rejects the spec's Base64(HMAC) policy +#: binding: opentdf/platform#2095 compared the decoded binding against an +#: untrimmed buffer, so only the legacy Base64(hex(HMAC)) form verifies. +#: Releases before it accepted both forms. +PLATFORM_SPEC_POLICY_BINDING_BUG = (0, 5, 3) + +#: First platform release carrying the fix, opentdf/platform#4223. +#: +#: Like :data:`JAVA_ZIP64_READER_FIX`, both directions of getting it wrong turn +#: the cell red. Not yet released when written; 0.29.1 is the next version +#: release-please would cut from main, and stays correct if it is 0.30.0 +#: instead. Confirm it when the release actually ships. +PLATFORM_SPEC_POLICY_BINDING_FIX = (0, 29, 1) + + +def spec_policy_binding_is_rejected() -> bool: + """True for platform releases known to reject a spec-form policy binding. + + Keyed on ``PLATFORM_TAG``, not ``PLATFORM_VERSION``: a source build reports + the last release's version -- platform ``main`` carrying #4223 still says + 0.29.0 -- so only the tag tells that release apart from the fix. An unset + tag or a branch name is never assumed broken, so a local run against an + old release fails rather than hides; export ``PLATFORM_TAG`` to match it. + """ + sv = _parse_semver(os.getenv("PLATFORM_TAG", "")) + return ( + sv is not None + and PLATFORM_SPEC_POLICY_BINDING_BUG <= sv < PLATFORM_SPEC_POLICY_BINDING_FIX + ) + + def _parse_semver(version: str) -> tuple[int, int, int] | None: """Parse a version string (with optional 'v' prefix) into (major, minor, patch).""" m = _version_re.match(version.lstrip("v")) diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index f3721d85..ff19ecca 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -713,9 +713,11 @@ def test_tdf_with_spec_policy_binding( ) -> None: """KAS accepts the spec's Base64(HMAC) policy binding, not only legacy hex. - Deliberately ungated: no released KAS accepts the spec form - (opentdf/platform#2095 compared an untrimmed buffer), so this fails until a - platform carrying opentdf/platform#4223 is under test. + Releases from opentdf/platform#2095 until #4223 reject it, so against those + the decrypt is asserted to fail with a KAS error rather than skipped or + xfailed: a node-level xfail would absorb unrelated failures, and a release + that has quietly gained the fix turns the cell red. See + :func:`tdfs.spec_policy_binding_is_rejected`. """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") @@ -730,6 +732,12 @@ def test_tdf_with_spec_policy_binding( b_file = tdfs.update_manifest("spec_policy_binding", ct_file, spec_policy_binding) rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) + if tdfs.spec_policy_binding_is_rejected(): + with pytest.raises(subprocess.CalledProcessError) as exc: + decrypt_sdk.decrypt(b_file, rt_file, "ztdf", expect_error=True) + assert_kas_request_error(exc.value, decrypt_sdk) + return + decrypt_sdk.decrypt(b_file, rt_file, "ztdf") assert filecmp.cmp(pt_file, rt_file, shallow=False) diff --git a/xtest/test_tdfs_units.py b/xtest/test_tdfs_units.py index 4c481272..fbeec689 100644 --- a/xtest/test_tdfs_units.py +++ b/xtest/test_tdfs_units.py @@ -182,6 +182,49 @@ def test_a_branch_build_is_not(self): assert not tdfs.zip64_reader_is_broken(_stub_sdk("java", None)) +# --- tdfs.spec_policy_binding_is_rejected ------------------------------------- + + +class TestSpecPolicyBindingIsRejected: + """Which platforms ``test_tdf_with_spec_policy_binding`` requires a failure from. + + Each "False" here is a platform held to a successful decrypt, so a + predicate that were too generous would hide a KAS that regressed. + """ + + def test_a_pre_fix_release_is_rejected(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("PLATFORM_TAG", "v0.29.0") + assert tdfs.spec_policy_binding_is_rejected() + + def test_the_bug_release_itself_is_rejected(self, monkeypatch: pytest.MonkeyPatch): + """Boundary: the constant names the first release *with* the bug.""" + major, minor, patch = tdfs.PLATFORM_SPEC_POLICY_BINDING_BUG + monkeypatch.setenv("PLATFORM_TAG", f"v{major}.{minor}.{patch}") + assert tdfs.spec_policy_binding_is_rejected() + + def test_a_release_before_the_bug_is_not(self, monkeypatch: pytest.MonkeyPatch): + """Before opentdf/platform#2095, KAS accepted the spec form.""" + monkeypatch.setenv("PLATFORM_TAG", "v0.5.2") + assert not tdfs.spec_policy_binding_is_rejected() + + def test_the_fix_release_itself_is_not(self, monkeypatch: pytest.MonkeyPatch): + """Boundary: the constant names the first release *with* the fix.""" + major, minor, patch = tdfs.PLATFORM_SPEC_POLICY_BINDING_FIX + monkeypatch.setenv("PLATFORM_TAG", f"v{major}.{minor}.{patch}") + assert not tdfs.spec_policy_binding_is_rejected() + + def test_a_branch_build_is_not(self, monkeypatch: pytest.MonkeyPatch): + """main reports the last release's PLATFORM_VERSION, so that is not consulted.""" + monkeypatch.setenv("PLATFORM_TAG", "main") + monkeypatch.setenv("PLATFORM_VERSION", "0.29.0") + assert not tdfs.spec_policy_binding_is_rejected() + + def test_no_tag_is_not(self, monkeypatch: pytest.MonkeyPatch): + monkeypatch.delenv("PLATFORM_TAG", raising=False) + monkeypatch.setenv("PLATFORM_VERSION", "0.29.0") + assert not tdfs.spec_policy_binding_is_rejected() + + # --- tdfs.elides_segment_sizes / tdfs.skip_chunky_skew ------------------------ From 75ced39d8a007e0f26e3243d37df7b8dcaa84b6c Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Thu, 15 Oct 2026 14:15:27 -0400 Subject: [PATCH 6/7] docs(xtest): tighten spec policy binding comments Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Paul Flynn --- xtest/tdfs.py | 21 +++++---------------- 1 file changed, 5 insertions(+), 16 deletions(-) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 5acb4034..3a7b0410 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -1130,29 +1130,18 @@ def zip64_reader_is_broken(decrypt_sdk: SDK) -> bool: return decrypt_sdk.sdk == "java" and sv is not None and sv < JAVA_ZIP64_READER_FIX -#: First platform release whose KAS rejects the spec's Base64(HMAC) policy -#: binding: opentdf/platform#2095 compared the decoded binding against an -#: untrimmed buffer, so only the legacy Base64(hex(HMAC)) form verifies. -#: Releases before it accepted both forms. +#: Platform releases in [BUG, FIX) reject the spec's Base64(HMAC) policy binding +#: and accept only legacy Base64(hex(HMAC)): opentdf/platform#2095 introduced +#: the bug, opentdf/platform#4223 fixed it. PLATFORM_SPEC_POLICY_BINDING_BUG = (0, 5, 3) - -#: First platform release carrying the fix, opentdf/platform#4223. -#: -#: Like :data:`JAVA_ZIP64_READER_FIX`, both directions of getting it wrong turn -#: the cell red. Not yet released when written; 0.29.1 is the next version -#: release-please would cut from main, and stays correct if it is 0.30.0 -#: instead. Confirm it when the release actually ships. PLATFORM_SPEC_POLICY_BINDING_FIX = (0, 29, 1) def spec_policy_binding_is_rejected() -> bool: """True for platform releases known to reject a spec-form policy binding. - Keyed on ``PLATFORM_TAG``, not ``PLATFORM_VERSION``: a source build reports - the last release's version -- platform ``main`` carrying #4223 still says - 0.29.0 -- so only the tag tells that release apart from the fix. An unset - tag or a branch name is never assumed broken, so a local run against an - old release fails rather than hides; export ``PLATFORM_TAG`` to match it. + Uses ``PLATFORM_TAG`` because source builds report a stale + ``PLATFORM_VERSION``. An unset tag or branch name is never assumed broken. """ sv = _parse_semver(os.getenv("PLATFORM_TAG", "")) return ( From f81ad0062b07ea2b131f2fac74d2a9eb4422c356 Mon Sep 17 00:00:00 2001 From: Paul Flynn Date: Mon, 19 Oct 2026 13:01:55 -0400 Subject: [PATCH 7/7] refactor(xtest): name policy binding helpers for the hexless form Signed-off-by: Paul Flynn Co-Authored-By: Claude Opus 5.5 (1M context) --- xtest/tdfs.py | 14 +++++----- xtest/test_tdfs.py | 30 ++++++++++++---------- xtest/test_tdfs_units.py | 55 +++++++++++++++++++++------------------- 3 files changed, 53 insertions(+), 46 deletions(-) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 3a7b0410..3a669885 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -1130,15 +1130,15 @@ def zip64_reader_is_broken(decrypt_sdk: SDK) -> bool: return decrypt_sdk.sdk == "java" and sv is not None and sv < JAVA_ZIP64_READER_FIX -#: Platform releases in [BUG, FIX) reject the spec's Base64(HMAC) policy binding +#: Platform releases in [BUG, FIX) reject the hexless Base64(HMAC) policy binding #: and accept only legacy Base64(hex(HMAC)): opentdf/platform#2095 introduced #: the bug, opentdf/platform#4223 fixed it. -PLATFORM_SPEC_POLICY_BINDING_BUG = (0, 5, 3) -PLATFORM_SPEC_POLICY_BINDING_FIX = (0, 29, 1) +PLATFORM_HEXLESS_POLICY_BINDING_BUG = (0, 5, 3) +PLATFORM_HEXLESS_POLICY_BINDING_FIX = (0, 29, 1) -def spec_policy_binding_is_rejected() -> bool: - """True for platform releases known to reject a spec-form policy binding. +def hexless_policy_binding_is_rejected() -> bool: + """True for platform releases known to reject a hexless-form policy binding. Uses ``PLATFORM_TAG`` because source builds report a stale ``PLATFORM_VERSION``. An unset tag or branch name is never assumed broken. @@ -1146,7 +1146,9 @@ def spec_policy_binding_is_rejected() -> bool: sv = _parse_semver(os.getenv("PLATFORM_TAG", "")) return ( sv is not None - and PLATFORM_SPEC_POLICY_BINDING_BUG <= sv < PLATFORM_SPEC_POLICY_BINDING_FIX + and PLATFORM_HEXLESS_POLICY_BINDING_BUG + <= sv + < PLATFORM_HEXLESS_POLICY_BINDING_FIX ) diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index ff19ecca..e6aef517 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -500,11 +500,11 @@ def change_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: return manifest -def spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: - """Re-encode every policy binding as the spec's Base64(HMAC). +def make_hexless_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: + """Re-encode every policy binding as the hexless Base64(HMAC). SDKs write the legacy Base64(hex(HMAC)). Both spell the same HMAC, so this - needs no key, and a KAS that accepts the spec form must still decrypt. + needs no key, and a KAS that accepts the hexless form must still decrypt. """ for kao in manifest.encryptionInformation.keyAccess: mac = base64.b64decode(_policy_binding_hash(kao)) @@ -515,9 +515,9 @@ def spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: return manifest -def change_spec_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: - """A spec-form binding with its HMAC altered: well-formed, but wrong.""" - return change_policy_binding(spec_policy_binding(manifest)) +def change_hexless_policy_binding(manifest: tdfs.Manifest) -> tdfs.Manifest: + """A hexless-form binding with its HMAC altered: well-formed, but wrong.""" + return change_policy_binding(make_hexless_policy_binding(manifest)) def change_root_signature(manifest: tdfs.Manifest) -> tdfs.Manifest: @@ -703,7 +703,7 @@ def test_tdf_with_altered_policy_binding( assert_kas_request_error(exc, decrypt_sdk) -def test_tdf_with_spec_policy_binding( +def test_tdf_with_hexless_policy_binding( encrypt_sdk: tdfs.SDK, decrypt_sdk: tdfs.SDK, pt_file: Path, @@ -711,13 +711,13 @@ def test_tdf_with_spec_policy_binding( attribute_default_rsa: Attribute, encrypted_tdf: EncryptFactory, ) -> None: - """KAS accepts the spec's Base64(HMAC) policy binding, not only legacy hex. + """KAS accepts the hexless Base64(HMAC) policy binding, not only legacy hex. Releases from opentdf/platform#2095 until #4223 reject it, so against those the decrypt is asserted to fail with a KAS error rather than skipped or xfailed: a node-level xfail would absorb unrelated failures, and a release that has quietly gained the fix turns the cell red. See - :func:`tdfs.spec_policy_binding_is_rejected`. + :func:`tdfs.hexless_policy_binding_is_rejected`. """ if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") @@ -729,10 +729,12 @@ def test_tdf_with_spec_policy_binding( target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), attr_values=attribute_default_rsa.value_fqns, ) - b_file = tdfs.update_manifest("spec_policy_binding", ct_file, spec_policy_binding) + b_file = tdfs.update_manifest( + "hexless_policy_binding", ct_file, make_hexless_policy_binding + ) rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) - if tdfs.spec_policy_binding_is_rejected(): + if tdfs.hexless_policy_binding_is_rejected(): with pytest.raises(subprocess.CalledProcessError) as exc: decrypt_sdk.decrypt(b_file, rt_file, "ztdf", expect_error=True) assert_kas_request_error(exc.value, decrypt_sdk) @@ -742,14 +744,14 @@ def test_tdf_with_spec_policy_binding( assert filecmp.cmp(pt_file, rt_file, shallow=False) -def test_tdf_with_altered_spec_policy_binding( +def test_tdf_with_altered_hexless_policy_binding( encrypt_sdk: tdfs.SDK, decrypt_sdk: tdfs.SDK, in_focus: set[tdfs.SDK], attribute_default_rsa: Attribute, encrypted_tdf: EncryptFactory, ) -> None: - """Accepting the spec form must not fail open on a wrong HMAC.""" + """Accepting the hexless form must not fail open on a wrong HMAC.""" if not in_focus & {encrypt_sdk, decrypt_sdk}: pytest.skip("Not in focus") pfs = tdfs.get_platform_features() @@ -761,7 +763,7 @@ def test_tdf_with_altered_spec_policy_binding( attr_values=attribute_default_rsa.value_fqns, ) b_file = tdfs.update_manifest( - "altered_spec_policy_binding", ct_file, change_spec_policy_binding + "altered_hexless_policy_binding", ct_file, change_hexless_policy_binding ) rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) diff --git a/xtest/test_tdfs_units.py b/xtest/test_tdfs_units.py index fbeec689..67a1e574 100644 --- a/xtest/test_tdfs_units.py +++ b/xtest/test_tdfs_units.py @@ -182,11 +182,11 @@ def test_a_branch_build_is_not(self): assert not tdfs.zip64_reader_is_broken(_stub_sdk("java", None)) -# --- tdfs.spec_policy_binding_is_rejected ------------------------------------- +# --- tdfs.hexless_policy_binding_is_rejected ------------------------------------- -class TestSpecPolicyBindingIsRejected: - """Which platforms ``test_tdf_with_spec_policy_binding`` requires a failure from. +class TestHexlessPolicyBindingIsRejected: + """Which platforms ``test_tdf_with_hexless_policy_binding`` requires a failure from. Each "False" here is a platform held to a successful decrypt, so a predicate that were too generous would hide a KAS that regressed. @@ -194,35 +194,35 @@ class TestSpecPolicyBindingIsRejected: def test_a_pre_fix_release_is_rejected(self, monkeypatch: pytest.MonkeyPatch): monkeypatch.setenv("PLATFORM_TAG", "v0.29.0") - assert tdfs.spec_policy_binding_is_rejected() + assert tdfs.hexless_policy_binding_is_rejected() def test_the_bug_release_itself_is_rejected(self, monkeypatch: pytest.MonkeyPatch): """Boundary: the constant names the first release *with* the bug.""" - major, minor, patch = tdfs.PLATFORM_SPEC_POLICY_BINDING_BUG + major, minor, patch = tdfs.PLATFORM_HEXLESS_POLICY_BINDING_BUG monkeypatch.setenv("PLATFORM_TAG", f"v{major}.{minor}.{patch}") - assert tdfs.spec_policy_binding_is_rejected() + assert tdfs.hexless_policy_binding_is_rejected() def test_a_release_before_the_bug_is_not(self, monkeypatch: pytest.MonkeyPatch): - """Before opentdf/platform#2095, KAS accepted the spec form.""" + """Before opentdf/platform#2095, KAS accepted the hexless form.""" monkeypatch.setenv("PLATFORM_TAG", "v0.5.2") - assert not tdfs.spec_policy_binding_is_rejected() + assert not tdfs.hexless_policy_binding_is_rejected() def test_the_fix_release_itself_is_not(self, monkeypatch: pytest.MonkeyPatch): """Boundary: the constant names the first release *with* the fix.""" - major, minor, patch = tdfs.PLATFORM_SPEC_POLICY_BINDING_FIX + major, minor, patch = tdfs.PLATFORM_HEXLESS_POLICY_BINDING_FIX monkeypatch.setenv("PLATFORM_TAG", f"v{major}.{minor}.{patch}") - assert not tdfs.spec_policy_binding_is_rejected() + assert not tdfs.hexless_policy_binding_is_rejected() def test_a_branch_build_is_not(self, monkeypatch: pytest.MonkeyPatch): """main reports the last release's PLATFORM_VERSION, so that is not consulted.""" monkeypatch.setenv("PLATFORM_TAG", "main") monkeypatch.setenv("PLATFORM_VERSION", "0.29.0") - assert not tdfs.spec_policy_binding_is_rejected() + assert not tdfs.hexless_policy_binding_is_rejected() def test_no_tag_is_not(self, monkeypatch: pytest.MonkeyPatch): monkeypatch.delenv("PLATFORM_TAG", raising=False) monkeypatch.setenv("PLATFORM_VERSION", "0.29.0") - assert not tdfs.spec_policy_binding_is_rejected() + assert not tdfs.hexless_policy_binding_is_rejected() # --- tdfs.elides_segment_sizes / tdfs.skip_chunky_skew ------------------------ @@ -420,9 +420,9 @@ def test_uses_the_per_segment_override(self, tmp_path: Path): assert tdfs.encrypted_segment_sizes(tdfs.manifest(ct_file)) == [1028, 1028] -# --- spec_policy_binding / change_spec_policy_binding ------------------------ +# --- make_hexless_policy_binding / change_hexless_policy_binding ------------------------ # -# The xtest for spec-form policy bindings rewrites a real manifest. If the +# The xtest for hexless-form policy bindings rewrites a real manifest. If the # rewrite produced the wrong bytes, the positive test would fail for the wrong # reason and the negative one would pass for it. @@ -461,7 +461,7 @@ def _binding_manifest(binding: object) -> tdfs.Manifest: _MAC = bytes(range(32)) _LEGACY = base64.b64encode(_MAC.hex().encode()).decode() -_SPEC = base64.b64encode(_MAC).decode() +_HEXLESS = base64.b64encode(_MAC).decode() def _binding_hash(m: tdfs.Manifest) -> str: @@ -469,27 +469,30 @@ def _binding_hash(m: tdfs.Manifest) -> str: return pb.hash if isinstance(pb, tdfs.PolicyBinding) else pb -class TestSpecPolicyBinding: +class TestHexlessPolicyBinding: @pytest.mark.parametrize( "binding", [{"alg": "HS256", "hash": _LEGACY}, _LEGACY], ids=["object", "string"], ) - def test_legacy_hex_becomes_spec_form_of_the_same_hmac(self, binding: object): - from test_tdfs import spec_policy_binding + def test_legacy_hex_becomes_hexless_form_of_the_same_hmac(self, binding: object): + from test_tdfs import make_hexless_policy_binding - assert _binding_hash(spec_policy_binding(_binding_manifest(binding))) == _SPEC + assert ( + _binding_hash(make_hexless_policy_binding(_binding_manifest(binding))) + == _HEXLESS + ) - def test_spec_form_is_left_as_is(self): - from test_tdfs import spec_policy_binding + def test_hexless_form_is_left_as_is(self): + from test_tdfs import make_hexless_policy_binding - m = _binding_manifest({"alg": "HS256", "hash": _SPEC}) - assert _binding_hash(spec_policy_binding(m)) == _SPEC + m = _binding_manifest({"alg": "HS256", "hash": _HEXLESS}) + assert _binding_hash(make_hexless_policy_binding(m)) == _HEXLESS - def test_altered_spec_form_is_well_formed_but_wrong(self): - from test_tdfs import change_spec_policy_binding + def test_altered_hexless_form_is_well_formed_but_wrong(self): + from test_tdfs import change_hexless_policy_binding - m = change_spec_policy_binding( + m = change_hexless_policy_binding( _binding_manifest({"alg": "HS256", "hash": _LEGACY}) ) altered = base64.b64decode(_binding_hash(m))