feat(xtest): cover hexless policy bindings, Base64(HMAC)

xtest/tdfs.py

@@ -1130,6 +1130,28 @@ 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 hexless Base64(HMAC) policy binding
+#: and accept only legacy Base64(hex(HMAC)): opentdf/platform#2095 introduced
+#: the bug, opentdf/platform#4223 fixed it.
+PLATFORM_HEXLESS_POLICY_BINDING_BUG = (0, 5, 3)
+PLATFORM_HEXLESS_POLICY_BINDING_FIX = (0, 29, 1)
+
+
+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.
+    """
+    sv = _parse_semver(os.getenv("PLATFORM_TAG", ""))
+    return (
+        sv is not None
+        and PLATFORM_HEXLESS_POLICY_BINDING_BUG
+        <= sv
+        < PLATFORM_HEXLESS_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"))

xtest/test_tdfs.py

@@ -472,21 +472,54 @@ 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
+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:
+    if isinstance(kao.policyBinding, tdfs.PolicyBinding):
+        kao.policyBinding.hash = hash
     else:
-        altered_hash = base64.b64encode(change_last_three(base64.b64decode(pb)))
-        manifest.encryptionInformation.keyAccess[0].policyBinding = str(altered_hash)
+        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 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 hexless form must still decrypt.
+    """
+    for kao in manifest.encryptionInformation.keyAccess:
+        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)}"
+        _set_policy_binding_hash(kao, base64.b64encode(mac).decode())
     return 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:
     root_sig = manifest.encryptionInformation.integrityInformation.rootSignature.sig
     altered_sig = base64.b64encode(change_last_three(base64.b64decode(root_sig)))
@@ -670,6 +703,77 @@ def test_tdf_with_altered_policy_binding(
         assert_kas_request_error(exc, decrypt_sdk)
 
 
+def test_tdf_with_hexless_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 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.hexless_policy_binding_is_rejected`.
+    """
+    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(
+        "hexless_policy_binding", ct_file, make_hexless_policy_binding
+    )
+    rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk)
+
+    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)
+        return
+
+    decrypt_sdk.decrypt(b_file, rt_file, "ztdf")
+    assert filecmp.cmp(pt_file, rt_file, shallow=False)
+
+
+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 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()
+    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_hexless_policy_binding", ct_file, change_hexless_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
 
 

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.hexless_policy_binding_is_rejected -------------------------------------
+
+
+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.
+    """
+
+    def test_a_pre_fix_release_is_rejected(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setenv("PLATFORM_TAG", "v0.29.0")
+        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_HEXLESS_POLICY_BINDING_BUG
+        monkeypatch.setenv("PLATFORM_TAG", f"v{major}.{minor}.{patch}")
+        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 hexless form."""
+        monkeypatch.setenv("PLATFORM_TAG", "v0.5.2")
+        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_HEXLESS_POLICY_BINDING_FIX
+        monkeypatch.setenv("PLATFORM_TAG", f"v{major}.{minor}.{patch}")
+        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.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.hexless_policy_binding_is_rejected()
+
+
 # --- tdfs.elides_segment_sizes / tdfs.skip_chunky_skew ------------------------
 
 
@@ -375,3 +418,99 @@ 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]
+
+
+# --- make_hexless_policy_binding / change_hexless_policy_binding ------------------------
+#
+# 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.
+
+
+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()
+_HEXLESS = 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 TestHexlessPolicyBinding:
+    @pytest.mark.parametrize(
+        "binding",
+        [{"alg": "HS256", "hash": _LEGACY}, _LEGACY],
+        ids=["object", "string"],
+    )
+    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(make_hexless_policy_binding(_binding_manifest(binding)))
+            == _HEXLESS
+        )
+
+    def test_hexless_form_is_left_as_is(self):
+        from test_tdfs import make_hexless_policy_binding
+
+        m = _binding_manifest({"alg": "HS256", "hash": _HEXLESS})
+        assert _binding_hash(make_hexless_policy_binding(m)) == _HEXLESS
+
+    def test_altered_hexless_form_is_well_formed_but_wrong(self):
+        from test_tdfs import change_hexless_policy_binding
+
+        m = change_hexless_policy_binding(
+            _binding_manifest({"alg": "HS256", "hash": _LEGACY})
+        )
+        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