diff --git a/xtest/sizes.py b/xtest/sizes.py index 32184049..3f9ddaa1 100644 --- a/xtest/sizes.py +++ b/xtest/sizes.py @@ -33,7 +33,7 @@ #: 2.1 GiB. Sits ~102 MiB inside the low edge of the broken window. #: #: The margin is the point. A TDF writes ``0.payload`` first and -#: ``0.manifest.json`` after it, so the manifest's local-header offset is +#: ``manifest.json`` after it, so the manifest's local-header offset is #: roughly the payload size -- and that offset is the value under test. The #: gap to 2**31 has to be wider than anything that could shift it: segment #: padding, manifest length, per-entry header overhead. 102 MiB is not a diff --git a/xtest/tdfs.py b/xtest/tdfs.py index 8f3bf8e0..c2f2f38d 100644 --- a/xtest/tdfs.py +++ b/xtest/tdfs.py @@ -8,7 +8,7 @@ import urllib.parse import urllib.request import zipfile -from collections.abc import Callable +from collections.abc import Callable, Iterable from pathlib import Path from typing import Any, Literal, TypeIs, get_args @@ -476,9 +476,37 @@ class Manifest(BaseModel): ) +#: The manifest entry name required by the OpenTDF spec. From +#: ``spec/schema/OpenTDF/manifest.md``: "The ``manifest.json`` file MUST be in +#: JSON format and reside within the root of the OpenTDF Zip archive." +#: +#: This suite reads and asserts this name and no other. Any other spelling is +#: a container the spec does not describe, and a reader that accepts one +#: cannot report that the writer produced it. +MANIFEST_ENTRY = "manifest.json" + + +def manifest_entry_name(names: Iterable[str]) -> str: + """Check that an archive carries the spec's manifest entry, and name it. + + Returns :data:`MANIFEST_ENTRY` or raises. The indirection buys the error + message: ``zipfile``'s own ``KeyError`` reports only the name it wanted, so + an archive that files the manifest elsewhere is indistinguishable from one + with no manifest at all. Listing the actual members says which it was. + """ + present = set(names) + if MANIFEST_ENTRY in present: + return MANIFEST_ENTRY + raise KeyError( + f"archive has no {MANIFEST_ENTRY!r} entry, which the OpenTDF spec " + f"requires at the archive root; it holds {sorted(present)}" + ) + + def manifest(tdf_file: Path) -> Manifest: with zipfile.ZipFile(tdf_file, "r") as tdfz: - with tdfz.open("0.manifest.json") as manifestEntry: + entry = manifest_entry_name(tdfz.namelist()) + with tdfz.open(entry) as manifestEntry: return Manifest.model_validate_json(manifestEntry.read()) @@ -491,11 +519,12 @@ def update_manifest( fname = tdf_file.stem unzipped_dir = tmp_dir / f"{fname}-{scenario_name}-unzipped" with zipfile.ZipFile(tdf_file, "r") as zipped: + entry = manifest_entry_name(zipped.namelist()) zipped.extractall(unzipped_dir) - with (unzipped_dir / "0.manifest.json").open("r") as manifest_file: + with (unzipped_dir / entry).open("r") as manifest_file: manifest_data = Manifest.model_validate_json(manifest_file.read()) new_manifest_data = manifest_change(manifest_data) - with (unzipped_dir / "0.manifest.json").open("w") as manifest_file: + with (unzipped_dir / entry).open("w") as manifest_file: # exclude_unset so the rewrite carries only what the original manifest # said plus whatever manifest_change touched. Without it every optional # field is re-emitted as an explicit null, and a reader that distinguishes @@ -641,7 +670,7 @@ def validate_manifest_schema(tdf_file: Path): ## Load the manifest file directly from the zipfile with zipfile.ZipFile(tdf_file, "r") as zipped: - with zipped.open("0.manifest.json") as manifest_file: + with zipped.open(manifest_entry_name(zipped.namelist())) as manifest_file: manifest = json.load(manifest_file) ## Validate diff --git a/xtest/test_tdfs.py b/xtest/test_tdfs.py index eb8b7fb2..0acc4a04 100644 --- a/xtest/test_tdfs.py +++ b/xtest/test_tdfs.py @@ -4,6 +4,7 @@ import re import string import subprocess +import zipfile from pathlib import Path import pytest @@ -339,6 +340,38 @@ def test_manifest_validity_with_assertions( tdfs.validate_manifest_schema(ct_file) +def test_manifest_entry_name_is_spec_compliant( + encrypt_sdk: tdfs.SDK, + in_focus: set[tdfs.SDK], + attribute_default_rsa: Attribute, + encrypted_tdf: EncryptFactory, +): + """The manifest member is named ``manifest.json``, as the spec requires. + + Distinct from ``test_manifest_validity``, which checks the manifest's + *contents* against manifest.schema.json: a manifest that satisfies the + schema byte for byte is still unreadable by a spec-conformant + implementation if the archive files it under some other name. + + Deliberately ungated. A writer that names the entry anything else fails + here, and that red cell is the finding, not a configuration problem to + route around. + """ + if not in_focus & {encrypt_sdk}: + pytest.skip("Not in focus") + ct_file = encrypted_tdf( + encrypt_sdk, + attr_values=attribute_default_rsa.value_fqns, + ) + + with zipfile.ZipFile(ct_file, "r") as z: + names = z.namelist() + assert tdfs.MANIFEST_ENTRY in names, ( + f"{encrypt_sdk} wrote {names}; the spec names the manifest entry " + f"{tdfs.MANIFEST_ENTRY}" + ) + + #### ASSERTION TESTS diff --git a/xtest/test_tdfs_units.py b/xtest/test_tdfs_units.py index e6d86ed7..992b8048 100644 --- a/xtest/test_tdfs_units.py +++ b/xtest/test_tdfs_units.py @@ -9,6 +9,11 @@ The GMAC root forgery (DSPX-4703) is here for a related reason: an exploit helper that quietly forges the wrong bytes would make the security tests in ``test_root_signature.py`` pass for the wrong reason. + +``manifest_entry_name`` is here because it decides which archive member the +whole suite reads as the manifest, and holds that to the spec's +``manifest.json`` alone. Its tests pin the rejection of other spellings as +much as the acceptance of that one. """ import base64 @@ -90,7 +95,13 @@ def test_a_branch_build_is_not(self): # --- tdfs.elides_segment_sizes / tdfs.skip_chunky_skew ------------------------ -def _manifest_zip(tmp_path: Path, name: str, *, elides: bool) -> Path: +def _manifest_zip( + tmp_path: Path, + name: str, + *, + elides: bool, + manifest_entry: str = tdfs.MANIFEST_ENTRY, +) -> Path: """A minimal container with just enough manifest to exercise elides_segment_sizes.""" segments = [ { @@ -129,7 +140,7 @@ def _manifest_zip(tmp_path: Path, name: str, *, elides: bool) -> Path: } p = tmp_path / name with zipfile.ZipFile(p, "w") as z: - z.writestr("0.manifest.json", json.dumps(manifest)) + z.writestr(manifest_entry, json.dumps(manifest)) z.writestr("0.payload", b"") return p @@ -176,6 +187,78 @@ def test_does_not_skip_when_container_does_not_elide(self, tmp_path: Path): tdfs.skip_chunky_skew(ct_file, decrypt_sdk) +# --- tdfs.manifest_entry_name ------------------------------------------------- + + +class TestManifestEntryName: + """Which archive member holds the manifest: ``manifest.json``, and no other. + + The rejection cases carry as much weight as the acceptance one. A reader + that also takes ``0.manifest.json`` reports a conformant archive and a + non-conformant one identically, which is the one thing this lookup must + not do. + """ + + def test_finds_the_spec_name(self): + assert ( + tdfs.manifest_entry_name(["0.payload", "manifest.json"]) == "manifest.json" + ) + + def test_rejects_the_zero_prefixed_name(self): + with pytest.raises(KeyError, match="manifest.json"): + tdfs.manifest_entry_name(["0.payload", "0.manifest.json"]) + + def test_rejects_a_suffix_match(self): + """``0.manifest.json`` ends with the spec name, so the lookup has to be + equality; an ``endswith`` test would readmit it.""" + with pytest.raises(KeyError): + tdfs.manifest_entry_name(["0.payload", "not-a-manifest.json"]) + + def test_the_error_reports_what_the_archive_held(self): + """Without the member list, a misnamed manifest and a missing one give + the same message.""" + with pytest.raises(KeyError) as exc: + tdfs.manifest_entry_name(["0.payload", "0.manifest.json"]) + assert "0.manifest.json" in str(exc.value) + assert "0.payload" in str(exc.value) + + +class TestManifestRequiresTheSpecEntry: + def test_manifest_parses_a_spec_named_container(self, tmp_path: Path): + ct_file = _manifest_zip(tmp_path, "m.tdf", elides=False) + assert tdfs.manifest(ct_file).payload.url == "0.payload" + + def test_manifest_rejects_an_off_spec_named_container(self, tmp_path: Path): + ct_file = _manifest_zip( + tmp_path, "off-spec.tdf", elides=False, manifest_entry="0.manifest.json" + ) + with pytest.raises(KeyError, match="manifest.json"): + tdfs.manifest(ct_file) + + def test_update_manifest_round_trips_under_the_spec_name(self, tmp_path: Path): + """A tamper fixture must differ from its source only in the bytes the + change touched, entry name included.""" + ct_file = _manifest_zip(tmp_path, "u.tdf", elides=False) + + def bump(m: tdfs.Manifest) -> tdfs.Manifest: + m.encryptionInformation.integrityInformation.rootSignature.sig = b"dGFtcGVy" + return m + + out = tdfs.update_manifest("tampered", ct_file, bump) + + with zipfile.ZipFile(out) as z: + assert tdfs.MANIFEST_ENTRY in z.namelist() + root = tdfs.manifest(out).encryptionInformation.integrityInformation + assert root.rootSignature.sig == b"dGFtcGVy" + + def test_update_manifest_rejects_an_off_spec_named_container(self, tmp_path: Path): + ct_file = _manifest_zip( + tmp_path, "off-spec.tdf", elides=False, manifest_entry="0.manifest.json" + ) + with pytest.raises(KeyError, match="manifest.json"): + tdfs.update_manifest("tampered", ct_file, lambda m: m) + + # --- tdfs.forge_gmac_root_signature (DSPX-4703) ------------------------------- diff --git a/xtest/test_zip64_units.py b/xtest/test_zip64_units.py index a935e294..02c1de01 100644 --- a/xtest/test_zip64_units.py +++ b/xtest/test_zip64_units.py @@ -20,6 +20,7 @@ import pytest import conftest +import tdfs import zipinspect from sizes import MEDIUM_BYTES, ZIP64_WINDOW_HIGH, ZIP64_WINDOW_LOW from zipinspect import ZIP64_SENTINEL_32, MalformedZipError @@ -283,10 +284,10 @@ def test_reads_a_real_zip(self, tmp_path: Path): p = tmp_path / "ordinary.zip" with zipfile.ZipFile(p, "w") as z: z.writestr("0.payload", b"a" * 4096) - z.writestr("0.manifest.json", b"{}") + z.writestr(tdfs.MANIFEST_ENTRY, b"{}") cd = zipinspect.central_directory(p) - assert [e.name for e in cd.entries] == ["0.payload", "0.manifest.json"] + assert [e.name for e in cd.entries] == ["0.payload", tdfs.MANIFEST_ENTRY] with zipfile.ZipFile(p) as z: expected = {i.filename: i.header_offset for i in z.infolist()} assert {e.name: e.local_header_offset for e in cd.entries} == expected @@ -326,11 +327,11 @@ def test_reads_a_zip64_end_of_central_directory(self, tmp_path: Path): """ records = [ cen_record("0.payload", raw_offset=0), - cen_record("0.manifest.json", raw_offset=MEDIUM_BYTES), + cen_record(tdfs.MANIFEST_ENTRY, raw_offset=MEDIUM_BYTES), ] p = synth_zip64_eocd(tmp_path / "z64-eocd.zip", records) cd = zipinspect.central_directory(p) - assert [e.name for e in cd.entries] == ["0.payload", "0.manifest.json"] + assert [e.name for e in cd.entries] == ["0.payload", tdfs.MANIFEST_ENTRY] assert cd.entries[1].raw_local_header_offset == MEDIUM_BYTES def test_rejects_a_locator_pointing_at_nothing(self, tmp_path: Path): @@ -354,7 +355,7 @@ def test_raw_value_in_the_window_is_preserved(self, tmp_path: Path): tmp_path / "window.zip", [ cen_record("0.payload", raw_offset=0, raw_usize=MEDIUM_BYTES), - cen_record("0.manifest.json", raw_offset=MEDIUM_BYTES + 64), + cen_record(tdfs.MANIFEST_ENTRY, raw_offset=MEDIUM_BYTES + 64), ], ) manifest = zipinspect.central_directory(p).entries[1] @@ -372,7 +373,7 @@ def test_signed_read_of_a_windowed_offset_goes_negative(self, tmp_path: Path): """ p = synth_zip( tmp_path / "signed.zip", - [cen_record("0.manifest.json", raw_offset=MEDIUM_BYTES)], + [cen_record(tdfs.MANIFEST_ENTRY, raw_offset=MEDIUM_BYTES)], ) (entry,) = zipinspect.central_directory(p).entries assert entry.signed_read_of_offset() < 0 @@ -382,7 +383,7 @@ def test_signed_read_is_harmless_below_the_window(self, tmp_path: Path): """Below 2**31 the two reads agree, which is why smaller payloads miss this.""" offset = ZIP64_WINDOW_LOW - 1 p = synth_zip( - tmp_path / "safe.zip", [cen_record("0.manifest.json", raw_offset=offset)] + tmp_path / "safe.zip", [cen_record(tdfs.MANIFEST_ENTRY, raw_offset=offset)] ) (entry,) = zipinspect.central_directory(p).entries assert entry.signed_read_of_offset() == offset @@ -394,7 +395,7 @@ def test_sentinel_resolves_through_the_extra_field(self, tmp_path: Path): tmp_path / "sentinel.zip", [ cen_record( - "0.manifest.json", + tdfs.MANIFEST_ENTRY, raw_offset=ZIP64_SENTINEL_32, zip64_offset=true_offset, ) @@ -445,7 +446,7 @@ def test_skips_a_foreign_extra_field_record(self, tmp_path: Path): tmp_path / "foreign-extra.zip", [ cen_record( - "0.manifest.json", + tdfs.MANIFEST_ENTRY, raw_offset=ZIP64_SENTINEL_32, zip64_offset=7 * 2**30, extra_prefix=timestamp, @@ -471,7 +472,7 @@ def test_rejects_a_sentinel_with_no_zip64_extra(self, tmp_path: Path): """ p = synth_zip( tmp_path / "sentinel-no-extra.zip", - [cen_record("0.manifest.json", raw_offset=ZIP64_SENTINEL_32)], + [cen_record(tdfs.MANIFEST_ENTRY, raw_offset=ZIP64_SENTINEL_32)], ) with pytest.raises(MalformedZipError, match="local header offset"): zipinspect.central_directory(p) @@ -492,7 +493,7 @@ def test_rejects_more_values_than_the_sentinels_call_for(self, tmp_path: Path): tmp_path / "over-long-extra.zip", [ cen_record( - "0.manifest.json", + tdfs.MANIFEST_ENTRY, raw_offset=ZIP64_SENTINEL_32, zip64_usize=MEDIUM_BYTES, zip64_csize=MEDIUM_BYTES, @@ -539,7 +540,7 @@ def test_rejects_a_record_claiming_more_than_remains(self, tmp_path: Path): tmp_path / "overrun-extra.zip", [ cen_record( - "0.manifest.json", + tdfs.MANIFEST_ENTRY, raw_offset=ZIP64_SENTINEL_32, zip64_offset=5 * 2**30, zip64_extra_len_override=64, @@ -559,7 +560,7 @@ def test_tolerates_the_trailing_disk_start_field(self, tmp_path: Path): tmp_path / "disk-start.zip", [ cen_record( - "0.manifest.json", + tdfs.MANIFEST_ENTRY, raw_offset=ZIP64_SENTINEL_32, zip64_offset=5 * 2**30, zip64_disk_start=0, @@ -586,7 +587,7 @@ def test_truncated_eocd(self, tmp_path: Path): zipinspect.central_directory(p) def test_truncated_central_directory_record(self, tmp_path: Path): - record = cen_record("0.manifest.json", raw_offset=MEDIUM_BYTES) + record = cen_record(tdfs.MANIFEST_ENTRY, raw_offset=MEDIUM_BYTES) p = tmp_path / "short-cen.zip" cd = record[:20] eocd = ( @@ -601,8 +602,8 @@ def test_truncated_central_directory_record(self, tmp_path: Path): def test_central_directory_record_with_a_name_past_the_end(self, tmp_path: Path): """A plausible header whose variable-length fields overrun the buffer.""" - record = cen_record("0.manifest.json", raw_offset=0) - # Claim a 4096-byte name where 15 bytes were written. + record = cen_record(tdfs.MANIFEST_ENTRY, raw_offset=0) + # Claim a 4096-byte name where only the entry name's bytes were written. record = record[:28] + struct.pack("