From 72e8d08ef71eac2b248d4e7203d548970f791c98 Mon Sep 17 00:00:00 2001 From: euphyra-dev Date: Sat, 29 Aug 2026 17:58:56 +0000 Subject: [PATCH 1/6] harden: fail-closed DTD/entity gate before Saxon (C+B) C: stdlib expat StartDoctypeDeclHandler refuses any DOCTYPE before Saxon; same path, no rewrite. B: allowedProtocols=file after PySaxonProcessor (never empty). B does not replace C. 10+10 fixture ids unchanged. --- docs/LIMITS.md | 13 +- src/validate.py | 61 +++++++- testdata/hostile/README.md | 8 + tests/test_dtd_gate.py | 302 +++++++++++++++++++++++++++++++++++++ 4 files changed, 382 insertions(+), 2 deletions(-) create mode 100644 testdata/hostile/README.md create mode 100644 tests/test_dtd_gate.py diff --git a/docs/LIMITS.md b/docs/LIMITS.md index ffe257c..b8c0489 100644 --- a/docs/LIMITS.md +++ b/docs/LIMITS.md @@ -19,7 +19,18 @@ not a published maximum. | `jdk.xml.maxElementDepth` | 100 | | `jdk.xml.entityExpansionLimit` | 2500 | -This change does not alter entity or URI resolution. +## DTD / protocol confine (C + B) + +C (required): stdlib `xml.parsers.expat` `StartDoctypeDeclHandler` refuses +any DOCTYPE before Saxon. Invoice bytes are not rewritten; Saxon uses +`source_file=` on the same path. `files[].sha256` is the hash of those +raw bytes. + +B (additional): after `PySaxonProcessor`, `allowedProtocols=file` (never +empty: that breaks `stylesheet_file=`). Linux-measured HTTP-only extra. +B does not block `file://` SYSTEM; C covers that. No portable knob +blocks both HTTP and file SYSTEM without breaking XSLT load. B never +replaces C. ## Annotation text diff --git a/src/validate.py b/src/validate.py index 1e9584b..040672a 100644 --- a/src/validate.py +++ b/src/validate.py @@ -15,6 +15,7 @@ import re import sys import xml.etree.ElementTree as ET +import xml.parsers.expat from pathlib import Path from typing import Any @@ -22,6 +23,7 @@ ENGINE_NAME = "SaxonC-HE 13.0" ENGINE_PKG_PIN = "saxonche==13.0.0" ENGINE_PKG_PREFIX = "13.0" +ALLOWED_PROTOCOLS_FEATURE = "http://saxon.sf.net/feature/allowedProtocols" CII_NS = "urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100" UBL_INVOICE_NS = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2" @@ -54,6 +56,55 @@ class EngineError(Exception): """XSLT engine error (exit 2).""" +class DtdRefused(Exception): + """Any DOCTYPE is refused before external resolution.""" + + def __init__( + self, + doctype_name: str | None, + sysid: str | None, + pubid: str | None, + has_internal_subset: bool, + ) -> None: + self.doctype_name = doctype_name + self.sysid = sysid + self.pubid = pubid + self.has_internal_subset = has_internal_subset + super().__init__( + f"DOCTYPE is not allowed (name={doctype_name!r} " + f"sysid={sysid!r} pubid={pubid!r})" + ) + + +def gate_doctype(data: bytes) -> None: + """Fail-closed on any DOCTYPE. Does not resolve file or HTTP entities. + + Default expat plus StartDoctypeDeclHandler; abort on first DOCTYPE. + No ExternalEntityRefHandler (would change resolution behaviour). + """ + parser = xml.parsers.expat.ParserCreate() + + def start_doctype(doctype_name, sysid, pubid, has_internal_subset): + raise DtdRefused(doctype_name, sysid, pubid, bool(has_internal_subset)) + + parser.StartDoctypeDeclHandler = start_doctype + parser.Parse(data, True) + + +def refuse_invoice_dtd(path: Path, data: bytes | None = None) -> bytes: + """Read invoice bytes once, refuse any DOCTYPE, return the same bytes.""" + if data is None: + data = path.read_bytes() + try: + gate_doctype(data) + except DtdRefused as exc: + raise ConfigError(f"XML DTD/DOCTYPE is not allowed ({path.name})") from exc + except xml.parsers.expat.ExpatError as exc: + raise ConfigError(f"XML is not well-formed ({path.name}): {exc}") from exc + return data + + + def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: @@ -264,10 +315,17 @@ def __init__(self) -> None: f"Install {ENGINE_PKG_PIN}." ) from exc self._proc = PySaxonProcessor(license=False) + try: + # B additional: HTTP-only extra. Never empty (breaks stylesheet_file=). + # Does not block file SYSTEM; C covers that. B never replaces C. + self._proc.set_configuration_property(ALLOWED_PROTOCOLS_FEATURE, "file") + except Exception: + pass self._xslt = self._proc.new_xslt30_processor() self._compiled: dict[str, Any] = {} def transform(self, xml_path: Path, xslt_path: Path) -> str: + refuse_invoice_dtd(xml_path) key = str(xslt_path) executable = self._compiled.get(key) if executable is None: @@ -457,6 +515,7 @@ def validate_files( annotations: list[str] = [] try: for path in files: + data = refuse_invoice_dtd(path) resolved_syntax = resolve_syntax(path, syntax) xslt_info = xslt[resolved_syntax] svrl = engine.transform(path, Path(xslt_info["path"])) @@ -467,7 +526,7 @@ def validate_files( "failed_assert_count": len(failed), "failed_assert_ids": ids, "path": rel, - "sha256": sha256_file(path), + "sha256": hashlib.sha256(data).hexdigest(), "syntax": resolved_syntax, "verdict": "fail" if ids else "pass", "xslt": xslt_info["logical"], diff --git a/testdata/hostile/README.md b/testdata/hostile/README.md new file mode 100644 index 0000000..9e74e9d --- /dev/null +++ b/testdata/hostile/README.md @@ -0,0 +1,8 @@ +# Hostile DTD fixtures + +Used only by `tests/test_dtd_gate.py` (temporary files with local canary +paths and a loopback HTTP server). Not globbed by selftest official or +mutant steps, nor by the Action jobs (`testdata/official/*.xml`, +`testdata/mutants/*.xml`). + +Do not add DOCTYPE samples under `testdata/official/` or `testdata/mutants/`. diff --git a/tests/test_dtd_gate.py b/tests/test_dtd_gate.py new file mode 100644 index 0000000..d05b63e --- /dev/null +++ b/tests/test_dtd_gate.py @@ -0,0 +1,302 @@ +#!/usr/bin/env python3 +"""C gate: refuse any DOCTYPE before Saxon. Local-runnable, not Action globs.""" + +from __future__ import annotations + +import hashlib +import http.server +import os +import socketserver +import sys +import tempfile +import threading +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +SRC = ROOT / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) + +import validate # noqa: E402 + +CII_MIN = ( + '\n' + f'x' + "\n" +) +CANARY_TOKEN = "CANARY_DTD_GATE_9f2c1a70" + + +class ReuseTCPServer(socketserver.TCPServer): + allow_reuse_address = True + + +def _start_hit_server(hits: list[str]) -> tuple[ReuseTCPServer, str]: + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self) -> None: + hits.append(self.path) + self.send_response(200) + self.send_header("Content-Type", "application/xml") + self.end_headers() + self.wfile.write(b"\n") + + def log_message(self, *_args) -> None: + return None + + httpd = ReuseTCPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + host, port = httpd.server_address[:2] + return httpd, f"http://{host}:{port}" + + +class FakeEngine: + def __init__(self) -> None: + self.calls: list[Path] = [] + + def transform(self, xml_path: Path, xslt_path: Path) -> str: + self.calls.append(Path(xml_path)) + return '' + + def close(self) -> None: + return None + + +def _run_validate(path: Path, engine: FakeEngine) -> dict: + tmp_root = path.parent + + def fake_load_xslt(root: Path, version: str) -> dict: + dummy = tmp_root / "dummy.xslt" + dummy.write_text("\n") + return { + "CII": {"path": dummy, "sha256": "0", "logical": "dummy"}, + "UBL": {"path": dummy, "sha256": "0", "logical": "dummy"}, + } + + orig_engine = validate.SaxonEngine + orig_load = validate.load_xslt + validate.SaxonEngine = lambda: engine # type: ignore[misc, assignment] + validate.load_xslt = fake_load_xslt # type: ignore[assignment] + try: + return validate.validate_files( + [path], + syntax="auto", + fail_on_raw="never", + version="1.3.16", + root=tmp_root, + cwd=tmp_root, + ) + finally: + validate.SaxonEngine = orig_engine + validate.load_xslt = orig_load + + +class GateDoctypeTests(unittest.TestCase): + def test_no_dtd_parse_ok(self) -> None: + data = CII_MIN.encode("utf-8") + before = hashlib.sha256(data).hexdigest() + validate.gate_doctype(data) + self.assertEqual(hashlib.sha256(data).hexdigest(), before) + + def test_internal_doctype_refused(self) -> None: + data = ( + b'\n' + b' ]>\n' + b"&x;\n" + ) + with self.assertRaises(validate.DtdRefused): + validate.gate_doctype(data) + + def test_system_file_refused_without_reading_canary(self) -> None: + with tempfile.TemporaryDirectory(prefix="dtd-file-") as td: + folder = Path(td) + canary = folder / "canary.txt" + canary.write_bytes(CANARY_TOKEN.encode("ascii")) + invoice = folder / "system-file.xml" + invoice.write_text( + '\n' + f' ]>\n' + "&ext;\n", + encoding="utf-8", + newline="\n", + ) + original_mode = canary.stat().st_mode + try: + os.chmod(canary, 0) + except OSError: + pass + try: + with self.assertRaises(validate.ConfigError) as ctx: + validate.refuse_invoice_dtd(invoice) + finally: + try: + os.chmod(canary, original_mode) + except OSError: + pass + message = str(ctx.exception) + self.assertIn("DOCTYPE", message) + self.assertNotIn("well-formed", message) + self.assertEqual(canary.read_bytes().decode("ascii"), CANARY_TOKEN) + + def test_system_http_refused_zero_get(self) -> None: + hits: list[str] = [] + httpd, base = _start_hit_server(hits) + try: + data = ( + '\n' + f' ]>\n' + "&ext;\n" + ).encode("utf-8") + with self.assertRaises(validate.DtdRefused): + validate.gate_doctype(data) + self.assertEqual(hits, []) + finally: + httpd.shutdown() + httpd.server_close() + + def test_public_doctype_refused_zero_get(self) -> None: + hits: list[str] = [] + httpd, base = _start_hit_server(hits) + try: + data = ( + '\n' + f'\n' + "ok\n" + ).encode("utf-8") + with self.assertRaises(validate.DtdRefused): + validate.gate_doctype(data) + self.assertEqual(hits, []) + finally: + httpd.shutdown() + httpd.server_close() + + def test_official_fixtures_have_no_doctype(self) -> None: + official = ROOT / "testdata" / "official" + mutants = ROOT / "testdata" / "mutants" + xml_files = sorted(official.glob("*.xml")) + sorted(mutants.glob("*.xml")) + if len(xml_files) != 20: + self.skipTest("official/mutant fixtures not present") + for path in xml_files: + raw = path.read_bytes() + validate.gate_doctype(raw) + self.assertNotIn(b" None: + with tempfile.TemporaryDirectory(prefix="dtd-ok-") as td: + path = Path(td) / "control.xml" + payload = CII_MIN.encode("utf-8") + path.write_bytes(payload) + engine = FakeEngine() + result = _run_validate(path, engine) + self.assertEqual(len(engine.calls), 1) + self.assertEqual(engine.calls[0].resolve(), path.resolve()) + self.assertEqual( + result["payload"]["files"][0]["sha256"], + hashlib.sha256(payload).hexdigest(), + ) + self.assertEqual(path.read_bytes(), payload) + + def test_internal_doctype_never_reaches_saxon(self) -> None: + with tempfile.TemporaryDirectory(prefix="dtd-int-") as td: + path = Path(td) / "internal.xml" + path.write_text( + '\n' + ' ]>\n' + f'&x;' + "\n", + encoding="utf-8", + newline="\n", + ) + engine = FakeEngine() + with self.assertRaises(validate.ConfigError) as ctx: + _run_validate(path, engine) + self.assertIn("DOCTYPE", str(ctx.exception)) + self.assertEqual(engine.calls, []) + + def test_system_file_never_reads_canary_never_reaches_saxon(self) -> None: + with tempfile.TemporaryDirectory(prefix="dtd-sysfile-") as td: + folder = Path(td) + canary = folder / "canary.txt" + canary.write_bytes(CANARY_TOKEN.encode("ascii")) + path = folder / "system-file.xml" + path.write_text( + '\n' + f'\n' + f'x' + "\n", + encoding="utf-8", + newline="\n", + ) + original_mode = canary.stat().st_mode + try: + os.chmod(canary, 0) + except OSError: + pass + engine = FakeEngine() + try: + with self.assertRaises(validate.ConfigError) as ctx: + _run_validate(path, engine) + finally: + try: + os.chmod(canary, original_mode) + except OSError: + pass + self.assertIn("DOCTYPE", str(ctx.exception)) + self.assertEqual(engine.calls, []) + self.assertEqual(canary.read_bytes().decode("ascii"), CANARY_TOKEN) + + def test_system_http_zero_get_never_reaches_saxon(self) -> None: + hits: list[str] = [] + httpd, base = _start_hit_server(hits) + try: + with tempfile.TemporaryDirectory(prefix="dtd-http-") as td: + path = Path(td) / "system-http.xml" + path.write_text( + '\n' + f'\n' + f'x' + "\n", + encoding="utf-8", + newline="\n", + ) + engine = FakeEngine() + with self.assertRaises(validate.ConfigError) as ctx: + _run_validate(path, engine) + self.assertIn("DOCTYPE", str(ctx.exception)) + self.assertEqual(engine.calls, []) + self.assertEqual(hits, []) + finally: + httpd.shutdown() + httpd.server_close() + + def test_public_zero_get_never_reaches_saxon(self) -> None: + hits: list[str] = [] + httpd, base = _start_hit_server(hits) + try: + with tempfile.TemporaryDirectory(prefix="dtd-pub-") as td: + path = Path(td) / "public.xml" + path.write_text( + '\n' + f'\n' + f'x' + "\n", + encoding="utf-8", + newline="\n", + ) + engine = FakeEngine() + with self.assertRaises(validate.ConfigError) as ctx: + _run_validate(path, engine) + self.assertIn("DOCTYPE", str(ctx.exception)) + self.assertEqual(engine.calls, []) + self.assertEqual(hits, []) + finally: + httpd.shutdown() + httpd.server_close() + + +if __name__ == "__main__": + unittest.main() From 6c0d1370b2ab641c44b1e7837cd0c45e8d8156fb Mon Sep 17 00:00:00 2001 From: euphyra-dev Date: Sat, 29 Aug 2026 18:00:18 +0000 Subject: [PATCH 2/6] chore: refresh supply-chain tree.sha256 for SECURITY.md Recompute the digest the same way check_supply_chain.py fingerprints the tree. SECURITY.md content is unchanged (PVR docs already on main). --- supply-chain/tree.sha256 | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/supply-chain/tree.sha256 b/supply-chain/tree.sha256 index 2c95b35..41bc8e6 100644 --- a/supply-chain/tree.sha256 +++ b/supply-chain/tree.sha256 @@ -1,7 +1,7 @@ 2b822e8f93016f265f6bca96a70c862ecedf16c49de7e07ece70554653bcec2f .github/workflows/selftest.yml 4fc0db0a2924b8d652e3b6c470afa221826ebf70ce5619694da135eda7bf2043 .github/workflows/supply-chain.yml cde5dd46b60a596e80d13b6cf65fda4c6909fc8db427d1b1696f7936d0bdd231 .github/workflows/upstream-drift.yml -8675774bd92b18e038b7a19786ef2cfc26531d4488566b830f48f8698b3284a9 SECURITY.md +54bbb865bb13b198b732f09fe5e21f8f43fb8698892df066956ec9eb17ef8b76 SECURITY.md 1bc5269de1bd0866a09da38607095ed34a557a4aafa94f1a95de04e474e63bd4 action.yml dbf524202f4601ced7e1ce680e53132a24f9f0382cc1111bb63d679c096cd177 docs/RELEASE_AND_RULESET.md fe6eac77aab78b4f1cc756f8c90c6ff6cc4ebacafea0f3e5a4ef2763da10f6a4 examples/validate-invoices.yml From a9cb20dd38027affa7c933bed914f08d764474c5 Mon Sep 17 00:00:00 2001 From: euphyra-dev Date: Sat, 29 Aug 2026 18:02:04 +0000 Subject: [PATCH 3/6] fix: hash invoices with sha256_file; keep C gate + B protocols validate_files still records files[].sha256 via sha256_file(path) (raw bytes, no rewrite). C remains expat StartDoctypeDeclHandler. B remains allowedProtocols=file. Hostile samples stay in tests, not testdata globs. --- README.md | 4 + docs/LIMITS.md | 22 +- src/validate.py | 84 +++--- testdata/hostile/README.md | 8 - tests/test_dtd_gate.py | 601 +++++++++++++++++++++++-------------- 5 files changed, 435 insertions(+), 284 deletions(-) delete mode 100644 testdata/hostile/README.md diff --git a/README.md b/README.md index b2e95a6..f99f0b7 100644 --- a/README.md +++ b/README.md @@ -153,6 +153,10 @@ Engine pin: `saxonche==13.0.0` (SaxonC-HE 13.0). The JSON report is canonical (`sort_keys`, no timestamps, relative paths). Two consecutive runs of the same files must produce the same `report-sha256`. +Invoices that contain a DOCTYPE are refused before XSLT (stdlib expat). +The invoice file is not rewritten; `files[].sha256` is the raw file +bytes. See [`docs/LIMITS.md`](docs/LIMITS.md). + `auto` maps: - `urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100` → CII diff --git a/docs/LIMITS.md b/docs/LIMITS.md index b8c0489..68adbb5 100644 --- a/docs/LIMITS.md +++ b/docs/LIMITS.md @@ -21,16 +21,18 @@ not a published maximum. ## DTD / protocol confine (C + B) -C (required): stdlib `xml.parsers.expat` `StartDoctypeDeclHandler` refuses -any DOCTYPE before Saxon. Invoice bytes are not rewritten; Saxon uses -`source_file=` on the same path. `files[].sha256` is the hash of those -raw bytes. - -B (additional): after `PySaxonProcessor`, `allowedProtocols=file` (never -empty: that breaks `stylesheet_file=`). Linux-measured HTTP-only extra. -B does not block `file://` SYSTEM; C covers that. No portable knob -blocks both HTTP and file SYSTEM without breaking XSLT load. B never -replaces C. +C (mandatory): stdlib `xml.parsers.expat` `StartDoctypeDeclHandler` refuses +any DOCTYPE before Saxon. The invoice file is not rewritten; Saxon +`source_file=` uses the same path and the same bytes that passed the +gate. `files[].sha256` is the SHA256 of those raw bytes. User invoices +are not newline-normalized. `ET.iterparse` is syntax detection only; it +is not this gate. + +B (additional): SaxonC 13 +`set_configuration_property('http://saxon.sf.net/feature/allowedProtocols', +'file')` blocks HTTP/HTTPS as an extra control. It does not replace C +(`file:` SYSTEM still needs C). Empty `allowedProtocols` is not used: +it breaks `stylesheet_file=` and `source_file=`. ## Annotation text diff --git a/src/validate.py b/src/validate.py index 040672a..18af667 100644 --- a/src/validate.py +++ b/src/validate.py @@ -23,7 +23,11 @@ ENGINE_NAME = "SaxonC-HE 13.0" ENGINE_PKG_PIN = "saxonche==13.0.0" ENGINE_PKG_PREFIX = "13.0" -ALLOWED_PROTOCOLS_FEATURE = "http://saxon.sf.net/feature/allowedProtocols" + +# B additional: extra HTTP block on SaxonC 13. Does not replace C. +# Empty string must not be used — it breaks stylesheet_file= and source_file=. +SAXON_ALLOWED_PROTOCOLS_FEATURE = "http://saxon.sf.net/feature/allowedProtocols" +SAXON_ALLOWED_PROTOCOLS_VALUE = "file" CII_NS = "urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100" UBL_INVOICE_NS = "urn:oasis:names:specification:ubl:schema:xsd:Invoice-2" @@ -56,54 +60,59 @@ class EngineError(Exception): """XSLT engine error (exit 2).""" -class DtdRefused(Exception): - """Any DOCTYPE is refused before external resolution.""" +class DtdRefused(ConfigError): + """Any DOCTYPE is refused before the XSLT engine sees the file.""" def __init__( self, - doctype_name: str | None, - sysid: str | None, - pubid: str | None, + name: str, + doctype_name: object, + sysid: object, + pubid: object, has_internal_subset: bool, ) -> None: self.doctype_name = doctype_name self.sysid = sysid self.pubid = pubid self.has_internal_subset = has_internal_subset - super().__init__( - f"DOCTYPE is not allowed (name={doctype_name!r} " - f"sysid={sysid!r} pubid={pubid!r})" - ) + super().__init__(f"DOCTYPE is not allowed ({name})") -def gate_doctype(data: bytes) -> None: - """Fail-closed on any DOCTYPE. Does not resolve file or HTTP entities. +def gate_invoice_bytes(data: bytes, *, name: str = "input") -> None: + """Fail-closed on any DOCTYPE via stdlib expat. Does not rewrite bytes. - Default expat plus StartDoctypeDeclHandler; abort on first DOCTYPE. - No ExternalEntityRefHandler (would change resolution behaviour). + ``StartDoctypeDeclHandler`` is the gate. This is not a regex scan. + ElementTree syntax detection is a separate step, not this gate. """ parser = xml.parsers.expat.ParserCreate() def start_doctype(doctype_name, sysid, pubid, has_internal_subset): - raise DtdRefused(doctype_name, sysid, pubid, bool(has_internal_subset)) + raise DtdRefused( + name, doctype_name, sysid, pubid, bool(has_internal_subset) + ) parser.StartDoctypeDeclHandler = start_doctype - parser.Parse(data, True) - - -def refuse_invoice_dtd(path: Path, data: bytes | None = None) -> bytes: - """Read invoice bytes once, refuse any DOCTYPE, return the same bytes.""" - if data is None: - data = path.read_bytes() try: - gate_doctype(data) - except DtdRefused as exc: - raise ConfigError(f"XML DTD/DOCTYPE is not allowed ({path.name})") from exc + parser.Parse(data, True) + except DtdRefused: + raise except xml.parsers.expat.ExpatError as exc: - raise ConfigError(f"XML is not well-formed ({path.name}): {exc}") from exc - return data + raise ConfigError(f"XML is not well-formed ({name}): {exc}") from exc +def gate_invoice_path(path: Path) -> str: + """Parse immutable path bytes; refuse DOCTYPE; return sha256 of those bytes. + + Saxon must be given this same path afterwards. The file is not rewritten. + """ + data = path.read_bytes() + before = hashlib.sha256(data).hexdigest() + gate_invoice_bytes(data, name=path.name) + after = hashlib.sha256(path.read_bytes()).hexdigest() + if after != before: + raise EngineError(f"invoice bytes changed after DTD gate ({path.name})") + return before + def sha256_file(path: Path) -> str: digest = hashlib.sha256() @@ -315,17 +324,20 @@ def __init__(self) -> None: f"Install {ENGINE_PKG_PIN}." ) from exc self._proc = PySaxonProcessor(license=False) - try: - # B additional: HTTP-only extra. Never empty (breaks stylesheet_file=). - # Does not block file SYSTEM; C covers that. B never replaces C. - self._proc.set_configuration_property(ALLOWED_PROTOCOLS_FEATURE, "file") - except Exception: - pass + if SAXON_ALLOWED_PROTOCOLS_VALUE != "file": + raise ConfigError( + "Saxon allowedProtocols must be 'file' " + "(empty string breaks file: XSLT)" + ) + self._proc.set_configuration_property( + SAXON_ALLOWED_PROTOCOLS_FEATURE, + SAXON_ALLOWED_PROTOCOLS_VALUE, + ) self._xslt = self._proc.new_xslt30_processor() self._compiled: dict[str, Any] = {} def transform(self, xml_path: Path, xslt_path: Path) -> str: - refuse_invoice_dtd(xml_path) + gate_invoice_path(xml_path) key = str(xslt_path) executable = self._compiled.get(key) if executable is None: @@ -515,7 +527,7 @@ def validate_files( annotations: list[str] = [] try: for path in files: - data = refuse_invoice_dtd(path) + gate_invoice_path(path) resolved_syntax = resolve_syntax(path, syntax) xslt_info = xslt[resolved_syntax] svrl = engine.transform(path, Path(xslt_info["path"])) @@ -526,7 +538,7 @@ def validate_files( "failed_assert_count": len(failed), "failed_assert_ids": ids, "path": rel, - "sha256": hashlib.sha256(data).hexdigest(), + "sha256": sha256_file(path), "syntax": resolved_syntax, "verdict": "fail" if ids else "pass", "xslt": xslt_info["logical"], diff --git a/testdata/hostile/README.md b/testdata/hostile/README.md deleted file mode 100644 index 9e74e9d..0000000 --- a/testdata/hostile/README.md +++ /dev/null @@ -1,8 +0,0 @@ -# Hostile DTD fixtures - -Used only by `tests/test_dtd_gate.py` (temporary files with local canary -paths and a loopback HTTP server). Not globbed by selftest official or -mutant steps, nor by the Action jobs (`testdata/official/*.xml`, -`testdata/mutants/*.xml`). - -Do not add DOCTYPE samples under `testdata/official/` or `testdata/mutants/`. diff --git a/tests/test_dtd_gate.py b/tests/test_dtd_gate.py index d05b63e..c6ff82b 100644 --- a/tests/test_dtd_gate.py +++ b/tests/test_dtd_gate.py @@ -1,12 +1,14 @@ #!/usr/bin/env python3 -"""C gate: refuse any DOCTYPE before Saxon. Local-runnable, not Action globs.""" +"""C gate: stdlib expat fail-closed on DOCTYPE. B is extra HTTP block only.""" from __future__ import annotations import hashlib import http.server +import inspect import os import socketserver +import subprocess import sys import tempfile import threading @@ -20,282 +22,421 @@ import validate # noqa: E402 -CII_MIN = ( - '\n' - f'x' - "\n" -) -CANARY_TOKEN = "CANARY_DTD_GATE_9f2c1a70" +CANARY = "CANARY_DTD_GATE_TOKEN_7c1e9b42" +try: + from saxonche import PySaxonProcessor -class ReuseTCPServer(socketserver.TCPServer): - allow_reuse_address = True - + HAS_SAXON = True +except ImportError: + HAS_SAXON = False -def _start_hit_server(hits: list[str]) -> tuple[ReuseTCPServer, str]: - class Handler(http.server.BaseHTTPRequestHandler): - def do_GET(self) -> None: - hits.append(self.path) - self.send_response(200) - self.send_header("Content-Type", "application/xml") - self.end_headers() - self.wfile.write(b"\n") - def log_message(self, *_args) -> None: - return None - - httpd = ReuseTCPServer(("127.0.0.1", 0), Handler) - thread = threading.Thread(target=httpd.serve_forever, daemon=True) - thread.start() - host, port = httpd.server_address[:2] - return httpd, f"http://{host}:{port}" +class ThreadingHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer): + allow_reuse_address = True + daemon_threads = True -class FakeEngine: - def __init__(self) -> None: - self.calls: list[Path] = [] +class HitHandler(http.server.BaseHTTPRequestHandler): + hits: list[str] - def transform(self, xml_path: Path, xslt_path: Path) -> str: - self.calls.append(Path(xml_path)) - return '' + def do_GET(self): + self.hits.append(self.path) + body = b"HTTP_PROBE_BODY_OK\n" + self.send_response(200) + self.send_header("Content-Type", "application/xml") + self.send_header("Content-Length", str(len(body))) + self.send_header("Connection", "close") + self.end_headers() + self.wfile.write(body) - def close(self) -> None: + def log_message(self, *args): return None -def _run_validate(path: Path, engine: FakeEngine) -> dict: - tmp_root = path.parent +def start_http() -> tuple[ThreadingHTTPServer, list[str]]: + hits: list[str] = [] - def fake_load_xslt(root: Path, version: str) -> dict: - dummy = tmp_root / "dummy.xslt" - dummy.write_text("\n") - return { - "CII": {"path": dummy, "sha256": "0", "logical": "dummy"}, - "UBL": {"path": dummy, "sha256": "0", "logical": "dummy"}, - } + class BoundHandler(HitHandler): + pass - orig_engine = validate.SaxonEngine - orig_load = validate.load_xslt - validate.SaxonEngine = lambda: engine # type: ignore[misc, assignment] - validate.load_xslt = fake_load_xslt # type: ignore[assignment] - try: - return validate.validate_files( - [path], - syntax="auto", - fail_on_raw="never", - version="1.3.16", - root=tmp_root, - cwd=tmp_root, - ) - finally: - validate.SaxonEngine = orig_engine - validate.load_xslt = orig_load - - -class GateDoctypeTests(unittest.TestCase): - def test_no_dtd_parse_ok(self) -> None: - data = CII_MIN.encode("utf-8") - before = hashlib.sha256(data).hexdigest() - validate.gate_doctype(data) - self.assertEqual(hashlib.sha256(data).hexdigest(), before) - - def test_internal_doctype_refused(self) -> None: + BoundHandler.hits = hits + httpd = ThreadingHTTPServer(("127.0.0.1", 0), BoundHandler) + thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread.start() + return httpd, hits + + +def sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def dump_xsl(dir_path: Path) -> Path: + path = dir_path / "dump.xsl" + path.write_bytes( + b""" + + + + +""" + ) + return path + + +def current_saxon_dump(xml_path: Path, xslt_path: Path) -> str: + """Unpatched SaxonC path: source_file= with no C gate and no B knob.""" + proc = PySaxonProcessor(license=False) + xslt = proc.new_xslt30_processor() + executable = xslt.compile_stylesheet(stylesheet_file=str(xslt_path)) + if executable is None: + raise RuntimeError("CURRENT compile failed") + out = executable.transform_to_string(source_file=str(xml_path)) + closer = getattr(proc, "release", None) + if callable(closer): + closer() + return "" if out is None else out + + +def current_saxon_dump_subprocess( + xml_path: Path, xslt_path: Path, *, timeout: float = 12.0 +) -> subprocess.CompletedProcess[str]: + """CURRENT Saxon in a child process so a hung HTTP fetch cannot stall tests.""" + script = ( + "from saxonche import PySaxonProcessor\n" + "import sys\n" + "xml_path, xslt_path = sys.argv[1], sys.argv[2]\n" + "proc = PySaxonProcessor(license=False)\n" + "xslt = proc.new_xslt30_processor()\n" + "exe = xslt.compile_stylesheet(stylesheet_file=xslt_path)\n" + "out = exe.transform_to_string(source_file=xml_path)\n" + "sys.stdout.write('' if out is None else out)\n" + ) + return subprocess.run( + [sys.executable, "-c", script, str(xml_path), str(xslt_path)], + capture_output=True, + text=True, + timeout=timeout, + ) + + +class GateMechanismTests(unittest.TestCase): + def test_gate_is_expat_start_doctype_not_regex_or_iterparse(self) -> None: + src = inspect.getsource(validate.gate_invoice_bytes) + module = Path(validate.__file__).read_text(encoding="utf-8") + self.assertIn("StartDoctypeDeclHandler", src) + self.assertIn("xml.parsers.expat", module) + self.assertNotIn("re.search", src) + self.assertNotIn("re.match", src) + self.assertNotIn("iterparse", src) + detect = inspect.getsource(validate.detect_syntax) + self.assertIn("ET.iterparse", detect) + self.assertNotIn("StartDoctypeDeclHandler", detect) + + def test_b_allowed_protocols_is_file_not_empty(self) -> None: + self.assertEqual(validate.SAXON_ALLOWED_PROTOCOLS_VALUE, "file") + self.assertNotEqual(validate.SAXON_ALLOWED_PROTOCOLS_VALUE, "") + src = inspect.getsource(validate.SaxonEngine.__init__) + self.assertIn("SAXON_ALLOWED_PROTOCOLS_FEATURE", src) + self.assertIn("SAXON_ALLOWED_PROTOCOLS_VALUE", src) + self.assertNotIn('""', src.split("set_configuration_property")[1][:400]) + + +class GateAdversarialTests(unittest.TestCase): + def test_no_dtd_control_parse_ok_bytes_unchanged(self) -> None: + data = b'\nNO_DTD_CONTROL\n' + before = sha256_bytes(data) + validate.gate_invoice_bytes(data, name="no-dtd.xml") + self.assertEqual(sha256_bytes(data), before) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "no-dtd.xml" + path.write_bytes(data) + digest = validate.gate_invoice_path(path) + self.assertEqual(digest, before) + self.assertEqual(sha256_bytes(path.read_bytes()), before) + + def test_internal_doctype_refused_bytes_unchanged(self) -> None: data = ( b'\n' - b' ]>\n' + b' ]>\n' b"&x;\n" ) - with self.assertRaises(validate.DtdRefused): - validate.gate_doctype(data) - - def test_system_file_refused_without_reading_canary(self) -> None: - with tempfile.TemporaryDirectory(prefix="dtd-file-") as td: - folder = Path(td) - canary = folder / "canary.txt" - canary.write_bytes(CANARY_TOKEN.encode("ascii")) - invoice = folder / "system-file.xml" - invoice.write_text( - '\n' - f' ]>\n' - "&ext;\n", - encoding="utf-8", - newline="\n", + before = sha256_bytes(data) + with self.assertRaises(validate.DtdRefused) as ctx: + validate.gate_invoice_bytes(data, name="internal-dtd.xml") + self.assertEqual(ctx.exception.doctype_name, "foo") + self.assertTrue(ctx.exception.has_internal_subset) + self.assertEqual(sha256_bytes(data), before) + self.assertNotIn(b"INTERNAL_ENTITY_VALUE", str(ctx.exception).encode("utf-8")) + + def test_system_file_refused_zero_canary_read(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + canary = base / "canary.txt" + canary.write_bytes((CANARY + "\n").encode("ascii")) + uri = canary.resolve().as_uri() + data = ( + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" ) - original_mode = canary.stat().st_mode - try: - os.chmod(canary, 0) - except OSError: - pass + xml_path = base / "system-file.xml" + xml_path.write_bytes(data) + before = sha256_bytes(xml_path.read_bytes()) + httpd, hits = start_http() try: - with self.assertRaises(validate.ConfigError) as ctx: - validate.refuse_invoice_dtd(invoice) + with self.assertRaises(validate.DtdRefused) as ctx: + validate.gate_invoice_path(xml_path) finally: - try: - os.chmod(canary, original_mode) - except OSError: - pass - message = str(ctx.exception) - self.assertIn("DOCTYPE", message) - self.assertNotIn("well-formed", message) - self.assertEqual(canary.read_bytes().decode("ascii"), CANARY_TOKEN) + httpd.shutdown() + self.assertEqual(hits, []) + self.assertNotIn(CANARY, str(ctx.exception)) + self.assertEqual(sha256_bytes(xml_path.read_bytes()), before) + self.assertEqual(canary.read_bytes(), (CANARY + "\n").encode("ascii")) def test_system_http_refused_zero_get(self) -> None: - hits: list[str] = [] - httpd, base = _start_hit_server(hits) + httpd, hits = start_http() try: + host, port = httpd.server_address[:2] + url = f"http://{host}:{port}/entity" data = ( - '\n' - f' ]>\n' - "&ext;\n" - ).encode("utf-8") + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" + ) + before = sha256_bytes(data) with self.assertRaises(validate.DtdRefused): - validate.gate_doctype(data) + validate.gate_invoice_bytes(data, name="system-http.xml") self.assertEqual(hits, []) + self.assertEqual(sha256_bytes(data), before) finally: httpd.shutdown() - httpd.server_close() - def test_public_doctype_refused_zero_get(self) -> None: - hits: list[str] = [] - httpd, base = _start_hit_server(hits) + def test_public_http_refused_zero_get(self) -> None: + httpd, hits = start_http() try: + host, port = httpd.server_address[:2] + url = f"http://{host}:{port}/public.dtd" data = ( - '\n' - f'\n' - "ok\n" - ).encode("utf-8") - with self.assertRaises(validate.DtdRefused): - validate.gate_doctype(data) + b'\n' + + ( + f'\n' + ).encode("ascii") + + b"ok\n" + ) + with self.assertRaises(validate.DtdRefused) as ctx: + validate.gate_invoice_bytes(data, name="public-http.xml") self.assertEqual(hits, []) + self.assertEqual(ctx.exception.pubid, "-//GATE//DTD Foo 1.0//EN") finally: httpd.shutdown() - httpd.server_close() - - def test_official_fixtures_have_no_doctype(self) -> None: - official = ROOT / "testdata" / "official" - mutants = ROOT / "testdata" / "mutants" - xml_files = sorted(official.glob("*.xml")) + sorted(mutants.glob("*.xml")) - if len(xml_files) != 20: - self.skipTest("official/mutant fixtures not present") - for path in xml_files: - raw = path.read_bytes() - validate.gate_doctype(raw) - self.assertNotIn(b" None: - with tempfile.TemporaryDirectory(prefix="dtd-ok-") as td: - path = Path(td) / "control.xml" - payload = CII_MIN.encode("utf-8") - path.write_bytes(payload) - engine = FakeEngine() - result = _run_validate(path, engine) - self.assertEqual(len(engine.calls), 1) - self.assertEqual(engine.calls[0].resolve(), path.resolve()) - self.assertEqual( - result["payload"]["files"][0]["sha256"], - hashlib.sha256(payload).hexdigest(), - ) - self.assertEqual(path.read_bytes(), payload) - - def test_internal_doctype_never_reaches_saxon(self) -> None: - with tempfile.TemporaryDirectory(prefix="dtd-int-") as td: - path = Path(td) / "internal.xml" - path.write_text( - '\n' - ' ]>\n' - f'&x;' - "\n", - encoding="utf-8", - newline="\n", - ) - engine = FakeEngine() - with self.assertRaises(validate.ConfigError) as ctx: - _run_validate(path, engine) - self.assertIn("DOCTYPE", str(ctx.exception)) - self.assertEqual(engine.calls, []) - - def test_system_file_never_reads_canary_never_reaches_saxon(self) -> None: - with tempfile.TemporaryDirectory(prefix="dtd-sysfile-") as td: - folder = Path(td) - canary = folder / "canary.txt" - canary.write_bytes(CANARY_TOKEN.encode("ascii")) - path = folder / "system-file.xml" - path.write_text( - '\n' - f'\n' - f'x' - "\n", - encoding="utf-8", - newline="\n", + + def test_external_dtd_system_file_refused(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + dtd = Path(tmp) / "ext.dtd" + dtd.write_bytes(b"\n") + uri = dtd.resolve().as_uri() + data = ( + b'\n' + + f'\n'.encode("ascii") + + b"ok\n" ) - original_mode = canary.stat().st_mode - try: - os.chmod(canary, 0) - except OSError: - pass - engine = FakeEngine() - try: - with self.assertRaises(validate.ConfigError) as ctx: - _run_validate(path, engine) - finally: - try: - os.chmod(canary, original_mode) - except OSError: - pass - self.assertIn("DOCTYPE", str(ctx.exception)) - self.assertEqual(engine.calls, []) - self.assertEqual(canary.read_bytes().decode("ascii"), CANARY_TOKEN) + with self.assertRaises(validate.DtdRefused) as ctx: + validate.gate_invoice_bytes(data, name="external-dtd-file.xml") + self.assertFalse(ctx.exception.has_internal_subset) + self.assertIsNotNone(ctx.exception.sysid) + + def test_et_iterparse_is_not_the_gate(self) -> None: + data = ( + b'\n' + b' ]>\n' + b"&x;\n" + ) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "internal-dtd.xml" + path.write_bytes(data) + import xml.etree.ElementTree as ET + + root_tag = None + for _event, elem in ET.iterparse(path, events=("start",)): + root_tag = elem.tag + break + self.assertEqual(root_tag, "foo") + with self.assertRaises(validate.DtdRefused): + validate.gate_invoice_path(path) - def test_system_http_zero_get_never_reaches_saxon(self) -> None: - hits: list[str] = [] - httpd, base = _start_hit_server(hits) + +@unittest.skipUnless(HAS_SAXON, "saxonche not installed") +class BeforeAfterSaxonTests(unittest.TestCase): + def test_current_follows_system_http_after_c_zero_get(self) -> None: + httpd, hits = start_http() try: - with tempfile.TemporaryDirectory(prefix="dtd-http-") as td: - path = Path(td) / "system-http.xml" - path.write_text( - '\n' - f'\n' - f'x' - "\n", - encoding="utf-8", - newline="\n", + host, port = httpd.server_address[:2] + url = f"http://{host}:{port}/entity" + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + xml_path = base / "system-http.xml" + xml_path.write_bytes( + b'\n' + + f' ]>\n'.encode( + "ascii" + ) + + b"&ext;\n" ) - engine = FakeEngine() - with self.assertRaises(validate.ConfigError) as ctx: - _run_validate(path, engine) - self.assertIn("DOCTYPE", str(ctx.exception)) - self.assertEqual(engine.calls, []) + xslt_path = dump_xsl(base) + hits.clear() + try: + current_saxon_dump_subprocess(xml_path, xslt_path, timeout=12.0) + except subprocess.TimeoutExpired: + pass + if sys.platform.startswith("linux"): + self.assertTrue(hits, msg="CURRENT must follow SYSTEM HTTP") + hits.clear() + with self.assertRaises(validate.DtdRefused): + validate.gate_invoice_path(xml_path) + self.assertEqual(hits, []) + hits.clear() + with self.assertRaises(validate.DtdRefused): + engine = validate.SaxonEngine() + try: + engine.transform(xml_path, xslt_path) + finally: + engine.close() self.assertEqual(hits, []) finally: httpd.shutdown() - httpd.server_close() - def test_public_zero_get_never_reaches_saxon(self) -> None: - hits: list[str] = [] - httpd, base = _start_hit_server(hits) + def test_current_follows_system_file_after_c_no_canary(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + canary = base / "canary.txt" + canary.write_bytes((CANARY + "\n").encode("ascii")) + uri = canary.resolve().as_uri() + xml_path = base / "system-file.xml" + xml_path.write_bytes( + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" + ) + xslt_path = dump_xsl(base) + current_out = current_saxon_dump(xml_path, xslt_path) + if sys.platform.startswith("linux"): + self.assertIn(CANARY, current_out) + with self.assertRaises(validate.DtdRefused) as ctx: + validate.gate_invoice_path(xml_path) + self.assertNotIn(CANARY, str(ctx.exception)) + with self.assertRaises(validate.DtdRefused): + engine = validate.SaxonEngine() + try: + engine.transform(xml_path, xslt_path) + finally: + engine.close() + + +@unittest.skipUnless(HAS_SAXON, "saxonche not installed") +class OfficialBytesAndCrlfTests(unittest.TestCase): + def setUp(self) -> None: + os.environ["EN16931_ACTION_ROOT"] = str(ROOT) + + def test_official_file_same_sha_before_after_gate_and_report(self) -> None: + path = ROOT / "testdata" / "official" / "CII_example3.xml" + before = sha256_bytes(path.read_bytes()) + digest = validate.gate_invoice_path(path) + after = sha256_bytes(path.read_bytes()) + self.assertEqual(digest, before) + self.assertEqual(after, before) + result = validate.validate_files( + [path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + cwd=ROOT, + ) + row = result["payload"]["files"][0] + self.assertEqual(row["sha256"], before) + self.assertEqual(row["sha256"], validate.sha256_file(path)) + self.assertEqual(row["verdict"], "pass") + + def test_crlf_user_invoice_hashed_as_crlf(self) -> None: + raw = (ROOT / "testdata" / "official" / "CII_example3.xml").read_bytes() + self.assertNotIn(b"\r", raw) + crlf = raw.replace(b"\r\n", b"\n").replace(b"\n", b"\r\n") + self.assertIn(b"\r\n", crlf) + self.assertNotEqual(sha256_bytes(raw), sha256_bytes(crlf)) + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "user-crlf.xml" + path.write_bytes(crlf) + self.assertEqual(validate.sha256_file(path), sha256_bytes(crlf)) + digest = validate.gate_invoice_path(path) + self.assertEqual(digest, sha256_bytes(crlf)) + self.assertEqual(sha256_bytes(path.read_bytes()), sha256_bytes(crlf)) + result = validate.validate_files( + [path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + cwd=Path(tmp), + ) + row = result["payload"]["files"][0] + self.assertEqual(row["sha256"], sha256_bytes(crlf)) + self.assertNotEqual(row["sha256"], sha256_bytes(raw)) + + def test_validate_files_refuses_doctype_before_saxon(self) -> None: + httpd, hits = start_http() try: - with tempfile.TemporaryDirectory(prefix="dtd-pub-") as td: - path = Path(td) / "public.xml" - path.write_text( - '\n' - f'\n' - f'x' - "\n", - encoding="utf-8", - newline="\n", + host, port = httpd.server_address[:2] + url = f"http://{host}:{port}/entity" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "hostile.xml" + path.write_bytes( + b'\n' + + f' ]>\n'.encode( + "ascii" + ) + + b"&ext;\n" ) - engine = FakeEngine() - with self.assertRaises(validate.ConfigError) as ctx: - _run_validate(path, engine) - self.assertIn("DOCTYPE", str(ctx.exception)) - self.assertEqual(engine.calls, []) + with self.assertRaises(validate.DtdRefused): + validate.validate_files( + [path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + ) self.assertEqual(hits, []) finally: httpd.shutdown() - httpd.server_close() + + def test_b_does_not_break_stylesheet_file_and_source_file(self) -> None: + path = ROOT / "testdata" / "official" / "CII_example3.xml" + engine = validate.SaxonEngine() + try: + xslt = ROOT / validate.CII_XSLT_REL + svrl = engine.transform(path, xslt) + finally: + engine.close() + self.assertIn("schematron-output", svrl) + + +class FifoNotOpenedTests(unittest.TestCase): + @unittest.skipIf(os.name == "nt", "named pipe is POSIX") + def test_system_fifo_not_opened(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + fifo = base / "canary.fifo" + os.mkfifo(fifo) + uri = fifo.resolve().as_uri() + data = ( + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" + ) + xml_path = base / "system-fifo.xml" + xml_path.write_bytes(data) + with self.assertRaises(validate.DtdRefused): + validate.gate_invoice_path(xml_path) if __name__ == "__main__": From a5f9960987734ef522a489d075990ab760e922c9 Mon Sep 17 00:00:00 2001 From: euphyra-dev Date: Sat, 29 Aug 2026 18:05:37 +0000 Subject: [PATCH 4/6] fix: hash files[].sha256 from DTD-gate bytes validate_files records files[].sha256 as hashlib.sha256(data) of the bytes returned by refuse_invoice_dtd (path.read_bytes, no rewrite). Do not hash via a second sha256_file(path) read. --- src/validate.py | 21 ++++++++++++--------- tests/test_cross_os_report.py | 13 ++++++++++++- 2 files changed, 24 insertions(+), 10 deletions(-) diff --git a/src/validate.py b/src/validate.py index 18af667..2433c76 100644 --- a/src/validate.py +++ b/src/validate.py @@ -100,18 +100,21 @@ def start_doctype(doctype_name, sysid, pubid, has_internal_subset): raise ConfigError(f"XML is not well-formed ({name}): {exc}") from exc +def refuse_invoice_dtd(path: Path, data: bytes | None = None) -> bytes: + """Read invoice bytes once, refuse any DOCTYPE, return the same bytes.""" + if data is None: + data = path.read_bytes() + gate_invoice_bytes(data, name=path.name) + return data + + def gate_invoice_path(path: Path) -> str: """Parse immutable path bytes; refuse DOCTYPE; return sha256 of those bytes. Saxon must be given this same path afterwards. The file is not rewritten. """ - data = path.read_bytes() - before = hashlib.sha256(data).hexdigest() - gate_invoice_bytes(data, name=path.name) - after = hashlib.sha256(path.read_bytes()).hexdigest() - if after != before: - raise EngineError(f"invoice bytes changed after DTD gate ({path.name})") - return before + data = refuse_invoice_dtd(path) + return hashlib.sha256(data).hexdigest() def sha256_file(path: Path) -> str: @@ -527,7 +530,7 @@ def validate_files( annotations: list[str] = [] try: for path in files: - gate_invoice_path(path) + data = refuse_invoice_dtd(path) resolved_syntax = resolve_syntax(path, syntax) xslt_info = xslt[resolved_syntax] svrl = engine.transform(path, Path(xslt_info["path"])) @@ -538,7 +541,7 @@ def validate_files( "failed_assert_count": len(failed), "failed_assert_ids": ids, "path": rel, - "sha256": sha256_file(path), + "sha256": hashlib.sha256(data).hexdigest(), "syntax": resolved_syntax, "verdict": "fail" if ids else "pass", "xslt": xslt_info["logical"], diff --git a/tests/test_cross_os_report.py b/tests/test_cross_os_report.py index 1e141ea..0d56134 100644 --- a/tests/test_cross_os_report.py +++ b/tests/test_cross_os_report.py @@ -173,8 +173,19 @@ def test_sha256_file_does_not_strip_crlf(self) -> None: def test_validate_files_hashes_via_sha256_file(self) -> None: source = inspect.getsource(validate.validate_files) - self.assertIn("sha256_file(path)", source) + self.assertIn("data = refuse_invoice_dtd(path)", source) + self.assertIn("hashlib.sha256(data).hexdigest()", source) + self.assertNotIn("sha256_file(path)", source) self.assertNotIn("replace(b", source) + self.assertNotIn("tostring", source) + self.assertNotIn("write_bytes", source) + self.assertNotIn("write_text", source) + refuse_src = inspect.getsource(validate.refuse_invoice_dtd) + self.assertIn("path.read_bytes()", refuse_src) + self.assertNotIn("tostring", refuse_src) + self.assertNotIn("serialize", refuse_src) + self.assertNotIn("write_bytes", refuse_src) + self.assertNotIn("write_text", refuse_src) module = Path(validate.__file__).read_text(encoding="utf-8") self.assertNotIn(".replace(b\"\\r\\n\"", module) self.assertNotIn(".replace('\\r\\n'", module) From 62bf4f44d0704a785d1091036941ae11777b8b99 Mon Sep 17 00:00:00 2001 From: euphyra-dev Date: Sat, 29 Aug 2026 18:30:58 +0000 Subject: [PATCH 5/6] fix: transform a private snapshot of gated invoice bytes Read user bytes once. Gate, sha256, syntax, and Saxon all use that snapshot. SaxonC 13 has no byte-backed source API, so source_file= is a private temp copy, never the original path. B remains allowedProtocols=file. --- README.md | 5 +- docs/LIMITS.md | 12 +- src/validate.py | 85 +++++++++++-- tests/test_cross_os_report.py | 1 + tests/test_dtd_gate.py | 219 ++++++++++++++++++++++++++++++++++ 5 files changed, 303 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index f99f0b7..33a1ddc 100644 --- a/README.md +++ b/README.md @@ -154,8 +154,9 @@ canonical (`sort_keys`, no timestamps, relative paths). Two consecutive runs of the same files must produce the same `report-sha256`. Invoices that contain a DOCTYPE are refused before XSLT (stdlib expat). -The invoice file is not rewritten; `files[].sha256` is the raw file -bytes. See [`docs/LIMITS.md`](docs/LIMITS.md). +User bytes are read once; `files[].sha256` is those raw bytes; Saxon +transforms a private snapshot of the same bytes. See +[`docs/LIMITS.md`](docs/LIMITS.md). `auto` maps: diff --git a/docs/LIMITS.md b/docs/LIMITS.md index 68adbb5..1f102f7 100644 --- a/docs/LIMITS.md +++ b/docs/LIMITS.md @@ -22,11 +22,13 @@ not a published maximum. ## DTD / protocol confine (C + B) C (mandatory): stdlib `xml.parsers.expat` `StartDoctypeDeclHandler` refuses -any DOCTYPE before Saxon. The invoice file is not rewritten; Saxon -`source_file=` uses the same path and the same bytes that passed the -gate. `files[].sha256` is the SHA256 of those raw bytes. User invoices -are not newline-normalized. `ET.iterparse` is syntax detection only; it -is not this gate. +any DOCTYPE before Saxon. User invoice bytes are read once into a snapshot. +The gate, `files[].sha256`, syntax detection, and Saxon all use that same +snapshot (no second read of the original path). SaxonC 13 has no byte-backed +source API, so Saxon `source_file=` is a private temp copy of those bytes, +never the original user path. The snapshot is not rewritten or +newline-normalized. `ET.iterparse` is syntax detection only; it is not this +gate. B (additional): SaxonC 13 `set_configuration_property('http://saxon.sf.net/feature/allowedProtocols', diff --git a/src/validate.py b/src/validate.py index 2433c76..69c7a9d 100644 --- a/src/validate.py +++ b/src/validate.py @@ -8,16 +8,19 @@ from __future__ import annotations import argparse +import contextlib import glob import hashlib +import io import json import os import re import sys +import tempfile import xml.etree.ElementTree as ET import xml.parsers.expat from pathlib import Path -from typing import Any +from typing import Any, Iterator SUPPORTED_VERSION = "1.3.16" ENGINE_NAME = "SaxonC-HE 13.0" @@ -109,14 +112,59 @@ def refuse_invoice_dtd(path: Path, data: bytes | None = None) -> bytes: def gate_invoice_path(path: Path) -> str: - """Parse immutable path bytes; refuse DOCTYPE; return sha256 of those bytes. + """Parse path bytes; refuse DOCTYPE; return sha256 of those bytes. - Saxon must be given this same path afterwards. The file is not rewritten. + The file is not rewritten. Callers that transform must hand Saxon a + private snapshot of these bytes, not this user path. """ data = refuse_invoice_dtd(path) return hashlib.sha256(data).hexdigest() +def _privatize_tmpdir(tmpdir: str) -> None: + """0o700 on POSIX; Windows mkdtemp is already under the user temp dir.""" + if os.name != "nt": + os.chmod(tmpdir, 0o700) + + +@contextlib.contextmanager +def private_snapshot_file(data: bytes) -> Iterator[Path]: + """Write snapshot bytes to a private temp file; yield that path; always delete. + + SaxonC 13 has no byte-backed source API (parse_xml xml_text is str only). + The original user path is never yielded. Bytes are copied as-is (no + decode, no re-serialize, LF/CRLF preserved). + """ + tmpdir = tempfile.mkdtemp(prefix="ve-snap-") + snap = None + fd = None + try: + _privatize_tmpdir(tmpdir) + fd, snap = tempfile.mkstemp(prefix="inv-", suffix=".xml", dir=tmpdir) + view = memoryview(data) + offset = 0 + while offset < len(data): + offset += os.write(fd, view[offset:]) + os.close(fd) + fd = None + yield Path(snap) + finally: + if fd is not None: + try: + os.close(fd) + except OSError: + pass + if snap is not None: + try: + os.unlink(snap) + except OSError: + pass + try: + os.rmdir(tmpdir) + except OSError: + pass + + def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: @@ -200,10 +248,13 @@ def display_path(path: Path, cwd: Path | None = None) -> str: return path.name -def detect_syntax(path: Path) -> str: +def detect_syntax(path: Path, data: bytes | None = None) -> str: + """Detect CII/UBL from snapshot bytes. Does not re-read path when data is given.""" + if data is None: + data = path.read_bytes() root_tag = None try: - for _event, elem in ET.iterparse(path, events=("start",)): + for _event, elem in ET.iterparse(io.BytesIO(data), events=("start",)): root_tag = elem.tag break except ET.ParseError as exc: @@ -223,8 +274,8 @@ def detect_syntax(path: Path) -> str: ) -def resolve_syntax(path: Path, requested: str) -> str: - detected = detect_syntax(path) +def resolve_syntax(path: Path, requested: str, data: bytes | None = None) -> str: + detected = detect_syntax(path, data=data) mode = requested.lower() if mode == "auto": return detected @@ -340,7 +391,14 @@ def __init__(self) -> None: self._compiled: dict[str, Any] = {} def transform(self, xml_path: Path, xslt_path: Path) -> str: - gate_invoice_path(xml_path) + """Read the user path once, then transform a private snapshot of those bytes.""" + data = xml_path.read_bytes() + gate_invoice_bytes(data, name=xml_path.name) + return self.transform_snapshot(data, xslt_path, name=xml_path.name) + + def transform_snapshot(self, data: bytes, xslt_path: Path, *, name: str) -> str: + """Transform exactly these snapshot bytes. Never the original user path.""" + gate_invoice_bytes(data, name=name) key = str(xslt_path) executable = self._compiled.get(key) if executable is None: @@ -348,9 +406,10 @@ def transform(self, xml_path: Path, xslt_path: Path) -> str: if executable is None: raise EngineError(f"XSLT compile failed: {xslt_path.name}") self._compiled[key] = executable - svrl = executable.transform_to_string(source_file=str(xml_path)) + with private_snapshot_file(data) as snap: + svrl = executable.transform_to_string(source_file=str(snap)) if svrl is None: - raise EngineError(f"XSLT produced no SVRL: {xml_path.name}") + raise EngineError(f"XSLT produced no SVRL: {name}") return svrl def close(self) -> None: @@ -531,9 +590,11 @@ def validate_files( try: for path in files: data = refuse_invoice_dtd(path) - resolved_syntax = resolve_syntax(path, syntax) + resolved_syntax = resolve_syntax(path, syntax, data=data) xslt_info = xslt[resolved_syntax] - svrl = engine.transform(path, Path(xslt_info["path"])) + svrl = engine.transform_snapshot( + data, Path(xslt_info["path"]), name=path.name + ) failed = parse_failed_asserts(svrl) ids = [item["id"] for item in failed] rel = display_path(path, cwd=cwd) diff --git a/tests/test_cross_os_report.py b/tests/test_cross_os_report.py index 0d56134..ea8095d 100644 --- a/tests/test_cross_os_report.py +++ b/tests/test_cross_os_report.py @@ -175,6 +175,7 @@ def test_validate_files_hashes_via_sha256_file(self) -> None: source = inspect.getsource(validate.validate_files) self.assertIn("data = refuse_invoice_dtd(path)", source) self.assertIn("hashlib.sha256(data).hexdigest()", source) + self.assertIn("transform_snapshot", source) self.assertNotIn("sha256_file(path)", source) self.assertNotIn("replace(b", source) self.assertNotIn("tostring", source) diff --git a/tests/test_dtd_gate.py b/tests/test_dtd_gate.py index c6ff82b..5587216 100644 --- a/tests/test_dtd_gate.py +++ b/tests/test_dtd_gate.py @@ -3,6 +3,7 @@ from __future__ import annotations +import contextlib import hashlib import http.server import inspect @@ -131,7 +132,18 @@ def test_gate_is_expat_start_doctype_not_regex_or_iterparse(self) -> None: self.assertNotIn("iterparse", src) detect = inspect.getsource(validate.detect_syntax) self.assertIn("ET.iterparse", detect) + self.assertIn("io.BytesIO", detect) self.assertNotIn("StartDoctypeDeclHandler", detect) + snap_src = inspect.getsource(validate.SaxonEngine.transform_snapshot) + self.assertIn("private_snapshot_file", snap_src) + self.assertIn("source_file=str(snap)", snap_src) + self.assertNotIn("xml_path", snap_src) + t_src = inspect.getsource(validate.SaxonEngine.transform) + self.assertIn("transform_snapshot", t_src) + self.assertNotIn("source_file=", t_src) + vf_src = inspect.getsource(validate.validate_files) + self.assertIn("transform_snapshot", vf_src) + self.assertNotIn("engine.transform(path", vf_src) def test_b_allowed_protocols_is_file_not_empty(self) -> None: self.assertEqual(validate.SAXON_ALLOWED_PROTOCOLS_VALUE, "file") @@ -420,6 +432,213 @@ def test_b_does_not_break_stylesheet_file_and_source_file(self) -> None: self.assertIn("schematron-output", svrl) + + +class PrivateSnapshotFileTests(unittest.TestCase): + def test_snapshot_preserves_crlf_and_is_not_user_path(self) -> None: + data = b"\r\n\n" + lf = b"\n\n" + self.assertNotEqual(sha256_bytes(data), sha256_bytes(lf)) + with tempfile.TemporaryDirectory() as tmp: + user = Path(tmp) / "user.xml" + user.write_bytes(data) + yielded: Path | None = None + with validate.private_snapshot_file(data) as snap: + yielded = snap + self.assertEqual(snap.read_bytes(), data) + self.assertNotEqual(snap.resolve(), user.resolve()) + if os.name != "nt": + mode = os.stat(snap.parent).st_mode & 0o777 + self.assertEqual(mode, 0o700) + assert yielded is not None + self.assertFalse(yielded.exists()) + self.assertFalse(yielded.parent.exists()) + self.assertEqual(user.read_bytes(), data) + + +def anti_pattern_transform_user_path( + xml_path: Path, + xslt_path: Path, + *, + after_gate=None, +) -> str: + """ANTI-PATTERN: gate the user path, then source_file= that same path. + + Same-path is not same-bytes: a replace after the gate is visible to Saxon. + Production must transform a private snapshot instead. Kept as the + documented old design for the FAIL proof below. + """ + validate.gate_invoice_path(xml_path) + if after_gate is not None: + after_gate(xml_path) + return current_saxon_dump(xml_path, xslt_path) + + +@unittest.skipUnless(HAS_SAXON, "saxonche not installed") +class SameBytesToctouTests(unittest.TestCase): + def setUp(self) -> None: + os.environ["EN16931_ACTION_ROOT"] = str(ROOT) + + def test_old_design_consumes_replacement_fix_transforms_snapshot(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + canary = base / "canary.txt" + canary.write_bytes((CANARY + "\n").encode("ascii")) + uri = canary.resolve().as_uri() + hostile = ( + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" + ) + mark = "BENIGN_SNAPSHOT_MARK" + benign = ( + b'\n' + + f"{mark}\n".encode("ascii") + ) + xml_path = base / "invoice.xml" + xml_path.write_bytes(benign) + xslt_path = dump_xsl(base) + + # Old design FAIL proof: last gate on original path, then replace. + self.assertIn("ANTI-PATTERN", anti_pattern_transform_user_path.__doc__ or "") + current_out = anti_pattern_transform_user_path( + xml_path, + xslt_path, + after_gate=lambda p: p.write_bytes(hostile), + ) + if sys.platform.startswith("linux"): + self.assertIn(CANARY, current_out) + self.assertNotIn(mark, current_out) + + # Restore canary + benign user file, then FIX path. + canary.write_bytes((CANARY + "\n").encode("ascii")) + xml_path.write_bytes(benign) + data = xml_path.read_bytes() + validate.gate_invoice_bytes(data, name=xml_path.name) + xml_path.write_bytes(hostile) + engine = validate.SaxonEngine() + try: + new_out = engine.transform_snapshot( + data, xslt_path, name=xml_path.name + ) + finally: + engine.close() + self.assertIn(mark, new_out) + self.assertNotIn(CANARY, new_out) + self.assertEqual(canary.read_bytes(), (CANARY + "\n").encode("ascii")) + self.assertEqual(xml_path.read_bytes(), hostile) + + def test_validate_files_hook_after_gate_never_transforms_replacement(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + base = Path(tmp) + canary = base / "canary.txt" + canary.write_bytes((CANARY + "\n").encode("ascii")) + uri = canary.resolve().as_uri() + hostile = ( + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" + ) + benign = (ROOT / "testdata" / "official" / "CII_example3.xml").read_bytes() + self.assertNotIn(b" bytes: + snapshot = original(path, data) + path.write_bytes(hostile) + return snapshot + + validate.refuse_invoice_dtd = refuse_then_replace # type: ignore[method-assign] + snap_paths: list[Path] = [] + real_snap = validate.private_snapshot_file + + @contextlib.contextmanager + def recording_snapshot(data: bytes): + with real_snap(data) as snap: + snap_paths.append(snap.resolve()) + self.assertEqual(snap.read_bytes(), benign) + self.assertEqual(xml_path.read_bytes(), hostile) + yield snap + + validate.private_snapshot_file = recording_snapshot # type: ignore[method-assign] + try: + result = validate.validate_files( + [xml_path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + cwd=base, + ) + finally: + validate.refuse_invoice_dtd = original + validate.private_snapshot_file = real_snap + + row = result["payload"]["files"][0] + self.assertEqual(row["sha256"], before) + self.assertEqual(row["verdict"], "pass") + self.assertEqual(canary.read_bytes(), (CANARY + "\n").encode("ascii")) + self.assertEqual(xml_path.read_bytes(), hostile) + self.assertTrue(snap_paths) + for snap in snap_paths: + self.assertNotEqual(snap, xml_path.resolve()) + self.assertFalse(snap.exists()) + + def test_validate_files_hook_http_zero_get(self) -> None: + httpd, hits = start_http() + try: + host, port = httpd.server_address[:2] + url = f"http://{host}:{port}/entity" + hostile = ( + b'\n' + + f' ]>\n'.encode("ascii") + + b"&ext;\n" + ) + benign = (ROOT / "testdata" / "official" / "CII_example3.xml").read_bytes() + with tempfile.TemporaryDirectory() as tmp: + xml_path = Path(tmp) / "invoice.xml" + xml_path.write_bytes(benign) + original = validate.refuse_invoice_dtd + + def refuse_then_replace(path: Path, data: bytes | None = None) -> bytes: + snapshot = original(path, data) + path.write_bytes(hostile) + return snapshot + + validate.refuse_invoice_dtd = refuse_then_replace # type: ignore[method-assign] + try: + result = validate.validate_files( + [xml_path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + ) + finally: + validate.refuse_invoice_dtd = original + self.assertEqual(hits, []) + self.assertEqual( + result["payload"]["files"][0]["sha256"], sha256_bytes(benign) + ) + self.assertEqual(result["payload"]["files"][0]["verdict"], "pass") + finally: + httpd.shutdown() + + +class DetectSyntaxSnapshotTests(unittest.TestCase): + def test_detect_syntax_uses_passed_bytes_not_path(self) -> None: + cii = (ROOT / "testdata" / "official" / "CII_example1.xml").read_bytes() + ubl = (ROOT / "testdata" / "official" / "ubl-tc434-creditnote1.xml").read_bytes() + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "mixed.xml" + path.write_bytes(ubl) + self.assertEqual(validate.detect_syntax(path, data=cii), "CII") + self.assertEqual(validate.detect_syntax(path), "UBL") + class FifoNotOpenedTests(unittest.TestCase): @unittest.skipIf(os.name == "nt", "named pipe is POSIX") def test_system_fifo_not_opened(self) -> None: From d0b702934bbbf4995b74489fbaa1a36a4b3036fd Mon Sep 17 00:00:00 2001 From: euphyra-dev Date: Sat, 29 Aug 2026 19:15:50 +0000 Subject: [PATCH 6/6] fix: verify private snapshot cleanup and prove B document() confine Unlink the snapshot only after Saxon finishes that source, then rmdir, retry, and Path.exists() both gone. Residue raises; no successful result. B: document() HTTP mutant gets; allowedProtocols=file is 0 GET fail-closed. --- docs/LIMITS.md | 10 +- src/validate.py | 150 +++++++++++++++-- tests/test_dtd_gate.py | 363 ++++++++++++++++++++++++++++++++++++++--- 3 files changed, 480 insertions(+), 43 deletions(-) diff --git a/docs/LIMITS.md b/docs/LIMITS.md index 1f102f7..97e7b38 100644 --- a/docs/LIMITS.md +++ b/docs/LIMITS.md @@ -27,8 +27,14 @@ The gate, `files[].sha256`, syntax detection, and Saxon all use that same snapshot (no second read of the original path). SaxonC 13 has no byte-backed source API, so Saxon `source_file=` is a private temp copy of those bytes, never the original user path. The snapshot is not rewritten or -newline-normalized. `ET.iterparse` is syntax detection only; it is not this -gate. +newline-normalized. After `transform_to_string` returns, processor refs +that can pin the snapshot are dropped (`gc.collect`, and `engine.close` +drops the compiled cache and processor). Cleanup then unlinks the snapshot +file, then removes the private directory, with bounded retry. Both paths +are checked with `Path.exists()`. If either remains, the engine raises and +does **not** return a successful validation result. That error does not +include the temp path or invoice bytes. `ET.iterparse` is syntax detection +only; it is not this gate. B (additional): SaxonC 13 `set_configuration_property('http://saxon.sf.net/feature/allowedProtocols', diff --git a/src/validate.py b/src/validate.py index 69c7a9d..636453c 100644 --- a/src/validate.py +++ b/src/validate.py @@ -9,6 +9,7 @@ import argparse import contextlib +import gc import glob import hashlib import io @@ -17,8 +18,10 @@ import re import sys import tempfile +import time import xml.etree.ElementTree as ET import xml.parsers.expat +from collections.abc import Callable from pathlib import Path from typing import Any, Iterator @@ -63,6 +66,16 @@ class EngineError(Exception): """XSLT engine error (exit 2).""" +class SnapshotCleanupError(EngineError): + """Private snapshot residue after transform (exit 2). + + Message is generic: no temp path, no invoice bytes. + """ + + def __init__(self) -> None: + super().__init__("private invoice snapshot could not be removed") + + class DtdRefused(ConfigError): """Any DOCTYPE is refused before the XSLT engine sees the file.""" @@ -127,13 +140,105 @@ def _privatize_tmpdir(tmpdir: str) -> None: os.chmod(tmpdir, 0o700) +SNAPSHOT_CLEANUP_ATTEMPTS = 8 +SNAPSHOT_CLEANUP_DELAY_SEC = 0.05 + + +def snapshot_cleanup_sleep(seconds: float) -> None: + """Sleep hook for snapshot cleanup retry. Tests replace this with a no-op.""" + if seconds: + time.sleep(seconds) + + +def remove_with_retry( + path: str, + *, + remover: Callable[[str], None], + attempts: int = SNAPSHOT_CLEANUP_ATTEMPTS, + delay: float = SNAPSHOT_CLEANUP_DELAY_SEC, + sleeper: Callable[[float], None] | None = None, +) -> None: + """Bounded remove. Tests pass attempts and a 0-delay sleeper.""" + if sleeper is None: + sleeper = snapshot_cleanup_sleep + n = max(1, int(attempts)) + for i in range(n): + try: + remover(path) + except FileNotFoundError: + return + except OSError: + pass + if not Path(path).exists(): + return + if i + 1 < n: + sleeper(delay) + + +def cleanup_private_snapshot( + snap: str | None, + tmpdir: str, + *, + attempts: int | None = None, + delay: float | None = None, + sleeper: Callable[[float], None] | None = None, + unlinker: Callable[[str], None] | None = None, + rmdirer: Callable[[str], None] | None = None, +) -> None: + """Unlink snapshot file, then rmdir private directory, then verify both gone. + + Raises SnapshotCleanupError on residue. Message has no path and no + invoice bytes. + """ + if attempts is None: + attempts = SNAPSHOT_CLEANUP_ATTEMPTS + if delay is None: + delay = SNAPSHOT_CLEANUP_DELAY_SEC + if unlinker is None: + unlinker = os.unlink + if rmdirer is None: + rmdirer = os.rmdir + if snap is not None: + remove_with_retry( + snap, + remover=unlinker, + attempts=attempts, + delay=delay, + sleeper=sleeper, + ) + remove_with_retry( + tmpdir, + remover=rmdirer, + attempts=attempts, + delay=delay, + sleeper=sleeper, + ) + snap_left = snap is not None and Path(snap).exists() + dir_left = Path(tmpdir).exists() + if snap_left or dir_left: + raise SnapshotCleanupError() + + @contextlib.contextmanager -def private_snapshot_file(data: bytes) -> Iterator[Path]: - """Write snapshot bytes to a private temp file; yield that path; always delete. +def private_snapshot_file( + data: bytes, + *, + attempts: int | None = None, + delay: float | None = None, + sleeper: Callable[[float], None] | None = None, + unlinker: Callable[[str], None] | None = None, + rmdirer: Callable[[str], None] | None = None, +) -> Iterator[Path]: + """Write snapshot bytes to a private temp file; yield that path. SaxonC 13 has no byte-backed source API (parse_xml xml_text is str only). The original user path is never yielded. Bytes are copied as-is (no decode, no re-serialize, LF/CRLF preserved). + + After the caller finishes (Saxon must have returned from + transform_to_string and dropped snapshot handles), cleanup unlinks the + file, then removes the directory, with bounded retry. Both paths are + then checked with Path.exists(). Residue raises SnapshotCleanupError. """ tmpdir = tempfile.mkdtemp(prefix="ve-snap-") snap = None @@ -154,15 +259,15 @@ def private_snapshot_file(data: bytes) -> Iterator[Path]: os.close(fd) except OSError: pass - if snap is not None: - try: - os.unlink(snap) - except OSError: - pass - try: - os.rmdir(tmpdir) - except OSError: - pass + cleanup_private_snapshot( + snap, + tmpdir, + attempts=attempts, + delay=delay, + sleeper=sleeper, + unlinker=unlinker, + rmdirer=rmdirer, + ) def sha256_file(path: Path) -> str: @@ -406,16 +511,31 @@ def transform_snapshot(self, data: bytes, xslt_path: Path, *, name: str) -> str: if executable is None: raise EngineError(f"XSLT compile failed: {xslt_path.name}") self._compiled[key] = executable + svrl = None with private_snapshot_file(data) as snap: - svrl = executable.transform_to_string(source_file=str(snap)) + try: + svrl = executable.transform_to_string(source_file=str(snap)) + finally: + # Saxon finished this source. Drop native wrappers so the + # snapshot file can be unlinked (needed on Windows). + gc.collect() if svrl is None: raise EngineError(f"XSLT produced no SVRL: {name}") return svrl def close(self) -> None: - closer = getattr(self._proc, "release", None) - if callable(closer): - closer() + compiled = getattr(self, "_compiled", None) + if compiled is not None: + compiled.clear() + self._xslt = None + proc = getattr(self, "_proc", None) + self._proc = None + if proc is not None: + closer = getattr(proc, "release", None) + if callable(closer): + closer() + del proc + gc.collect() def canonical_json(payload: Any) -> str: diff --git a/tests/test_dtd_gate.py b/tests/test_dtd_gate.py index 5587216..15e18e2 100644 --- a/tests/test_dtd_gate.py +++ b/tests/test_dtd_gate.py @@ -14,6 +14,7 @@ import tempfile import threading import unittest +import warnings from pathlib import Path ROOT = Path(__file__).resolve().parent.parent @@ -36,6 +37,7 @@ class ThreadingHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer): allow_reuse_address = True daemon_threads = True + block_on_close = False class HitHandler(http.server.BaseHTTPRequestHandler): @@ -55,7 +57,12 @@ def log_message(self, *args): return None -def start_http() -> tuple[ThreadingHTTPServer, list[str]]: +@contextlib.contextmanager +def http_probe_server(*, join_timeout: float = 5.0): + """One HTTP probe used by SYSTEM tests and the B document() probe. + + Start in a thread; always shutdown, server_close, join; assert stopped. + """ hits: list[str] = [] class BoundHandler(HitHandler): @@ -63,9 +70,20 @@ class BoundHandler(HitHandler): BoundHandler.hits = hits httpd = ThreadingHTTPServer(("127.0.0.1", 0), BoundHandler) - thread = threading.Thread(target=httpd.serve_forever, daemon=True) + thread = threading.Thread( + target=httpd.serve_forever, name="ve-http-probe", daemon=True + ) thread.start() - return httpd, hits + try: + yield httpd, hits + finally: + try: + httpd.shutdown() + finally: + httpd.server_close() + thread.join(timeout=join_timeout) + if thread.is_alive(): + raise AssertionError("http probe thread still running") def sha256_bytes(data: bytes) -> str: @@ -85,6 +103,21 @@ def dump_xsl(dir_path: Path) -> Path: return path +def document_http_xsl(dir_path: Path) -> Path: + """Test-only stylesheet: document($uri). Not under testdata/ official or mutants.""" + path = dir_path / "document-http.xsl" + path.write_bytes( + b""" + + + + + +""" + ) + return path + + def current_saxon_dump(xml_path: Path, xslt_path: Path) -> str: """Unpatched SaxonC path: source_file= with no C gate and no B knob.""" proc = PySaxonProcessor(license=False) @@ -153,6 +186,18 @@ def test_b_allowed_protocols_is_file_not_empty(self) -> None: self.assertIn("SAXON_ALLOWED_PROTOCOLS_VALUE", src) self.assertNotIn('""', src.split("set_configuration_property")[1][:400]) + def test_snapshot_cleanup_is_verified_not_swallowed(self) -> None: + src = inspect.getsource(validate.cleanup_private_snapshot) + self.assertIn("SnapshotCleanupError", src) + self.assertIn("exists()", src) + helper = inspect.getsource(validate.remove_with_retry) + self.assertIn("attempts", helper) + snap_src = inspect.getsource(validate.private_snapshot_file) + self.assertNotIn("always delete", snap_src) + self.assertIn("cleanup_private_snapshot", snap_src) + close_src = inspect.getsource(validate.SaxonEngine.close) + self.assertIn("gc.collect", close_src) + class GateAdversarialTests(unittest.TestCase): def test_no_dtd_control_parse_ok_bytes_unchanged(self) -> None: @@ -195,20 +240,16 @@ def test_system_file_refused_zero_canary_read(self) -> None: xml_path = base / "system-file.xml" xml_path.write_bytes(data) before = sha256_bytes(xml_path.read_bytes()) - httpd, hits = start_http() - try: + with http_probe_server() as (_httpd, hits): with self.assertRaises(validate.DtdRefused) as ctx: validate.gate_invoice_path(xml_path) - finally: - httpd.shutdown() self.assertEqual(hits, []) self.assertNotIn(CANARY, str(ctx.exception)) self.assertEqual(sha256_bytes(xml_path.read_bytes()), before) self.assertEqual(canary.read_bytes(), (CANARY + "\n").encode("ascii")) def test_system_http_refused_zero_get(self) -> None: - httpd, hits = start_http() - try: + with http_probe_server() as (httpd, hits): host, port = httpd.server_address[:2] url = f"http://{host}:{port}/entity" data = ( @@ -221,12 +262,9 @@ def test_system_http_refused_zero_get(self) -> None: validate.gate_invoice_bytes(data, name="system-http.xml") self.assertEqual(hits, []) self.assertEqual(sha256_bytes(data), before) - finally: - httpd.shutdown() def test_public_http_refused_zero_get(self) -> None: - httpd, hits = start_http() - try: + with http_probe_server() as (httpd, hits): host, port = httpd.server_address[:2] url = f"http://{host}:{port}/public.dtd" data = ( @@ -240,8 +278,6 @@ def test_public_http_refused_zero_get(self) -> None: validate.gate_invoice_bytes(data, name="public-http.xml") self.assertEqual(hits, []) self.assertEqual(ctx.exception.pubid, "-//GATE//DTD Foo 1.0//EN") - finally: - httpd.shutdown() def test_external_dtd_system_file_refused(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -281,8 +317,7 @@ def test_et_iterparse_is_not_the_gate(self) -> None: @unittest.skipUnless(HAS_SAXON, "saxonche not installed") class BeforeAfterSaxonTests(unittest.TestCase): def test_current_follows_system_http_after_c_zero_get(self) -> None: - httpd, hits = start_http() - try: + with http_probe_server() as (httpd, hits): host, port = httpd.server_address[:2] url = f"http://{host}:{port}/entity" with tempfile.TemporaryDirectory() as tmp: @@ -315,8 +350,6 @@ def test_current_follows_system_http_after_c_zero_get(self) -> None: finally: engine.close() self.assertEqual(hits, []) - finally: - httpd.shutdown() def test_current_follows_system_file_after_c_no_canary(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -396,8 +429,7 @@ def test_crlf_user_invoice_hashed_as_crlf(self) -> None: self.assertNotEqual(row["sha256"], sha256_bytes(raw)) def test_validate_files_refuses_doctype_before_saxon(self) -> None: - httpd, hits = start_http() - try: + with http_probe_server() as (httpd, hits): host, port = httpd.server_address[:2] url = f"http://{host}:{port}/entity" with tempfile.TemporaryDirectory() as tmp: @@ -418,8 +450,6 @@ def test_validate_files_refuses_doctype_before_saxon(self) -> None: root=ROOT, ) self.assertEqual(hits, []) - finally: - httpd.shutdown() def test_b_does_not_break_stylesheet_file_and_source_file(self) -> None: path = ROOT / "testdata" / "official" / "CII_example3.xml" @@ -445,6 +475,7 @@ def test_snapshot_preserves_crlf_and_is_not_user_path(self) -> None: yielded: Path | None = None with validate.private_snapshot_file(data) as snap: yielded = snap + self.assertTrue(snap.exists()) self.assertEqual(snap.read_bytes(), data) self.assertNotEqual(snap.resolve(), user.resolve()) if os.name != "nt": @@ -455,6 +486,113 @@ def test_snapshot_preserves_crlf_and_is_not_user_path(self) -> None: self.assertFalse(yielded.parent.exists()) self.assertEqual(user.read_bytes(), data) + def test_normal_cleanup_snapshot_gone_after_use(self) -> None: + data = b"SNAPSHOT_CLEANUP_CONTROL\n" + yielded: Path | None = None + parent: Path | None = None + with validate.private_snapshot_file(data) as snap: + yielded = snap + parent = snap.parent + self.assertTrue(snap.exists()) + self.assertTrue(parent.exists()) + self.assertEqual(snap.read_bytes(), data) + assert yielded is not None + assert parent is not None + self.assertFalse(yielded.exists()) + self.assertFalse(parent.exists()) + + def test_unlink_always_fails_raises_cleanup_error_no_paths_in_message(self) -> None: + secret = b"INVOICE_SECRET_BYTES_9f3a\n" + calls: list[str] = [] + sleeps: list[float] = [] + + def boom(path: str) -> None: + calls.append(path) + raise OSError("simulated unlink failure") + + yielded: Path | None = None + try: + with self.assertRaises(validate.SnapshotCleanupError) as ctx: + with validate.private_snapshot_file( + secret, + attempts=4, + delay=0.0, + sleeper=sleeps.append, + unlinker=boom, + ) as snap: + yielded = snap + self.assertTrue(snap.exists()) + assert yielded is not None + self.assertEqual(len(calls), 4) + self.assertTrue(sleeps) + self.assertTrue(all(s == 0.0 for s in sleeps)) + msg = str(ctx.exception) + self.assertNotIn(str(yielded), msg) + self.assertNotIn(str(yielded.parent), msg) + self.assertNotIn("INVOICE_SECRET_BYTES_9f3a", msg) + self.assertNotIn("ve-snap-", msg) + self.assertIsInstance(ctx.exception, validate.EngineError) + finally: + if yielded is not None: + try: + os.unlink(yielded) + except OSError: + pass + try: + os.rmdir(yielded.parent) + except OSError: + pass + + def test_unlink_fails_once_then_succeeds(self) -> None: + data = b"RETRY_THEN_OK\n" + real_unlink = os.unlink + state = {"n": 0} + + def flaky(path: str) -> None: + state["n"] += 1 + if state["n"] == 1: + raise OSError("first unlink fails") + real_unlink(path) + + sleeps: list[float] = [] + yielded: Path | None = None + with validate.private_snapshot_file( + data, + attempts=4, + delay=0.0, + sleeper=sleeps.append, + unlinker=flaky, + ) as snap: + yielded = snap + self.assertTrue(snap.exists()) + assert yielded is not None + self.assertGreaterEqual(state["n"], 2) + self.assertTrue(sleeps) + self.assertFalse(yielded.exists()) + self.assertFalse(yielded.parent.exists()) + + +class HttpProbeServerTests(unittest.TestCase): + def test_context_manager_stops_thread_no_resource_warning(self) -> None: + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", ResourceWarning) + with http_probe_server() as (httpd, hits): + host, port = httpd.server_address[:2] + import urllib.request + + with urllib.request.urlopen( + f"http://{host}:{port}/ping", timeout=2 + ) as resp: + resp.read() + self.assertEqual(hits, ["/ping"]) + import gc as _gc + + _gc.collect() + sock_warns = [ + w for w in caught if issubclass(w.category, ResourceWarning) + ] + self.assertEqual(sock_warns, []) + def anti_pattern_transform_user_path( xml_path: Path, @@ -589,8 +727,7 @@ def recording_snapshot(data: bytes): self.assertFalse(snap.exists()) def test_validate_files_hook_http_zero_get(self) -> None: - httpd, hits = start_http() - try: + with http_probe_server() as (httpd, hits): host, port = httpd.server_address[:2] url = f"http://{host}:{port}/entity" hostile = ( @@ -625,8 +762,182 @@ def refuse_then_replace(path: Path, data: bytes | None = None) -> bytes: result["payload"]["files"][0]["sha256"], sha256_bytes(benign) ) self.assertEqual(result["payload"]["files"][0]["verdict"], "pass") + + +@unittest.skipUnless(HAS_SAXON, "saxonche not installed") +class SnapshotCleanupValidationTests(unittest.TestCase): + def setUp(self) -> None: + os.environ["EN16931_ACTION_ROOT"] = str(ROOT) + + def test_unlink_always_fails_does_not_return_success(self) -> None: + path = ROOT / "testdata" / "official" / "CII_example3.xml" + secret = path.read_bytes() + self.assertNotIn(b" None: + calls.append(path_str) + raise OSError("simulated unlink failure") + + original = validate.private_snapshot_file + + @contextlib.contextmanager + def wrapped(data: bytes, **kwargs): + kwargs.setdefault("attempts", 3) + kwargs.setdefault("delay", 0.0) + kwargs.setdefault("sleeper", lambda _s: None) + kwargs["unlinker"] = boom + with original(data, **kwargs) as snap: + yield snap + + validate.private_snapshot_file = wrapped # type: ignore[method-assign] + try: + with self.assertRaises(validate.SnapshotCleanupError) as ctx: + validate.validate_files( + [path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + cwd=ROOT, + ) + self.assertGreaterEqual(len(calls), 3) + msg = str(ctx.exception) + for item in calls: + self.assertNotIn(item, msg) + self.assertNotIn("ve-snap-", msg) + snippet = secret[:40].decode("utf-8", "replace") + self.assertNotIn(snippet, msg) finally: - httpd.shutdown() + validate.private_snapshot_file = original + for item in calls: + try: + os.unlink(item) + except OSError: + pass + try: + os.rmdir(str(Path(item).parent)) + except OSError: + pass + + def test_unlink_retry_then_validation_completes(self) -> None: + path = ROOT / "testdata" / "official" / "CII_example3.xml" + real_unlink = os.unlink + state = {"n": 0} + + def flaky(path_str: str) -> None: + state["n"] += 1 + if state["n"] == 1: + raise OSError("first unlink fails") + real_unlink(path_str) + + original = validate.private_snapshot_file + + @contextlib.contextmanager + def wrapped(data: bytes, **kwargs): + kwargs.setdefault("attempts", 4) + kwargs.setdefault("delay", 0.0) + kwargs.setdefault("sleeper", lambda _s: None) + kwargs["unlinker"] = flaky + with original(data, **kwargs) as snap: + yield snap + + validate.private_snapshot_file = wrapped # type: ignore[method-assign] + try: + result = validate.validate_files( + [path], + syntax="auto", + fail_on_raw="failed-assert", + version="1.3.16", + root=ROOT, + cwd=ROOT, + ) + finally: + validate.private_snapshot_file = original + self.assertGreaterEqual(state["n"], 2) + self.assertEqual(result["payload"]["verdict"], "pass") + self.assertEqual( + result["payload"]["files"][0]["sha256"], sha256_bytes(path.read_bytes()) + ) + + +def _document_http_subprocess( + xml_path: Path, + xslt_path: Path, + url: str, + *, + allowed_protocols: str | None, + timeout: float = 12.0, +) -> subprocess.CompletedProcess[str]: + """Run document() against url. allowed_protocols=None is the B mutant.""" + script = ( + "from saxonche import PySaxonProcessor\n" + "import sys\n" + "xml_path, xslt_path, url, mode = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4]\n" + "proc = PySaxonProcessor(license=False)\n" + "if mode == 'file':\n" + " proc.set_configuration_property(\n" + " 'http://saxon.sf.net/feature/allowedProtocols', 'file'\n" + " )\n" + "xslt = proc.new_xslt30_processor()\n" + "exe = xslt.compile_stylesheet(stylesheet_file=xslt_path)\n" + "exe.set_parameter('uri', proc.make_string_value(url))\n" + "try:\n" + " out = exe.transform_to_string(source_file=xml_path)\n" + " sys.stdout.write('TRANSFORM_OK\\n')\n" + "except Exception:\n" + " sys.stdout.write('TRANSFORM_FAIL\\n')\n" + ) + mode = "file" if allowed_protocols == "file" else "none" + return subprocess.run( + [sys.executable, "-c", script, str(xml_path), str(xslt_path), url, mode], + capture_output=True, + text=True, + timeout=timeout, + ) + + +@unittest.skipUnless(HAS_SAXON, "saxonche not installed") +class AllowedProtocolsBTests(unittest.TestCase): + def test_document_http_mutant_gets_implementation_zero_get_fails_closed(self) -> None: + xml = ( + b'\n' + b"NO_DTD_DOCUMENT_PROBE\n" + ) + self.assertNotIn(b"