diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 60bf26e9..3a669885 100644 --- a/xtest/tdfs.py +++ b/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")) diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index eb8b7fb2..e6aef517 100644 --- a/xtest/test_tdfs.py +++ b/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 diff --git a/xtest/test_tdfs_units.py b/xtest/test_tdfs_units.py index 2a9c17b4..67a1e574 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.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