diff --git a/README.md b/README.md
index b2e95a6..33a1ddc 100644
--- a/README.md
+++ b/README.md
@@ -153,6 +153,11 @@ 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).
+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:
- `urn:un:unece:uncefact:data:standard:CrossIndustryInvoice:100` → CII
diff --git a/docs/LIMITS.md b/docs/LIMITS.md
index ffe257c..97e7b38 100644
--- a/docs/LIMITS.md
+++ b/docs/LIMITS.md
@@ -19,7 +19,28 @@ 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 (mandatory): stdlib `xml.parsers.expat` `StartDoctypeDeclHandler` refuses
+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. 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',
+'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 1e9584b..636453c 100644
--- a/src/validate.py
+++ b/src/validate.py
@@ -8,21 +8,33 @@
from __future__ import annotations
import argparse
+import contextlib
+import gc
import glob
import hashlib
+import io
import json
import os
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
+from typing import Any, Iterator
SUPPORTED_VERSION = "1.3.16"
ENGINE_NAME = "SaxonC-HE 13.0"
ENGINE_PKG_PIN = "saxonche==13.0.0"
ENGINE_PKG_PREFIX = "13.0"
+# 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"
UBL_CREDIT_NS = "urn:oasis:names:specification:ubl:schema:xsd:CreditNote-2"
@@ -54,6 +66,210 @@ 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."""
+
+ def __init__(
+ self,
+ 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})")
+
+
+def gate_invoice_bytes(data: bytes, *, name: str = "input") -> None:
+ """Fail-closed on any DOCTYPE via stdlib expat. Does not rewrite bytes.
+
+ ``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(
+ name, doctype_name, sysid, pubid, bool(has_internal_subset)
+ )
+
+ parser.StartDoctypeDeclHandler = start_doctype
+ try:
+ parser.Parse(data, True)
+ except DtdRefused:
+ raise
+ except xml.parsers.expat.ExpatError as exc:
+ 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 path bytes; refuse DOCTYPE; return sha256 of those bytes.
+
+ 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)
+
+
+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,
+ *,
+ 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
+ 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
+ cleanup_private_snapshot(
+ snap,
+ tmpdir,
+ attempts=attempts,
+ delay=delay,
+ sleeper=sleeper,
+ unlinker=unlinker,
+ rmdirer=rmdirer,
+ )
+
+
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
@@ -137,10 +353,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:
@@ -160,8 +379,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
@@ -264,10 +483,27 @@ def __init__(self) -> None:
f"Install {ENGINE_PKG_PIN}."
) from exc
self._proc = PySaxonProcessor(license=False)
+ 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:
+ """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:
@@ -275,15 +511,31 @@ 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))
+ svrl = None
+ with private_snapshot_file(data) as 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: {xml_path.name}")
+ 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:
@@ -457,9 +709,12 @@ def validate_files(
annotations: list[str] = []
try:
for path in files:
- resolved_syntax = resolve_syntax(path, syntax)
+ data = refuse_invoice_dtd(path)
+ 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)
@@ -467,7 +722,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/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
diff --git a/tests/test_cross_os_report.py b/tests/test_cross_os_report.py
index 1e141ea..ea8095d 100644
--- a/tests/test_cross_os_report.py
+++ b/tests/test_cross_os_report.py
@@ -173,8 +173,20 @@ 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.assertIn("transform_snapshot", 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)
diff --git a/tests/test_dtd_gate.py b/tests/test_dtd_gate.py
new file mode 100644
index 0000000..15e18e2
--- /dev/null
+++ b/tests/test_dtd_gate.py
@@ -0,0 +1,973 @@
+#!/usr/bin/env python3
+"""C gate: stdlib expat fail-closed on DOCTYPE. B is extra HTTP block only."""
+
+from __future__ import annotations
+
+import contextlib
+import hashlib
+import http.server
+import inspect
+import os
+import socketserver
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+import warnings
+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
+
+CANARY = "CANARY_DTD_GATE_TOKEN_7c1e9b42"
+
+try:
+ from saxonche import PySaxonProcessor
+
+ HAS_SAXON = True
+except ImportError:
+ HAS_SAXON = False
+
+
+class ThreadingHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
+ allow_reuse_address = True
+ daemon_threads = True
+ block_on_close = False
+
+
+class HitHandler(http.server.BaseHTTPRequestHandler):
+ hits: list[str]
+
+ 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 log_message(self, *args):
+ return None
+
+
+@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):
+ pass
+
+ BoundHandler.hits = hits
+ httpd = ThreadingHTTPServer(("127.0.0.1", 0), BoundHandler)
+ thread = threading.Thread(
+ target=httpd.serve_forever, name="ve-http-probe", daemon=True
+ )
+ thread.start()
+ 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:
+ return hashlib.sha256(data).hexdigest()
+
+
+def dump_xsl(dir_path: Path) -> Path:
+ path = dir_path / "dump.xsl"
+ path.write_bytes(
+ b"""
+
+
+
+
+"""
+ )
+ 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)
+ 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.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")
+ 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])
+
+ 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:
+ 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"&x;\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"
+ )
+ xml_path = base / "system-file.xml"
+ xml_path.write_bytes(data)
+ before = sha256_bytes(xml_path.read_bytes())
+ with http_probe_server() as (_httpd, hits):
+ with self.assertRaises(validate.DtdRefused) as ctx:
+ validate.gate_invoice_path(xml_path)
+ 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:
+ with http_probe_server() as (httpd, hits):
+ host, port = httpd.server_address[:2]
+ url = f"http://{host}:{port}/entity"
+ data = (
+ b'\n'
+ + f' ]>\n'.encode("ascii")
+ + b"&ext;\n"
+ )
+ before = sha256_bytes(data)
+ with self.assertRaises(validate.DtdRefused):
+ validate.gate_invoice_bytes(data, name="system-http.xml")
+ self.assertEqual(hits, [])
+ self.assertEqual(sha256_bytes(data), before)
+
+ def test_public_http_refused_zero_get(self) -> None:
+ with http_probe_server() as (httpd, hits):
+ host, port = httpd.server_address[:2]
+ url = f"http://{host}:{port}/public.dtd"
+ 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")
+
+ 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"
+ )
+ 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)
+
+
+@unittest.skipUnless(HAS_SAXON, "saxonche not installed")
+class BeforeAfterSaxonTests(unittest.TestCase):
+ def test_current_follows_system_http_after_c_zero_get(self) -> None:
+ with http_probe_server() as (httpd, hits):
+ 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"
+ )
+ 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, [])
+
+ 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:
+ with http_probe_server() as (httpd, hits):
+ 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"
+ )
+ 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, [])
+
+ 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 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.assertTrue(snap.exists())
+ 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 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,
+ 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:
+ with http_probe_server() as (httpd, hits):
+ 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")
+
+
+@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:
+ 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" 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:
+ 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__":
+ unittest.main()