From bf8294e52589a7bddc15c404ffcd34efeaf5fba7 Mon Sep 17 00:00:00 2001 From: Dave Mihalcik Date: Mon, 19 Oct 2026 12:18:35 -0400 Subject: [PATCH 1/3] refactor(xtest): move assertion tests into test_tdf_assertions.py Pure move: the assertion round-trip, manifest-validity, and assertion tamper tests (and change_assertion_statement) leave test_tdfs.py unchanged. The X-Test workflow lists test modules explicitly, so add the new module there so the tests keep running in CI. Signed-off-by: Dave Mihalcik --- .github/workflows/xtest.yml | 4 +- xtest/AGENTS.md | 2 +- xtest/conftest.py | 6 +- xtest/test_tdf_assertions.py | 232 +++++++++++++++++++++++++++++++++++ xtest/test_tdfs.py | 215 -------------------------------- 5 files changed, 238 insertions(+), 221 deletions(-) create mode 100644 xtest/test_tdf_assertions.py diff --git a/.github/workflows/xtest.yml b/.github/workflows/xtest.yml index be200b12..46cc9d12 100644 --- a/.github/workflows/xtest.yml +++ b/.github/workflows/xtest.yml @@ -669,7 +669,7 @@ jobs: if: ${{ env.FOCUS_SDK == 'all' }} run: |- skip_flag=$([[ "$SKIP_RELEASED_PAIRS" == "true" ]] && echo "--skip-released-pairs" || echo "") - uv run pytest -n auto --dist loadscope --html=test-results/sdk-${FOCUS_SDK}-${PLATFORM_TAG}.html --self-contained-html --sdks-encrypt "${ENCRYPT_SDK}" -ra -v $skip_flag test_tdfs.py test_policytypes.py + uv run pytest -n auto --dist loadscope --html=test-results/sdk-${FOCUS_SDK}-${PLATFORM_TAG}.html --self-contained-html --sdks-encrypt "${ENCRYPT_SDK}" -ra -v $skip_flag test_tdfs.py test_tdf_assertions.py test_policytypes.py working-directory: otdftests/xtest env: PLATFORM_DIR: "../../${{ steps.run-platform.outputs.platform-working-dir }}" @@ -681,7 +681,7 @@ jobs: if: ${{ env.FOCUS_SDK != 'all' }} run: |- skip_flag=$([[ "$SKIP_RELEASED_PAIRS" == "true" ]] && echo "--skip-released-pairs" || echo "") - uv run pytest -n auto --dist loadscope --html=test-results/sdk-${FOCUS_SDK}-${PLATFORM_TAG}.html --self-contained-html --sdks-encrypt "${ENCRYPT_SDK}" -ra -v --focus "$FOCUS_SDK" $skip_flag test_tdfs.py test_policytypes.py + uv run pytest -n auto --dist loadscope --html=test-results/sdk-${FOCUS_SDK}-${PLATFORM_TAG}.html --self-contained-html --sdks-encrypt "${ENCRYPT_SDK}" -ra -v --focus "$FOCUS_SDK" $skip_flag test_tdfs.py test_tdf_assertions.py test_policytypes.py working-directory: otdftests/xtest env: PLATFORM_DIR: "../../${{ steps.run-platform.outputs.platform-working-dir }}" diff --git a/xtest/AGENTS.md b/xtest/AGENTS.md index 29bf2ea7..8b8667d0 100644 --- a/xtest/AGENTS.md +++ b/xtest/AGENTS.md @@ -11,7 +11,7 @@ fixture system. | Path | Contents | |------|----------| -| `test_*.py` | Test modules. One per concern: `test_tdfs.py` (core roundtrip), `test_abac.py` (ABAC), `test_legacy.py` (golden TDFs), `test_policytypes.py`, `test_self.py`, `test_audit_logs.py`, `test_pqc.py`. | +| `test_*.py` | Test modules. One per concern: `test_tdfs.py` (core roundtrip), `test_tdf_assertions.py` (assertions), `test_abac.py` (ABAC), `test_legacy.py` (golden TDFs), `test_policytypes.py`, `test_self.py`, `test_audit_logs.py`, `test_pqc.py`. | | `conftest.py` | `pytest_addoption` + the encrypt/decrypt SDK parametrization. Defines `--sdks`, `--sdks-encrypt`, `--sdks-decrypt`, `--containers`, `--no-audit-logs`. | | `fixtures/` | Module-scoped pytest fixtures: `attributes.py`, `keys.py`, `audit.py`, `assertions.py`, `kas.py`, `encryption.py`, `obligations.py`. | | `tdfs.py` | SDK abstraction layer — wraps the `cli.sh` shims under `sdk//dist//`. | diff --git a/xtest/conftest.py b/xtest/conftest.py index ac9e4aaa..d5108194 100644 --- a/xtest/conftest.py +++ b/xtest/conftest.py @@ -683,9 +683,9 @@ def chunky_pt_file(tmp_dir: Path) -> Path: Independent of ``--sizes`` on purpose. Adding 'chunky' to the session's sizes would fan out every test that takes :func:`pt_file` -- the whole of - test_tdfs.py and test_policytypes.py -- to pay for a property one test - needs. A separate fixture buys the coverage for one extra encrypt and - decrypt of 5 MiB, which is cheap enough for the PR gate. + test_tdfs.py, test_tdf_assertions.py and test_policytypes.py -- to pay for + a property one test needs. A separate fixture buys the coverage for one + extra encrypt and decrypt of 5 MiB, which is cheap enough for the PR gate. """ return _plaintext_of(tmp_dir, "chunky") diff --git a/xtest/test_tdf_assertions.py b/xtest/test_tdf_assertions.py new file mode 100644 index 00000000..e1e65809 --- /dev/null +++ b/xtest/test_tdf_assertions.py @@ -0,0 +1,232 @@ +import filecmp +import subprocess +from pathlib import Path + +import pytest + +import tdfs +from abac import Attribute +from fixtures.encryption import EncryptFactory +from test_tdfs import assert_tamper_error, looks_like_422 + +#### MANIFEST VALIDITY TESTS + + +def test_manifest_validity_with_assertions( + encrypt_sdk: tdfs.SDK, + assertion_file_no_keys: str, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + if not in_focus & {encrypt_sdk}: + pytest.skip("Not in focus") + if not encrypt_sdk.supports("assertions"): + pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") + ct_file = encrypted_tdf( + encrypt_sdk, + az=assertion_file_no_keys, + attr_values=attribute_default_rsa.value_fqns, + ) + + tdfs.validate_manifest_schema(ct_file) + + +#### ASSERTION TESTS + + +def test_tdf_assertions_unkeyed( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + pt_file: Path, + assertion_file_no_keys: str, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + 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) + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + if not encrypt_sdk.supports("assertions"): + pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") + if not decrypt_sdk.supports("assertions"): + pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertions") + ct_file = encrypted_tdf( + encrypt_sdk, + az=assertion_file_no_keys, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + attr_values=attribute_default_rsa.value_fqns, + ) + rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) + decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") + assert filecmp.cmp(pt_file, rt_file) + + +def test_tdf_assertions_with_keys( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + pt_file: Path, + assertion_file_rs_and_hs_keys: str, + assertion_verification_file_rs_and_hs_keys: str, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + 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) + tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) + if not encrypt_sdk.supports("assertions"): + pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") + if not decrypt_sdk.supports("assertion_verification"): + pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertion_verification") + ct_file = encrypted_tdf( + encrypt_sdk, + az=assertion_file_rs_and_hs_keys, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + attr_values=attribute_default_rsa.value_fqns, + ) + rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) + + decrypt_sdk.decrypt( + ct_file, + rt_file, + "ztdf", + assertion_verification_file_rs_and_hs_keys, + ) + assert filecmp.cmp(pt_file, rt_file) + + +def test_tdf_assertions_422_format( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + pt_file: Path, + assertion_file_rs_and_hs_keys: str, + assertion_verification_file_rs_and_hs_keys: str, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + 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) + if not encrypt_sdk.supports("hexaflexible"): + pytest.skip( + f"Encrypt SDK {encrypt_sdk} doesn't support targeting container format 4.2.2" + ) + if not encrypt_sdk.supports("assertions"): + pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") + if not decrypt_sdk.supports("assertion_verification"): + pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertion_verification") + ct_file = encrypted_tdf( + encrypt_sdk, + az=assertion_file_rs_and_hs_keys, + target_mode="4.2.2", + attr_values=attribute_default_rsa.value_fqns, + ) + looks_like_422(tdfs.manifest(ct_file)) + + rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) + + decrypt_sdk.decrypt( + ct_file, + rt_file, + "ztdf", + assertion_verification_file_rs_and_hs_keys, + ) + assert filecmp.cmp(pt_file, rt_file) + + +#### TAMPER + +## TAMPER FUNCTIONS + + +def change_assertion_statement(manifest: tdfs.Manifest) -> tdfs.Manifest: + assert manifest.assertions + assertion = manifest.assertions[0] + assertion.statement.value = "tampered" + manifest.assertions[0] = assertion + return manifest + + +## ASSERTION TAMPER TESTS + + +def test_tdf_with_altered_assertion_statement( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + assertion_file_no_keys: str, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + 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) + if not encrypt_sdk.supports("assertions"): + pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") + if not decrypt_sdk.supports("assertions"): + pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertions") + ct_file = encrypted_tdf( + encrypt_sdk, + az=assertion_file_no_keys, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + attr_values=attribute_default_rsa.value_fqns, + ) + b_file = tdfs.update_manifest( + "altered_assertion_statement", ct_file, change_assertion_statement + ) + 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_tamper_error(exc, "assertion", decrypt_sdk) + + +def test_tdf_with_altered_assertion_with_keys( + encrypt_sdk: tdfs.SDK, + decrypt_sdk: tdfs.SDK, + assertion_file_rs_and_hs_keys: str, + assertion_verification_file_rs_and_hs_keys: str, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + 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) + if not encrypt_sdk.supports("assertions"): + pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") + if not decrypt_sdk.supports("assertion_verification"): + pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertion_verification") + ct_file = encrypted_tdf( + encrypt_sdk, + az=assertion_file_rs_and_hs_keys, + target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), + attr_values=attribute_default_rsa.value_fqns, + ) + b_file = tdfs.update_manifest( + "altered_assertion_statement", ct_file, change_assertion_statement + ) + rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) + try: + decrypt_sdk.decrypt( + b_file, + rt_file, + "ztdf", + assertion_verification_file_rs_and_hs_keys, + expect_error=True, + ) + assert False, "decrypt succeeded unexpectedly" + except subprocess.CalledProcessError as exc: + assert_tamper_error(exc, "assertion", decrypt_sdk) diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index eb8b7fb2..39d7d974 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -319,135 +319,6 @@ def test_manifest_validity( tdfs.validate_manifest_schema(ct_file) -def test_manifest_validity_with_assertions( - encrypt_sdk: tdfs.SDK, - assertion_file_no_keys: str, - in_focus: set[tdfs.SDK], - attribute_default_rsa: Attribute, - encrypted_tdf: EncryptFactory, -): - if not in_focus & {encrypt_sdk}: - pytest.skip("Not in focus") - if not encrypt_sdk.supports("assertions"): - pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") - ct_file = encrypted_tdf( - encrypt_sdk, - az=assertion_file_no_keys, - attr_values=attribute_default_rsa.value_fqns, - ) - - tdfs.validate_manifest_schema(ct_file) - - -#### ASSERTION TESTS - - -def test_tdf_assertions_unkeyed( - encrypt_sdk: tdfs.SDK, - decrypt_sdk: tdfs.SDK, - pt_file: Path, - assertion_file_no_keys: str, - in_focus: set[tdfs.SDK], - attribute_default_rsa: Attribute, - encrypted_tdf: EncryptFactory, -): - 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) - tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) - if not encrypt_sdk.supports("assertions"): - pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") - if not decrypt_sdk.supports("assertions"): - pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertions") - ct_file = encrypted_tdf( - encrypt_sdk, - az=assertion_file_no_keys, - target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), - attr_values=attribute_default_rsa.value_fqns, - ) - rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) - decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") - assert filecmp.cmp(pt_file, rt_file) - - -def test_tdf_assertions_with_keys( - encrypt_sdk: tdfs.SDK, - decrypt_sdk: tdfs.SDK, - pt_file: Path, - assertion_file_rs_and_hs_keys: str, - assertion_verification_file_rs_and_hs_keys: str, - in_focus: set[tdfs.SDK], - attribute_default_rsa: Attribute, - encrypted_tdf: EncryptFactory, -): - 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) - tdfs.skip_connectrpc_skew(encrypt_sdk, decrypt_sdk, pfs) - if not encrypt_sdk.supports("assertions"): - pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") - if not decrypt_sdk.supports("assertion_verification"): - pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertion_verification") - ct_file = encrypted_tdf( - encrypt_sdk, - az=assertion_file_rs_and_hs_keys, - target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), - attr_values=attribute_default_rsa.value_fqns, - ) - rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) - - decrypt_sdk.decrypt( - ct_file, - rt_file, - "ztdf", - assertion_verification_file_rs_and_hs_keys, - ) - assert filecmp.cmp(pt_file, rt_file) - - -def test_tdf_assertions_422_format( - encrypt_sdk: tdfs.SDK, - decrypt_sdk: tdfs.SDK, - pt_file: Path, - assertion_file_rs_and_hs_keys: str, - assertion_verification_file_rs_and_hs_keys: str, - in_focus: set[tdfs.SDK], - attribute_default_rsa: Attribute, - encrypted_tdf: EncryptFactory, -): - 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) - if not encrypt_sdk.supports("hexaflexible"): - pytest.skip( - f"Encrypt SDK {encrypt_sdk} doesn't support targeting container format 4.2.2" - ) - if not encrypt_sdk.supports("assertions"): - pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") - if not decrypt_sdk.supports("assertion_verification"): - pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertion_verification") - ct_file = encrypted_tdf( - encrypt_sdk, - az=assertion_file_rs_and_hs_keys, - target_mode="4.2.2", - attr_values=attribute_default_rsa.value_fqns, - ) - looks_like_422(tdfs.manifest(ct_file)) - - rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) - - decrypt_sdk.decrypt( - ct_file, - rt_file, - "ztdf", - assertion_verification_file_rs_and_hs_keys, - ) - assert filecmp.cmp(pt_file, rt_file) - - #### TAMPER ## TAMPER FUNCTIONS @@ -517,14 +388,6 @@ def change_encrypted_segment_size(manifest: tdfs.Manifest) -> tdfs.Manifest: return manifest -def change_assertion_statement(manifest: tdfs.Manifest) -> tdfs.Manifest: - assert manifest.assertions - assertion = manifest.assertions[0] - assertion.statement.value = "tampered" - manifest.assertions[0] = assertion - return manifest - - def change_payload_end(payload_bytes: bytes) -> bytes: return change_last_three(payload_bytes) @@ -758,84 +621,6 @@ def test_tdf_with_altered_enc_seg_size( assert_tamper_error(exc, "", decrypt_sdk) -## ASSERTION TAMPER TESTS - - -def test_tdf_with_altered_assertion_statement( - encrypt_sdk: tdfs.SDK, - decrypt_sdk: tdfs.SDK, - assertion_file_no_keys: str, - in_focus: set[tdfs.SDK], - attribute_default_rsa: Attribute, - encrypted_tdf: EncryptFactory, -): - 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) - if not encrypt_sdk.supports("assertions"): - pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") - if not decrypt_sdk.supports("assertions"): - pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertions") - ct_file = encrypted_tdf( - encrypt_sdk, - az=assertion_file_no_keys, - target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), - attr_values=attribute_default_rsa.value_fqns, - ) - b_file = tdfs.update_manifest( - "altered_assertion_statement", ct_file, change_assertion_statement - ) - 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_tamper_error(exc, "assertion", decrypt_sdk) - - -def test_tdf_with_altered_assertion_with_keys( - encrypt_sdk: tdfs.SDK, - decrypt_sdk: tdfs.SDK, - assertion_file_rs_and_hs_keys: str, - assertion_verification_file_rs_and_hs_keys: str, - in_focus: set[tdfs.SDK], - attribute_default_rsa: Attribute, - encrypted_tdf: EncryptFactory, -): - 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) - if not encrypt_sdk.supports("assertions"): - pytest.skip(f"{encrypt_sdk} sdk doesn't yet support assertions") - if not decrypt_sdk.supports("assertion_verification"): - pytest.skip(f"{decrypt_sdk} sdk doesn't yet support assertion_verification") - ct_file = encrypted_tdf( - encrypt_sdk, - az=assertion_file_rs_and_hs_keys, - target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), - attr_values=attribute_default_rsa.value_fqns, - ) - b_file = tdfs.update_manifest( - "altered_assertion_statement", ct_file, change_assertion_statement - ) - rt_file = encrypted_tdf.rt_file(b_file, decrypt_sdk) - try: - decrypt_sdk.decrypt( - b_file, - rt_file, - "ztdf", - assertion_verification_file_rs_and_hs_keys, - expect_error=True, - ) - assert False, "decrypt succeeded unexpectedly" - except subprocess.CalledProcessError as exc: - assert_tamper_error(exc, "assertion", decrypt_sdk) - - ## PAYLOAD TAMPER TESTS From b61c429b4a0755add85a33f5badbad55c9872090 Mon Sep 17 00:00:00 2001 From: Dave Mihalcik Date: Mon, 19 Oct 2026 12:19:38 -0400 Subject: [PATCH 2/3] test(xtest): check assertion contents and bindings in Python The assertion tests only checked that decrypt succeeds, so two SDKs that agreed on a wrong binding would pass. assertions.assert_bound now re-derives each binding from the written manifest: - each requested assertion is present and was copied through unchanged - no signingKey leaks into the manifest - binding is a compact JWS whose assertionHash is the JCS hash of the assertion and whose assertionSig is base64(aggregateHash || hash), using the hex form for manifests without a schemaVersion - RS256/HS256 signatures verify against the fixture keys; assertions without their own key must be HS256 (over the payload key) test_assertions_units.py covers the helper offline and runs in check.yml. Signed-off-by: Dave Mihalcik --- .github/workflows/check.yml | 2 +- xtest/assertions.py | 160 ++++++++++++++++++++++++++++++- xtest/tdfs.py | 11 +++ xtest/test_assertions_units.py | 167 +++++++++++++++++++++++++++++++++ xtest/test_tdf_assertions.py | 19 ++++ 5 files changed, 356 insertions(+), 3 deletions(-) create mode 100644 xtest/test_assertions_units.py diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index c463033b..555de17f 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -53,7 +53,7 @@ jobs: test_bench_stats.py test_bench_measure.py test_bench_runner.py test_bench_arms.py test_bench_config.py test_bench_report.py test_sdk_commands.py test_tdfs_units.py test_encryption_units.py - test_sizes_units.py test_zip64_units.py + test_sizes_units.py test_zip64_units.py test_assertions_units.py working-directory: xtest - name: Lint and test otdf-local run: | diff --git a/xtest/assertions.py b/xtest/assertions.py index 2d2aa1a5..e3d63a0b 100644 --- a/xtest/assertions.py +++ b/xtest/assertions.py @@ -1,6 +1,15 @@ -from typing import Literal +import base64 +import hashlib +import hmac +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Literal -from pydantic import BaseModel, ConfigDict, Field +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter Type = Literal["handling", "other"] Scope = Literal["payload", "tdo"] @@ -38,3 +47,150 @@ class Assertion(BaseModel): class AssertionVerificationKeys(BaseModel): keys: dict[str, AssertionKey] + + +def load_assertions(path: Path | str) -> list[Assertion]: + return TypeAdapter(list[Assertion]).validate_json(Path(path).read_text()) + + +def load_verification_keys(path: Path | str) -> AssertionVerificationKeys: + return AssertionVerificationKeys.model_validate_json(Path(path).read_text()) + + +#### BINDING CHECKS +# +# Every SDK binds an assertion the same way: a compact JWS whose claims are +# assertionHash: hex(SHA-256(JCS(assertion without its binding))) +# assertionSig: base64(aggregateHash || assertionHash) +# where aggregateHash is the concatenated, base64-decoded segment hashes, and +# assertionHash is appended as raw bytes -- or as its hex text, for manifests +# with no schemaVersion (4.2.2 and earlier). The JWS is signed with the +# assertion's own key if it has one, else HS256 over the payload key. + + +@dataclass(frozen=True) +class Jws: + header: dict[str, Any] + claims: dict[str, Any] + signing_input: bytes + signature: bytes + + +def _b64url_decode(part: str) -> bytes: + return base64.urlsafe_b64decode(part + "=" * (-len(part) % 4)) + + +def parse_jws(token: str) -> Jws: + parts = token.split(".") + assert len(parts) == 3, ( + f"binding signature is not a compact JWS ({len(parts)} part(s))" + ) + header, claims = (json.loads(_b64url_decode(p)) for p in parts[:2]) + return Jws( + header=header, + claims=claims, + signing_input=f"{parts[0]}.{parts[1]}".encode("ascii"), + signature=_b64url_decode(parts[2]), + ) + + +def verify_jws_signature(jws: Jws, key: AssertionKey) -> None: + assert jws.header.get("alg") == key.alg, ( + f"binding signed with {jws.header.get('alg')!r}, expected {key.alg!r}" + ) + match key.alg: + case "HS256": + # The CLIs all take the HS256 key string's UTF-8 bytes as the secret. + mac = hmac.new(key.key.encode("utf-8"), jws.signing_input, hashlib.sha256) + assert hmac.compare_digest(mac.digest(), jws.signature), ( + "HS256 binding signature does not verify" + ) + case "RS256": + public_key = serialization.load_pem_public_key(key.key.encode("ascii")) + assert isinstance(public_key, rsa.RSAPublicKey) + try: + public_key.verify( + jws.signature, + jws.signing_input, + padding.PKCS1v15(), + hashes.SHA256(), + ) + except InvalidSignature: + raise AssertionError( + "RS256 binding signature does not verify" + ) from None + case _: + assert False, f"unsupported binding algorithm {key.alg!r}" + + +def assertion_hash(assertion: dict[str, Any]) -> str: + """Hex SHA-256 of the RFC 8785 (JCS) form of ``assertion`` minus its binding. + + Sorted keys and no whitespace is JCS for what an assertion holds: strings + and ASCII keys. It is not JCS for numbers, which no assertion carries. + """ + unbound = {k: v for k, v in assertion.items() if k != "binding"} + canonical = json.dumps( + unbound, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +def assertion_sig(manifest: dict[str, Any], hash_hex: str) -> str: + """The ``assertionSig`` claim a writer must produce for ``hash_hex``.""" + segments = manifest["encryptionInformation"]["integrityInformation"]["segments"] + aggregate = b"".join(base64.b64decode(s["hash"], validate=True) for s in segments) + if manifest.get("schemaVersion"): + bound = bytes.fromhex(hash_hex) + else: + bound = hash_hex.encode("ascii") + return base64.b64encode(aggregate + bound).decode("ascii") + + +def assert_bound( + manifest: dict[str, Any], + expected: list[Assertion], + keys: AssertionVerificationKeys | None = None, +) -> None: + """Assert ``manifest`` carries ``expected``, each bound to this payload. + + Checks what a reader checks, plus what no reader can: that the writer + copied each assertion through unchanged and kept its signing key out of + the manifest. Every assertion's claims are checked, including any the + writer added on its own (java adds system-metadata). Signatures are only + verified for assertions in ``keys``; the rest are signed with the payload + key, which the test never sees. + """ + written = {a["id"]: a for a in manifest.get("assertions") or []} + for want in expected: + assert want.id in written, ( + f"assertion {want.id!r} missing from manifest; found {sorted(written)}" + ) + raw = written[want.id] + assert "signingKey" not in raw, ( + f"assertion {want.id!r} leaked its signing key into the manifest" + ) + got = Assertion.model_validate(raw) + unbound = {"binding", "signingKey"} + assert got.model_dump(exclude=unbound) == want.model_dump(exclude=unbound), ( + f"assertion {want.id!r} was altered on write" + ) + + for id, raw in written.items(): + binding = Binding.model_validate(raw.get("binding")) + jws = parse_jws(binding.signature) + hash_hex = assertion_hash(raw) + assert jws.claims.get("assertionHash") == hash_hex, ( + f"assertion {id!r}: assertionHash claim is not the hash of the assertion" + ) + assert jws.claims.get("assertionSig") == assertion_sig(manifest, hash_hex), ( + f"assertion {id!r}: assertionSig claim does not bind this payload" + ) + key = keys.keys.get(id) if keys else None + if key: + verify_jws_signature(jws, key) + else: + assert jws.header.get("alg") == "HS256", ( + f"assertion {id!r} has no key of its own, so must be HS256 over " + f"the payload key; got {jws.header.get('alg')!r}" + ) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 60bf26e9..b1b741bd 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -543,6 +543,17 @@ def manifest(tdf_file: Path) -> Manifest: return Manifest.model_validate_json(manifestEntry.read()) +def manifest_json(tdf_file: Path) -> dict[str, Any]: + """The manifest as written, for checks that a model round-trip would blur. + + Assertion hashes are computed over the JSON the writer emitted; rebuilding + it from :class:`Manifest` drops fields the model does not declare. + """ + with zipfile.ZipFile(tdf_file, "r") as tdfz: + with tdfz.open("0.manifest.json") as manifestEntry: + return json.load(manifestEntry) + + # Create a modified variant of a TDF by manipulating its manifest def update_manifest( scenario_name: str, tdf_file: Path, manifest_change: Callable[[Manifest], Manifest] diff --git a/xtest/test_assertions_units.py b/xtest/test_assertions_units.py new file mode 100644 index 00000000..1e5e2156 --- /dev/null +++ b/xtest/test_assertions_units.py @@ -0,0 +1,167 @@ +"""Offline tests for the assertion binding checks in assertions.py. + +No platform, no SDK, no subprocess. ``assert_bound`` re-derives every binding +the SDKs write; if it quietly accepted a bad one, the assertion tests in +``test_tdf_assertions.py`` would pass for the wrong reason. So each case here +signs a manifest independently, the way the SDKs do, and checks that the +helper accepts it and rejects each kind of damage. +""" + +import base64 +import copy +import hashlib +import hmac +import json +from collections.abc import Callable +from typing import Any + +import pytest +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa + +import assertions + +HS256_KEY = base64.b64encode(b"0123456789abcdef0123456789abcdef").decode() +RSA_KEY = rsa.generate_private_key(public_exponent=65537, key_size=2048) +RSA_PUBLIC_PEM = ( + RSA_KEY.public_key() + .public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo + ) + .decode() +) +KEYS = assertions.AssertionVerificationKeys( + keys={ + "hs": assertions.AssertionKey(alg="HS256", key=HS256_KEY), + "rs": assertions.AssertionKey(alg="RS256", key=RSA_PUBLIC_PEM), + } +) + + +def _b64url(b: bytes) -> str: + return base64.urlsafe_b64encode(b).rstrip(b"=").decode() + + +def _sign(alg: str, claims: dict[str, str]) -> str: + header = _b64url(json.dumps({"alg": alg, "typ": "JWT"}).encode()) + payload = _b64url(json.dumps(claims).encode()) + signing_input = f"{header}.{payload}".encode() + if alg == "HS256": + sig = hmac.new(HS256_KEY.encode(), signing_input, hashlib.sha256).digest() + else: + sig = RSA_KEY.sign(signing_input, padding.PKCS1v15(), hashes.SHA256()) + return f"{header}.{payload}.{_b64url(sig)}" + + +def _manifest(schema_version: str | None) -> dict[str, Any]: + """A manifest with two segments and an HS256 and an RS256 assertion.""" + digests = [hashlib.sha256(b"one").digest(), hashlib.sha256(b"two").digest()] + if schema_version is None: + # 4.2.2 segment hashes are base64 of the hex text. + digests = [d.hex().encode() for d in digests] + manifest: dict[str, Any] = { + "encryptionInformation": { + "integrityInformation": { + "segments": [{"hash": base64.b64encode(d).decode()} for d in digests] + } + }, + "assertions": [], + } + if schema_version: + manifest["schemaVersion"] = schema_version + for id, alg in (("hs", "HS256"), ("rs", "RS256")): + assertion: dict[str, Any] = { + "id": id, + "type": "handling", + "scope": "tdo", + "appliesToState": "encrypted", + "statement": {"format": "json", "schema": "urn:x", "value": '{"a":"b"}'}, + } + h = assertions.assertion_hash(assertion) + claims = { + "assertionHash": h, + "assertionSig": assertions.assertion_sig(manifest, h), + } + assertion["binding"] = {"method": "jws", "signature": _sign(alg, claims)} + manifest["assertions"].append(assertion) + return manifest + + +def _expected(manifest: dict[str, Any]) -> list[assertions.Assertion]: + return [ + assertions.Assertion.model_validate( + {k: v for k, v in a.items() if k != "binding"} + ) + for a in manifest["assertions"] + ] + + +@pytest.mark.parametrize("schema_version", ["4.3.0", None]) +def test_accepts_a_correct_binding(schema_version: str | None): + manifest = _manifest(schema_version) + assertions.assert_bound(manifest, _expected(manifest), KEYS) + + +def test_legacy_and_current_bind_differently(): + """The schemaVersion switch is load-bearing, not decorative.""" + h = "00" * 32 + current = _manifest("4.3.0") + legacy = copy.deepcopy(current) + del legacy["schemaVersion"] + assert assertions.assertion_sig(current, h) != assertions.assertion_sig(legacy, h) + + +def _set_statement(m: dict[str, Any]) -> None: + m["assertions"][0]["statement"]["value"] = "tampered" + + +def _break_signature(m: dict[str, Any]) -> None: + binding = m["assertions"][1]["binding"] + binding["signature"] = binding["signature"][:-4] + "AAAA" + + +def _leak_key(m: dict[str, Any]) -> None: + m["assertions"][0]["signingKey"] = {"alg": "HS256", "key": HS256_KEY} + + +def _drop(m: dict[str, Any]) -> None: + del m["assertions"][1] + + +def _drop_binding(m: dict[str, Any]) -> None: + del m["assertions"][0]["binding"] + + +def _change_segment(m: dict[str, Any]) -> None: + segments = m["encryptionInformation"]["integrityInformation"]["segments"] + segments[0]["hash"] = base64.b64encode(b"\0" * 32).decode() + + +@pytest.mark.parametrize( + "damage, message", + [ + (_set_statement, "altered on write"), + (_break_signature, "does not verify"), + (_leak_key, "leaked its signing key"), + (_drop, "missing from manifest"), + (_drop_binding, "validation error"), + (_change_segment, "does not bind this payload"), + ], +) +def test_rejects_damage(damage: Callable[[dict[str, Any]], None], message: str): + manifest = _manifest("4.3.0") + expected = _expected(manifest) + damage(manifest) + with pytest.raises((AssertionError, ValueError), match=message): + assertions.assert_bound(manifest, expected, KEYS) + + +def test_unkeyed_assertion_must_be_hs256(): + """An assertion with no key of its own is signed with the payload key.""" + manifest = _manifest("4.3.0") + with pytest.raises(AssertionError, match="must be HS256"): + assertions.assert_bound( + manifest, + _expected(manifest), + assertions.AssertionVerificationKeys(keys={"hs": KEYS.keys["hs"]}), + ) diff --git a/xtest/test_tdf_assertions.py b/xtest/test_tdf_assertions.py index e1e65809..9219a02b 100644 --- a/xtest/test_tdf_assertions.py +++ b/xtest/test_tdf_assertions.py @@ -4,6 +4,7 @@ import pytest +import assertions import tdfs from abac import Attribute from fixtures.encryption import EncryptFactory @@ -30,6 +31,10 @@ def test_manifest_validity_with_assertions( ) tdfs.validate_manifest_schema(ct_file) + assertions.assert_bound( + tdfs.manifest_json(ct_file), + assertions.load_assertions(assertion_file_no_keys), + ) #### ASSERTION TESTS @@ -59,6 +64,10 @@ def test_tdf_assertions_unkeyed( target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), attr_values=attribute_default_rsa.value_fqns, ) + assertions.assert_bound( + tdfs.manifest_json(ct_file), + assertions.load_assertions(assertion_file_no_keys), + ) rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) decrypt_sdk.decrypt(ct_file, rt_file, "ztdf") assert filecmp.cmp(pt_file, rt_file) @@ -89,6 +98,11 @@ def test_tdf_assertions_with_keys( target_mode=tdfs.select_target_version(encrypt_sdk, decrypt_sdk), attr_values=attribute_default_rsa.value_fqns, ) + assertions.assert_bound( + tdfs.manifest_json(ct_file), + assertions.load_assertions(assertion_file_rs_and_hs_keys), + assertions.load_verification_keys(assertion_verification_file_rs_and_hs_keys), + ) rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) decrypt_sdk.decrypt( @@ -129,6 +143,11 @@ def test_tdf_assertions_422_format( attr_values=attribute_default_rsa.value_fqns, ) looks_like_422(tdfs.manifest(ct_file)) + assertions.assert_bound( + tdfs.manifest_json(ct_file), + assertions.load_assertions(assertion_file_rs_and_hs_keys), + assertions.load_verification_keys(assertion_verification_file_rs_and_hs_keys), + ) rt_file = encrypted_tdf.rt_file(ct_file, decrypt_sdk) From 64793d8bade9e96b97456df11736d4bf1bd66501 Mon Sep 17 00:00:00 2001 From: Dave Mihalcik Date: Tue, 20 Oct 2026 12:39:49 -0400 Subject: [PATCH 3/3] clarify a doc comment --- xtest/tdfs.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/xtest/tdfs.py b/xtest/tdfs.py index b1b741bd..8cabeffc 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -544,10 +544,10 @@ def manifest(tdf_file: Path) -> Manifest: def manifest_json(tdf_file: Path) -> dict[str, Any]: - """The manifest as written, for checks that a model round-trip would blur. + """The manifest as written, including fields not present in the Manifest type. - Assertion hashes are computed over the JSON the writer emitted; rebuilding - it from :class:`Manifest` drops fields the model does not declare. + This is useful for computing assertion over the JSON the writer emitted, including + removed or experimental fields the model does not declare. """ with zipfile.ZipFile(tdf_file, "r") as tdfz: with tdfz.open("0.manifest.json") as manifestEntry: