diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index c463033b..ec719858 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_rangeserve_units.py working-directory: xtest - name: Lint and test otdf-local run: | diff --git a/xtest/conftest.py b/xtest/conftest.py index ac9e4aaa..3205d183 100644 --- a/xtest/conftest.py +++ b/xtest/conftest.py @@ -67,6 +67,7 @@ def pytest_report_header() -> list[str]: "fixtures.audit", "fixtures.encryption", "fixtures.bench", + "fixtures.range_server", ] diff --git a/xtest/fixtures/range_server.py b/xtest/fixtures/range_server.py new file mode 100644 index 00000000..901f2e31 --- /dev/null +++ b/xtest/fixtures/range_server.py @@ -0,0 +1,17 @@ +"""The DSPX-5111 range server as a fixture, stopped when the test ends.""" + +from collections.abc import Callable, Iterator +from contextlib import ExitStack + +import pytest + +import rangeserve + + +@pytest.fixture +def range_server() -> Iterator[ + Callable[[rangeserve.Source], tuple[str, list[rangeserve.Request]]] +]: + """Call with a source; returns ``(url, request_log)``. Any number per test.""" + with ExitStack() as stack: + yield lambda source: stack.enter_context(rangeserve.serve(source)) diff --git a/xtest/rangeserve.py b/xtest/rangeserve.py new file mode 100644 index 00000000..385c25b9 --- /dev/null +++ b/xtest/rangeserve.py @@ -0,0 +1,305 @@ +"""Serve a TDF over HTTP ``Range`` requests, optionally with most of it missing. + +DSPX-5111. A partial decrypt of a 100 GiB-50 TiB TDF needs only the ZIP head, +the encrypted segments covering the requested plaintext, and the tail +(manifest, central directory, end records). :func:`retention_plan` says which +ciphertext bytes those are, :class:`ExtentStore` keeps just them, and +:func:`serve` puts either a store or an ordinary file behind a loopback HTTP +server that logs every request. + +A request touching a byte that was not kept gets ``410 Gone``, never zeros: a +reader that over-fetches must fail loudly rather than decrypt garbage, and +must not pass a test by accident. + +All ranges here are half-open ``(start, end)`` pairs. +""" + +import bisect +import os +import re +import struct +import threading +from collections.abc import Iterator, Sequence +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import NamedTuple, Protocol + +import zipinspect + +#: AES-GCM adds a 12-byte IV and a 16-byte tag to every segment. +PAYLOAD_GCM_OVERHEAD = 28 + +_LOCAL_HEADER_SIG = b"PK\x03\x04" +_LOCAL_HEADER_FIXED_SIZE = 30 +_CHUNK = 1 << 20 + + +class Dropped(Exception): + """A read touched bytes that were deliberately not kept.""" + + +def payload_start(path: Path) -> int: + """Offset of the first byte of ``0.payload``'s data in the TDF at ``path``.""" + cd = zipinspect.central_directory(path) + entry = next((e for e in cd.entries if e.name == "0.payload"), None) + if entry is None: + raise zipinspect.MalformedZipError(f"{path}: no 0.payload entry") + with path.open("rb") as f: + f.seek(entry.local_header_offset) + header = f.read(_LOCAL_HEADER_FIXED_SIZE) + if len(header) < _LOCAL_HEADER_FIXED_SIZE or header[:4] != _LOCAL_HEADER_SIG: + raise zipinspect.MalformedZipError( + f"{path}: no local file header at {entry.local_header_offset}" + ) + name_len, extra_len = struct.unpack_from(" list[tuple[int, int]]: + """Sort extents and merge any that overlap or touch; drop empty ones.""" + merged: list[tuple[int, int]] = [] + for start, end in sorted(e for e in extents if e[1] > e[0]): + if merged and start <= merged[-1][1]: + merged[-1] = (merged[-1][0], max(merged[-1][1], end)) + else: + merged.append((start, end)) + return merged + + +def retention_plan( + ranges: Sequence[tuple[int, int]], + *, + plaintext_length: int, + segment_size: int, + payload_start: int, +) -> list[tuple[int, int]]: + """Ciphertext extents a partial decrypt of plaintext ``ranges`` reads. + + Always includes the head, ``[0, payload_start)``. Excludes the tail, + whose start is not known until the payload has been written. + + Fixed before encrypt: ciphertext is not deterministic, so a dropped + extent cannot be regenerated afterwards. + """ + if segment_size <= 0: + raise ValueError(f"segment_size must be positive, got {segment_size}") + stride = segment_size + PAYLOAD_GCM_OVERHEAD + extents = [(0, payload_start)] + for start, end in ranges: + if not 0 <= start <= end <= plaintext_length: + raise ValueError( + f"plaintext range {start}-{end} is outside 0-{plaintext_length}" + ) + if start == end: + continue + first, last = start // segment_size, (end - 1) // segment_size + last_size = min(segment_size, plaintext_length - last * segment_size) + extents.append( + ( + payload_start + first * stride, + payload_start + last * stride + last_size + PAYLOAD_GCM_OVERHEAD, + ) + ) + return merge(extents) + + +class Source(Protocol): + @property + def length(self) -> int: ... + + def check(self, start: int, end: int) -> None: + """Raise :class:`Dropped` if any of ``[start, end)`` was not kept.""" + ... + + def read(self, start: int, end: int) -> bytes: ... + + +class FullFile: + """An ordinary file: every byte is available.""" + + def __init__(self, path: Path): + self.path = path + self.length = path.stat().st_size + + def check(self, start: int, end: int) -> None: + pass + + def read(self, start: int, end: int) -> bytes: + with self.path.open("rb") as f: + return os.pread(f.fileno(), end - start, start) + + +class ExtentStore: + """Kept extents of a large logical file, packed into one small data file. + + Not a sparse file: ext4 caps a file at 16 TiB, and the 50 TiB tier must + not depend on the filesystem. Disk use is the kept bytes, nothing more. + """ + + def __init__(self, directory: Path, logical_length: int): + self.length = logical_length + self.data_path = directory / "extents.bin" + self._data = self.data_path.open("w+b") + # Sorted, non-overlapping (logical_start, logical_end, physical_start). + self._index: list[tuple[int, int, int]] = [] + + def close(self) -> None: + self._data.close() + + @property + def physical_size(self) -> int: + return self.data_path.stat().st_size + + def add(self, offset: int, data: bytes) -> None: + end = offset + len(data) + if not data: + return + if offset < 0 or end > self.length: + raise ValueError(f"extent {offset}-{end} is outside 0-{self.length}") + i = bisect.bisect(self._index, (offset,)) + if (i > 0 and self._index[i - 1][1] > offset) or ( + i < len(self._index) and self._index[i][0] < end + ): + raise ValueError(f"extent {offset}-{end} overlaps one already kept") + physical = self._data.seek(0, os.SEEK_END) + self._data.write(data) + self._data.flush() + prev = self._index[i - 1] if i else None + # Streaming writes arrive as many contiguous chunks; keep one entry. + if prev and prev[1] == offset and prev[2] + (prev[1] - prev[0]) == physical: + self._index[i - 1] = (prev[0], end, prev[2]) + else: + self._index.insert(i, (offset, end, physical)) + + def _pieces(self, start: int, end: int) -> Iterator[tuple[int, int]]: + """Yield ``(physical_offset, n)`` covering ``[start, end)``, or raise.""" + i = bisect.bisect(self._index, (start, float("inf"))) - 1 + pos = start + while pos < end: + if ( + i < 0 + or i >= len(self._index) + or not (self._index[i][0] <= pos < self._index[i][1]) + ): + raise Dropped(f"bytes {pos}-{end} were not kept") + lstart, lend, physical = self._index[i] + n = min(lend, end) - pos + yield physical + pos - lstart, n + pos += n + i += 1 + + def check(self, start: int, end: int) -> None: + for _ in self._pieces(start, end): + pass + + def read(self, start: int, end: int) -> bytes: + fd = self._data.fileno() + return b"".join(os.pread(fd, n, p) for p, n in self._pieces(start, end)) + + +class Request(NamedTuple): + method: str + range: str | None + status: int + bytes_served: int + + +_RANGE = re.compile(r"bytes=(\d*)-(\d*)") + + +def _parse_range(header: str, length: int) -> tuple[int, int] | None: + """Resolve a single-range header to ``(start, end)``; None if unsatisfiable. + + Raises ValueError for anything that is not exactly one byte range. + """ + m = _RANGE.fullmatch(header.strip()) + if not m or m.group(1) == m.group(2) == "": + raise ValueError(f"unsupported Range: {header!r}") + first, last = m.group(1), m.group(2) + if first == "": + start, end = max(0, length - int(last)), length + return (start, end) if int(last) > 0 and length > 0 else None + start = int(first) + end = length if last == "" else min(int(last) + 1, length) + if last != "" and int(last) < start: + raise ValueError(f"unsupported Range: {header!r}") + return (start, end) if start < length else None + + +@contextmanager +def serve(source: Source) -> Iterator[tuple[str, list[Request]]]: + """Serve ``source`` on an ephemeral loopback port. + + Yields ``(url, log)``. ``log`` gains one :class:`Request` per request. + The server thread is stopped and joined on exit. + """ + log: list[Request] = [] + + class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, format: str, *args: object) -> None: + pass + + def _reply(self, status: int, body: bytes = b"", **headers: str) -> None: + self.send_response(status) + self.send_header("Content-Length", str(len(body))) + for k, v in headers.items(): + self.send_header(k.replace("_", "-"), v) + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + log.append(Request(self.command, self.headers["Range"], status, 0)) + + def do_HEAD(self) -> None: + self.send_response(200) + self.send_header("Content-Length", str(source.length)) + self.send_header("Accept-Ranges", "bytes") + self.end_headers() + log.append(Request("HEAD", self.headers["Range"], 200, 0)) + + def do_GET(self) -> None: + header = self.headers["Range"] + start, end, status = 0, source.length, 200 + if header is not None: + try: + resolved = _parse_range(header, source.length) + except ValueError as e: + return self._reply(400, str(e).encode()) + if resolved is None: + return self._reply(416, Content_Range=f"bytes */{source.length}") + (start, end), status = resolved, 206 + try: + source.check(start, end) + except Dropped as e: + return self._reply(410, str(e).encode()) + self.send_response(status) + self.send_header("Content-Length", str(end - start)) + self.send_header("Accept-Ranges", "bytes") + if status == 206: + self.send_header( + "Content-Range", f"bytes {start}-{end - 1}/{source.length}" + ) + self.end_headers() + served = 0 + try: + for pos in range(start, end, _CHUNK): + self.wfile.write(source.read(pos, min(pos + _CHUNK, end))) + served = min(pos + _CHUNK, end) - start + finally: + log.append(Request("GET", header, status, served)) + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server.daemon_threads = True + port = server.server_address[1] + thread = threading.Thread( + target=server.serve_forever, name=f"rangeserve-{port}", daemon=True + ) + thread.start() + try: + yield f"http://127.0.0.1:{port}/payload.tdf", log + finally: + server.shutdown() + server.server_close() + thread.join() diff --git a/xtest/test_rangeserve_units.py b/xtest/test_rangeserve_units.py new file mode 100644 index 00000000..42bf3da2 --- /dev/null +++ b/xtest/test_rangeserve_units.py @@ -0,0 +1,208 @@ +"""Offline tests for the sparse TDF store and range server (DSPX-5111). + +No platform, no SDK. The "TDF" is a stored ZIP whose payload has the layout +of real segments; nothing here decrypts, so random bytes stand in for +ciphertext. +""" + +import random +import socket +import threading +import urllib.error +import urllib.request +import zipfile +from pathlib import Path + +import pytest + +import rangeserve +from rangeserve import PAYLOAD_GCM_OVERHEAD as GCM +from rangeserve import ExtentStore, FullFile, retention_plan, serve + +SEG = 1000 +PT_LEN = 4 * SEG + 300 # four whole segments and a short fifth + + +def get(url: str, range_header: str | None = None) -> tuple[int, bytes]: + req = urllib.request.Request(url) + if range_header: + req.add_header("Range", range_header) + try: + with urllib.request.urlopen(req, timeout=10) as r: + return r.status, r.read() + except urllib.error.HTTPError as e: + return e.code, e.read() + + +@pytest.fixture +def tdf(tmp_path: Path) -> Path: + path = tmp_path / "fake.tdf" + ct_len = PT_LEN + 5 * GCM + with zipfile.ZipFile(path, "w", zipfile.ZIP_STORED) as z: + z.writestr("0.payload", random.Random(5111).randbytes(ct_len)) + z.writestr("0.manifest.json", '{"payload": {}}' + " " * 3000) + return path + + +def plan(*ranges: tuple[int, int], start: int = 100) -> list[tuple[int, int]]: + return retention_plan( + ranges, plaintext_length=PT_LEN, segment_size=SEG, payload_start=start + ) + + +# --- retention_plan --------------------------------------------------------- + +STRIDE = SEG + GCM + + +def test_plan_first_segment(): + assert plan((0, 1)) == [(0, 100 + STRIDE)] + + +def test_plan_last_short_segment(): + last = 100 + 4 * STRIDE + assert plan((PT_LEN - 1, PT_LEN)) == [(0, 100), (last, last + 300 + GCM)] + + +def test_plan_crosses_segment_boundary(): + got = plan((2 * SEG - 1, 2 * SEG + 1)) + assert got == [(0, 100), (100 + STRIDE, 100 + 3 * STRIDE)] + + +def test_plan_disjoint_and_overlapping_ranges(): + got = plan((2 * SEG + 5, 2 * SEG + 6), (3 * SEG, 3 * SEG + 1), (10, 20), start=50) + assert got == [(0, 50 + STRIDE), (50 + 2 * STRIDE, 50 + 4 * STRIDE)] + + +def test_plan_empty_range_keeps_only_head(): + assert plan((7, 7)) == [(0, 100)] + + +@pytest.mark.parametrize("bad", [(-1, 3), (5, 4), (0, PT_LEN + 1)]) +def test_plan_rejects_out_of_bounds(bad: tuple[int, int]): + with pytest.raises(ValueError): + plan(bad) + + +# --- store and server against a small TDF ----------------------------------- + + +def test_payload_start_finds_data(tdf: Path): + start = rangeserve.payload_start(tdf) + with zipfile.ZipFile(tdf) as z: + first = z.read("0.payload")[:64] + assert tdf.read_bytes()[start : start + 64] == first + + +def test_full_mode_serves_whole_file(tdf: Path): + data = tdf.read_bytes() + with serve(FullFile(tdf)) as (url, log): + assert get(url) == (200, data) + assert get(url, "bytes=10-19") == (206, data[10:20]) + assert get(url, "bytes=-22") == (206, data[-22:]) + assert get(url, f"bytes={len(data) - 5}-") == (206, data[-5:]) + assert get(url, f"bytes=0-{len(data) * 2}") == (206, data) + assert get(url, f"bytes={len(data)}-")[0] == 416 + assert get(url, "bytes=0-1,5-6")[0] == 400 + assert [r.status for r in log] == [200, 206, 206, 206, 206, 416, 400] + assert log[1] == rangeserve.Request("GET", "bytes=10-19", 206, 10) + + +def test_head_reports_length(tdf: Path): + with serve(FullFile(tdf)) as (url, log): + req = urllib.request.Request(url, method="HEAD") + with urllib.request.urlopen(req, timeout=10) as r: + assert int(r.headers["Content-Length"]) == tdf.stat().st_size + assert r.headers["Accept-Ranges"] == "bytes" + assert log == [rangeserve.Request("HEAD", None, 200, 0)] + + +def test_sparse_store_serves_kept_and_refuses_dropped(tdf: Path, tmp_path: Path): + data = tdf.read_bytes() + start = rangeserve.payload_start(tdf) + payload_end = start + PT_LEN + 5 * GCM + kept = rangeserve.merge( + plan((SEG - 1, SEG + 1), (2 * SEG + 1, 2 * SEG + 2), start=start) + + [(payload_end, len(data))] # the tail + ) + store = ExtentStore(tmp_path, len(data)) + for s, e in kept: + store.add(s, data[s:e]) + assert store.physical_size == sum(e - s for s, e in kept) + + dropped = [(kept[i][1], kept[i + 1][0]) for i in range(len(kept) - 1)] + assert dropped, "plan should leave gaps" + with serve(store) as (url, log): + for s, e in kept: + assert get(url, f"bytes={s}-{e - 1}") == (206, data[s:e]) + for s, e in dropped: + assert get(url, f"bytes={s}-{e - 1}")[0] == 410 + # Straddling a kept/dropped edge is still refused. + assert get(url, f"bytes={s - 1}-{s}")[0] == 410 + assert get(url)[0] == 410 + assert all(r.bytes_served == 0 for r in log if r.status == 410) + store.close() + + +def test_store_rejects_overlap(tmp_path: Path): + store = ExtentStore(tmp_path, 100) + store.add(10, b"x" * 10) + with pytest.raises(ValueError): + store.add(15, b"y") + with pytest.raises(ValueError): + store.add(95, b"z" * 10) + store.add(20, b"w" * 5) # contiguous: merged, readable across the seam + assert store.read(15, 25) == b"x" * 5 + b"w" * 5 + store.close() + + +# --- huge logical lengths, tiny disk ---------------------------------------- + + +@pytest.mark.parametrize( + "logical", [100 * 2**30, 5 * 2**40, 50 * 2**40], ids=["100GiB", "5TiB", "50TiB"] +) +def test_huge_logical_length_bounded_disk(tmp_path: Path, logical: int): + pieces = { + 0: b"H" * 64, + 2**32 - 3: b"A" * 7, # straddles 2^32 + 2**40 + 11: b"B" * 13, + logical - 4096: b"T" * 4096, + } + store = ExtentStore(tmp_path, logical) + for off, b in pieces.items(): + if off + len(b) <= logical: + store.add(off, b) + assert store.physical_size <= 64 + 7 + 13 + 4096 + with serve(store) as (url, _): + req = urllib.request.Request(url, method="HEAD") + with urllib.request.urlopen(req, timeout=10) as r: + assert int(r.headers["Content-Length"]) == logical + assert get(url, "bytes=-4096") == (206, b"T" * 4096) + assert get(url, f"bytes={2**32 - 3}-{2**32 + 3}") == (206, b"A" * 7) + want = (206, b"B" * 13) if 2**40 < logical else (416, b"") + assert get(url, f"bytes={2**40 + 11}-{2**40 + 23}") == want + assert get(url, f"bytes={logical // 2}-{logical // 2 + 1}")[0] == 410 + store.close() + + +# --- lifecycle -------------------------------------------------------------- + + +def test_server_stops_on_exit(tdf: Path): + with serve(FullFile(tdf)) as (url, _): + port = int(url.split(":")[2].split("/")[0]) + assert get(url, "bytes=0-0")[0] == 206 + assert not any(t.name == f"rangeserve-{port}" for t in threading.enumerate()) + with ( + pytest.raises(ConnectionRefusedError), + socket.create_connection(("127.0.0.1", port), timeout=2), + ): + pass + + +def test_fixture_serves_several(range_server, tdf: Path): + a, _ = range_server(FullFile(tdf)) + b, _ = range_server(FullFile(tdf)) + assert a != b + assert get(a, "bytes=0-3") == get(b, "bytes=0-3")