diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 185ad04..c27c594 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -18,3 +18,59 @@ jobs: - run: python tools/run_validation.py - run: python -m pip install --no-deps . - run: trace-gc --version + windows-fence: + runs-on: windows-latest + strategy: + fail-fast: false + matrix: + python-version: ['3.12', '3.13'] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - run: python -m pip install -r requirements.txt + - name: Run the nine native Windows admission-fence cases + shell: bash + run: | + set -o pipefail + python -m unittest discover -s tests -p "test_paper_pilot_phase_cutover_guarded_v2.py" -v 2>&1 | tee "$RUNNER_TEMP/windows-fence.log" + # A bare "OK" means every case executed; skips would print "OK (skipped=N)". + grep -Eq '^Ran 9 tests in ' "$RUNNER_TEMP/windows-fence.log" + grep -Eq '^OK[[:space:]]*$' "$RUNNER_TEMP/windows-fence.log" + - name: Run the fourteen Windows SID/SDDL trustee-alias regression cases + if: always() + shell: bash + run: | + set -o pipefail + python -m unittest discover -s tests -p "test_paper_pilot_phase_cutover_guarded_v2_trustees.py" -v 2>&1 | tee "$RUNNER_TEMP/windows-fence-trustees.log" + grep -Eq '^Ran 14 tests in ' "$RUNNER_TEMP/windows-fence-trustees.log" + grep -Eq '^OK[[:space:]]*$' "$RUNNER_TEMP/windows-fence-trustees.log" + - name: Run the thirteen owned-fixture ACL stabilization cases (nine contract, four native) + if: always() + shell: bash + run: | + set -o pipefail + python -m unittest discover -s tests -p "test_windows_fixture_acl.py" -v 2>&1 | tee "$RUNNER_TEMP/windows-fixture-acl.log" + grep -Eq '^Ran 13 tests in ' "$RUNNER_TEMP/windows-fixture-acl.log" + grep -Eq '^OK[[:space:]]*$' "$RUNNER_TEMP/windows-fixture-acl.log" + - name: Run the ten strict signed-head DACL fence cases (five contract, five native) + if: always() + shell: bash + run: | + set -o pipefail + python -m unittest discover -s tests -p "test_paper_pilot_phase_cutover_guarded_v2_strict_dacl.py" -v 2>&1 | tee "$RUNNER_TEMP/windows-fence-strict-dacl.log" + grep -Eq '^Ran 10 tests in ' "$RUNNER_TEMP/windows-fence-strict-dacl.log" + grep -Eq '^OK[[:space:]]*$' "$RUNNER_TEMP/windows-fence-strict-dacl.log" + - name: Run the eighteen diagnostics-tool sanitization, trace and cleanup cases + if: always() + shell: bash + run: | + set -o pipefail + python -m unittest discover -s tests -p "test_windows_fence_diagnostics.py" -v 2>&1 | tee "$RUNNER_TEMP/windows-fence-diagnostics-tests.log" + grep -Eq '^Ran 18 tests in ' "$RUNNER_TEMP/windows-fence-diagnostics-tests.log" + grep -Eq '^OK[[:space:]]*$' "$RUNNER_TEMP/windows-fence-diagnostics-tests.log" + - name: Sanitized Windows trustee serialization diagnostics + if: always() + continue-on-error: true + run: python tools/windows_fence_diagnostics.py diff --git a/MANIFEST.json b/MANIFEST.json index d6b97e6..359630a 100644 --- a/MANIFEST.json +++ b/MANIFEST.json @@ -3,12 +3,12 @@ "hash_algorithm": "SHA-256", "manifest_self_excluded": true, "source_only": true, - "file_count": 702, + "file_count": 709, "files": [ { "path": ".github/workflows/test.yml", - "bytes": 615, - "sha256": "ba3c6064c1fb430eda6f7fedf60561637be3daedfff4d1d533a61e5ef851ed2b" + "bytes": 3589, + "sha256": "9c147d3ff67e4bc298c4376bcef076390c1a86d25ee9333ddf1d01ff861f4b23" }, { "path": ".gitignore", @@ -2297,8 +2297,8 @@ }, { "path": "src/paper_pilot_phase_cutover_guarded_v2.py", - "bytes": 16677, - "sha256": "618cb280f69353aba783b68c34b3c8344a3d04dd4442fd805cf96b964f4d9af7" + "bytes": 19252, + "sha256": "eebb08bc51209e23a50d3b8e89afc84b834ac89d86c6a54e52f03f8ab06ee2d8" }, { "path": "src/paper_pilot_phase_cutover_guarded_v4.py", @@ -2632,8 +2632,23 @@ }, { "path": "tests/test_paper_pilot_phase_cutover_guarded_v2.py", - "bytes": 11173, - "sha256": "c1fab4e9adbe973c53d01fa2d0340559ffef0dc4d482e2b5b8bb68cfa9f43c33" + "bytes": 13300, + "sha256": "f7c72b60e3e0de9fc2b5dc14056ce88dd27d2db8a82157c07250f088049f30c7" + }, + { + "path": "tests/test_paper_pilot_phase_cutover_guarded_v2_simulated_acl.py", + "bytes": 46860, + "sha256": "eec4d28a522f207bcd28ccf5a1ffcaaaf526144d07c0bd551ef717e9ad6deb72" + }, + { + "path": "tests/test_paper_pilot_phase_cutover_guarded_v2_strict_dacl.py", + "bytes": 18172, + "sha256": "f714325983bec8b08ced75008a9e4c22cc59b76a0e3524ee720f621f8ca7b154" + }, + { + "path": "tests/test_paper_pilot_phase_cutover_guarded_v2_trustees.py", + "bytes": 18461, + "sha256": "13ac20e781fb6580636252871137743a834af56dbe2f031f2208550b94d70611" }, { "path": "tests/test_paper_pilot_phase_cutover_guarded_v4.py", @@ -2940,6 +2955,16 @@ "bytes": 7256, "sha256": "6164938f20b9a3aff039df703d8adc20d6ad5c0ead9f7eaba2ce5b458da5d227" }, + { + "path": "tests/test_windows_fence_diagnostics.py", + "bytes": 20559, + "sha256": "c99e5d60eb8deb0bb20f1a5b64d6888f7f7134841ba3016f6991266e9498bab4" + }, + { + "path": "tests/test_windows_fixture_acl.py", + "bytes": 11977, + "sha256": "438d68653cd88d60f4376f1bda7cb33e443d32059e0b86ef0f32b788ba5b4563" + }, { "path": "tools/build_schemas.py", "bytes": 12978, @@ -3015,6 +3040,16 @@ "bytes": 21568, "sha256": "d461d0c14365d97630ea216608cdbb095e3f8806e76a0a954ea2dfcdbbee655a" }, + { + "path": "tools/windows_fence_diagnostics.py", + "bytes": 21532, + "sha256": "9dabba98d9af2c92b77231e597194204c5a3918c214b319736a2d8ccad371471" + }, + { + "path": "tools/windows_fixture_acl.py", + "bytes": 4355, + "sha256": "e435337541de6216b0f1634da1a0e36a36915d2897414db80ba5b17cad7272f6" + }, { "path": "tools/windows_image_ocr.ps1", "bytes": 5451, diff --git a/src/paper_pilot_phase_cutover_guarded_v2.py b/src/paper_pilot_phase_cutover_guarded_v2.py index 8fc34c7..dd6a42c 100644 --- a/src/paper_pilot_phase_cutover_guarded_v2.py +++ b/src/paper_pilot_phase_cutover_guarded_v2.py @@ -12,7 +12,9 @@ from __future__ import annotations +import ctypes from contextlib import ExitStack +import functools import hashlib import json import os @@ -73,8 +75,71 @@ def _icacls(path: Path, *args: str) -> None: require(result.returncode == 0, "PHASE_FENCE", "Windows DACL update failed") +_NUMERIC_SID = re.compile(r"S-1-\d+(?:-\d+)+") + + +@functools.lru_cache(maxsize=None) +def _canonical_trustee(trustee: str) -> str: + """Return a trustee spelled exactly as Windows serializes it in a DACL string. + + Windows writes some numeric SIDs as SDDL aliases (for example a machine's + built-in Administrator, RID 500, is written ``LA``). Round-tripping through + the same Windows conversion that produced the DACL text makes the numeric + SID reviewed by the caller comparable to what the DACL reader returns, + without changing the serialized DACL or any receipt. + """ + if _NUMERIC_SID.fullmatch(trustee) is None: + return trustee + advapi = ctypes.WinDLL("advapi32", use_last_error=True) + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + kernel.LocalFree.argtypes = (ctypes.c_void_p,) + kernel.LocalFree.restype = ctypes.c_void_p + parse = advapi.ConvertStringSecurityDescriptorToSecurityDescriptorW + parse.argtypes = (ctypes.c_wchar_p, ctypes.c_uint32, + ctypes.POINTER(ctypes.c_void_p), ctypes.c_void_p) + parse.restype = ctypes.c_int + encode = advapi.ConvertSecurityDescriptorToStringSecurityDescriptorW + encode.argtypes = (ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint32, + ctypes.POINTER(ctypes.c_wchar_p), ctypes.c_void_p) + encode.restype = ctypes.c_int + descriptor = ctypes.c_void_p() + require(bool(parse("D:(D;;DCLC;;;" + trustee + ")", 1, ctypes.byref(descriptor), None)), + "PHASE_FENCE", "cannot parse SID trustee") + try: + value = ctypes.c_wchar_p() + require(bool(encode(descriptor, 1, 4, ctypes.byref(value), None)), + "PHASE_FENCE", "cannot encode SID trustee") + try: + text = value.value + finally: + kernel.LocalFree(ctypes.cast(value, ctypes.c_void_p)) + finally: + kernel.LocalFree(descriptor) + match = re.fullmatch(r"D:\(D;;DCLC;;;([^;()]+)\)", text) + require(match is not None, "PHASE_FENCE", "unexpected SID trustee encoding") + return match.group(1) + + +def _canonical_ace(ace: str) -> str: + # Only the trustee field is normalized; type, flags, rights, object fields + # and ACE order must still match exactly. + fields = ace[1:-1].split(";") + if len(fields) != 6: + return ace + fields[5] = _canonical_trustee(fields[5]) + return "(" + ";".join(fields) + ")" + + +def _aces(sddl: str) -> list[str]: + return [_canonical_ace(ace) for ace in re.findall(r"\([^)]*\)", sddl)] + + +def _has_ace(sddl: str, template: str, sid: str) -> bool: + return _canonical_ace(template.format(sid=sid)) in _aces(sddl) + + def _file_deny(path: Path, sid: str) -> bool: - return DENY_FILE_ACE.format(sid=sid) in directory_dacl_sddl(path) + return _has_ace(directory_dacl_sddl(path), DENY_FILE_ACE, sid) def _readable_pins(root: Path, expected: dict[str, str]) -> None: @@ -94,15 +159,13 @@ def _same_effective_dacl(before: str, after: str) -> bool: # protection (P) must still match exactly; no broadened access is accepted. before_flags = before.split("(", 1)[0].removeprefix("D:").replace("AI", "") after_flags = after.split("(", 1)[0].removeprefix("D:").replace("AI", "") - return (before_flags == after_flags and - re.findall(r"\([^)]*\)", before) == - re.findall(r"\([^)]*\)", after)) + return before_flags == after_flags and _aces(before) == _aces(after) def _only_admission_dacl_added(original: str, current: str, sid: str) -> bool: - old_aces = re.findall(r"\([^)]*\)", original) - now_aces = re.findall(r"\([^)]*\)", current) - deny = DENY_FILE_ACE.format(sid=sid) + old_aces = _aces(original) + now_aces = _aces(current) + deny = _canonical_ace(DENY_FILE_ACE.format(sid=sid)) if deny in now_aces: now_aces.remove(deny) old_flags = original.split("(", 1)[0].removeprefix("D:").replace("AI", "") @@ -111,14 +174,15 @@ def _only_admission_dacl_added(original: str, current: str, sid: str) -> bool: def _owner_deny_present(sddl: str, sid: str) -> bool: - return any(ace.startswith("(D;") and ace.endswith(";;;" + sid + ")") - for ace in re.findall(r"\([^)]*\)", sddl)) + trustee = _canonical_trustee(sid) + return any(ace.startswith("(D;") and ace.endswith(";;;" + trustee + ")") + for ace in _aces(sddl)) def _only_parent_admission_added(original: str, current: str, sid: str) -> bool: - old_aces = re.findall(r"\([^)]*\)", original) - now_aces = re.findall(r"\([^)]*\)", current) - deny = DENY_PARENT_ACE.format(sid=sid) + old_aces = _aces(original) + now_aces = _aces(current) + deny = _canonical_ace(DENY_PARENT_ACE.format(sid=sid)) if deny in now_aces: now_aces.remove(deny) old_flags = original.split("(", 1)[0].removeprefix("D:").replace("AI", "") @@ -205,7 +269,7 @@ def fence_closed_old_phase_guarded(*, old_root: str | Path, {name: bytes_digest(value.encode()) for name, value in original.items()} == expected_file_dacl_sha256, "PHASE_FENCE", "original Windows DACL identity changed") - require(all(DENY_FILE_ACE.format(sid=deny_sid) not in value + require(all(not _has_ace(value, DENY_FILE_ACE, deny_sid) for value in original.values()), "PHASE_FENCE", "unreviewed preexisting file deny") recorded = {"version": "paper-pilot-admission-barrier-v1", @@ -252,11 +316,11 @@ def fence_closed_old_phase_guarded(*, old_root: str | Path, require(_file_deny(path, deny_sid), "PHASE_FENCE", "new old-state write open not denied") parent_sddl = directory_dacl_sddl(old.parent) - if DENY_PARENT_ACE.format(sid=deny_sid) not in parent_sddl: + if not _has_ace(parent_sddl, DENY_PARENT_ACE, deny_sid): _icacls(old.parent, "/deny", "*" + deny_sid + ":(AD)") - require(DENY_PARENT_ACE.format(sid=deny_sid) in - directory_dacl_sddl(old.parent), "PHASE_FENCE", - "old state recreation remains possible") + require(_has_ace(directory_dacl_sddl(old.parent), DENY_PARENT_ACE, + deny_sid), "PHASE_FENCE", + "old state recreation remains possible") require(_only_parent_admission_added( recorded["original_parent_dacl_sddl"], directory_dacl_sddl(old.parent), deny_sid), @@ -285,9 +349,8 @@ def fence_closed_old_phase_guarded(*, old_root: str | Path, require(not _file_deny(path, deny_sid), "PHASE_FENCE", "archived readback still denied") _readable_pins(archived, expected_file_sha256) - require(DENY_PARENT_ACE.format(sid=deny_sid) in - directory_dacl_sddl(old.parent), "PHASE_FENCE", - "old fixed path admission reopened") + require(_has_ace(directory_dacl_sddl(old.parent), DENY_PARENT_ACE, deny_sid), + "PHASE_FENCE", "old fixed path admission reopened") require(_only_parent_admission_added( recorded["original_parent_dacl_sddl"], directory_dacl_sddl(old.parent), deny_sid), diff --git a/tests/test_paper_pilot_phase_cutover_guarded_v2.py b/tests/test_paper_pilot_phase_cutover_guarded_v2.py index 8d731e2..57fd61d 100644 --- a/tests/test_paper_pilot_phase_cutover_guarded_v2.py +++ b/tests/test_paper_pilot_phase_cutover_guarded_v2.py @@ -1,4 +1,8 @@ -"""Windows-only owned disposable tests for the historical admission barrier.""" +"""Windows-only owned disposable tests for the historical admission barrier. + +Every case builds its own fixture inside a unique temporary directory created +by this test, so it never depends on (or creates) any historical task state. +""" from __future__ import annotations import hashlib @@ -14,10 +18,11 @@ from src import paper_pilot_phase_cutover_guarded_v2 as guarded from src.paper_pilot_phase_cutover import _current_sid +from tools import windows_fixture_acl from trace_gc.budget import RunBudget -OWNED = Path(r"D:\codex-task-runs\issue90-real-historical-transition-disabled-plan-20260930-01\fence-owned-tests") +TEMP_PREFIX = "jraphyte-fence-" LIMITS = {"retrieval_requests": 1, "model_calls": 1, "retries": 0, "solver_expansions": 1, "review_actions": 1, "request_bytes": 1000} @@ -38,11 +43,14 @@ def test_module_load_does_not_require_msvcrt(self): def setUp(self): if os.name != "nt": self.skipTest("Windows ACL contract") - OWNED.mkdir(exist_ok=True) - self.temp = tempfile.TemporaryDirectory(dir=OWNED) + self.sid = _current_sid() + self.temp = tempfile.TemporaryDirectory(prefix=TEMP_PREFIX) self.addCleanup(self._cleanup) self.base = Path(self.temp.name) - self.assertEqual(self.base.resolve().parent, OWNED.resolve()) + # The fixture owns this fresh directory; only paths beneath it are ever + # given a barrier ACL or have one removed again. + self.assertTrue(self.base.is_dir()) + self.assertTrue(self.base.name.startswith(TEMP_PREFIX)) self.root = self.base / "old-app" self.root.mkdir() self.old = self.root / "state" @@ -67,20 +75,37 @@ def setUp(self): (self.old / "budget.sqlite3").read_bytes()).hexdigest() for name in ("source-stage.lock", "phase-stage.lock"): (self.root / name).write_bytes(b"0") + # The fixture ACL is made explicit, protected and verified before any pin is taken, so + # nothing depends on what the temporary parent happens to pass down. + windows_fixture_acl.stabilize_owned_fixture( + self.base, [("parent", self.root), ("state", self.old)] + + [("file:" + name, self.old / name) for name in sorted(guarded.NAMES)], + read=guarded.directory_dacl_sddl) self.parent_acl_sha = hashlib.sha256( guarded.directory_dacl_sddl(self.root).encode()).hexdigest() self.file_acl_sha = {name: hashlib.sha256( guarded.directory_dacl_sddl(self.old / name).encode()).hexdigest() for name in guarded.NAMES} + def _owned(self, path: Path) -> Path: + resolved = Path(path).resolve() + if not resolved.is_relative_to(self.base.resolve()): + raise AssertionError(f"refusing ACL change outside fixture: {path}") + return resolved + def _cleanup(self): + # Lift only this test's own deny ACEs so the owned tree can be removed; + # no ancestor of the temporary directory has its ACL touched. if hasattr(self, "root") and self.root.exists(): - _call("icacls.exe", str(self.root), "/remove:d", "*" + self.sid) + _call("icacls.exe", str(self._owned(self.root)), "/remove:d", "*" + self.sid) for root in (getattr(self, "old", None), getattr(self, "archive", None)): if root is not None and root.exists(): for name in guarded.NAMES: - _call("icacls.exe", str(root / name), "/remove:d", "*" + self.sid) + if (root / name).exists(): + _call("icacls.exe", str(self._owned(root / name)), + "/remove:d", "*" + self.sid) self.temp.cleanup() + self.assertFalse(self.base.exists()) def invoke(self): return guarded.fence_closed_old_phase_guarded( @@ -125,8 +150,14 @@ def observe_then_rename(source, destination): self.assertEqual(result, self.invoke()) self.assertEqual({name: hashlib.sha256((self.archive / name).read_bytes()).hexdigest() for name in guarded.NAMES}, self.content) - self.assertIn(guarded.DENY_PARENT_ACE.format(sid=self.sid), - guarded.directory_dacl_sddl(self.root)) + # The exact parent deny ACE (type D, mask LC, no flags) for this trustee. + # Windows may serialize the numeric SID as an alias (RID 500 -> LA), so + # the trustee is compared by its Windows serialization, not literal text. + parent_dacl = guarded.directory_dacl_sddl(self.root) + self.assertTrue(guarded._has_ace(parent_dacl, guarded.DENY_PARENT_ACE, self.sid)) + # A different trustee's deny, or the file-deny mask, is not this ACE. + self.assertFalse(guarded._has_ace(parent_dacl, guarded.DENY_PARENT_ACE, "S-1-5-32-546")) + self.assertFalse(guarded._has_ace(parent_dacl, guarded.DENY_FILE_ACE, self.sid)) self.assertFalse(self.old.exists()) def test_failed_rename_retains_denial_then_same_request_recovers(self): @@ -197,6 +228,11 @@ def test_preexisting_parent_owner_deny_is_preserved_and_rejected(self): "*" + self.sid + ":(X)").returncode, 0) prior = guarded.directory_dacl_sddl(self.root) self.parent_acl_sha = hashlib.sha256(prior.encode()).hexdigest() + # Changing the parent ACL makes Windows re-propagate inheritance to the + # children (adding AI). Re-pin them so only the owner deny can reject. + self.file_acl_sha = {name: hashlib.sha256( + guarded.directory_dacl_sddl(self.old / name).encode()).hexdigest() + for name in guarded.NAMES} with self.assertRaisesRegex(Exception, "existing owner deny"): self.invoke() self.assertEqual(guarded.directory_dacl_sddl(self.root), prior) diff --git a/tests/test_paper_pilot_phase_cutover_guarded_v2_simulated_acl.py b/tests/test_paper_pilot_phase_cutover_guarded_v2_simulated_acl.py new file mode 100644 index 0000000..9217705 --- /dev/null +++ b/tests/test_paper_pilot_phase_cutover_guarded_v2_simulated_acl.py @@ -0,0 +1,913 @@ +"""Platform-independent checks of the guarded fence and the hosted diagnostics. + +The real ``fence_closed_old_phase_guarded`` and the real ``trace_fence`` run here +on any OS against an in-memory stand-in for Windows DACL reads and ``icacls`` +writes. The stand-in is a test model written from the SDDL shapes seen in the +hosted logs; it is NOT Windows. Passing here proves the strict comparison logic +and the diagnostics' structure and sanitization, and nothing about what NTFS or +``icacls`` actually do. Native behavior is covered only by the Windows-only +fence tests. +""" +from __future__ import annotations + +from contextlib import ExitStack, contextmanager +import hashlib +import importlib.util +import inspect +import json +import os +from pathlib import Path +import re +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +from src import paper_pilot_phase_cutover_guarded_v2 as guarded +from tools import windows_fixture_acl +from trace_gc.errors import ContractError + +ROOT = Path(__file__).resolve().parents[1] +_SPEC = importlib.util.spec_from_file_location( + "windows_fence_diagnostics_under_simulation", ROOT / "tools" / "windows_fence_diagnostics.py") +diagnostics = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(diagnostics) + +SID = "S-1-5-21-111-222-333-1001" +PREFIX = "jraphyte-fence-" +CRASH = "injected post-deny crash" +PARENT_COPIES = "D:AI(D;;LC;;;BU)(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)" \ + "(A;OICIID;FA;;;SY)(A;OICIID;FA;;;BA)(A;OICIID;FA;;;OW)" + + +def _tokens(flags: str) -> list[str]: + return re.findall("..", flags) + + +def _inherited_ace(ace: str, is_dir: bool) -> str | None: + kind, flags, *rest = ace[1:-1].split(";") + tokens = _tokens(flags) + if kind != "A": + return None + if is_dir and "CI" in tokens: + kept = "".join(t for t in tokens if t in ("OI", "CI")) + elif not is_dir and "OI" in tokens: + kept = "" + else: + return None + return "(" + ";".join([kind, kept + "ID"] + rest) + ")" + + +class SimulatedDacls: + """In-memory DACLs: lazy inheritance from the parent, icacls /deny and /remove:d.""" + + PROTECTED_BASE = ["(A;OICI;FA;;;SY)", "(A;OICI;FA;;;BA)", "(A;OICI;FA;;;OW)"] + MASKS = {"(WD,AD)": "DCLC", "(AD)": "LC"} + ALIASES = {"S-1-5-18": "SY", "S-1-5-32-544": "BA", "S-1-3-4": "OW"} + + def __init__(self): + self.entries: dict[str, dict] = {} + # A test switch that forces the hosted SHAPE after a deny; it does not assert Windows does this. + self.deny_adds_explicit_copies = False + self.hooks = [] + + def ensure(self, path) -> dict: + path = Path(path) + key = str(path) + if key not in self.entries: + if path.name.startswith(PREFIX): + entry = {"control": "P", "aces": list(self.PROTECTED_BASE)} + elif path.parent == path: + entry = {"control": "", "aces": ["(A;OICI;FA;;;SY)", "(A;OICI;FA;;;BA)"]} + else: + parent = self.ensure(path.parent) + inherited = (_inherited_ace(ace, path.is_dir()) for ace in parent["aces"]) + entry = {"control": "", "aces": [ace for ace in inherited if ace]} + self.entries[key] = entry + return self.entries[key] + + def read(self, path) -> str: + entry = self.ensure(path) + return "D:" + entry["control"] + "".join(entry["aces"]) + + def _leading_denies(self, entry) -> int: + count = 0 + while count < len(entry["aces"]) and entry["aces"][count].startswith("(D;"): + count += 1 + return count + + def icacls(self, path, *args) -> None: + entry = self.ensure(path) + if args[0] == "/deny": + trustee, rights = args[1].split(":", 1) + ace = "(D;;" + self.MASKS[rights] + ";;;" + trustee.lstrip("*") + ")" + if ace not in entry["aces"]: + entry["aces"].insert(0, ace) + if "AI" not in entry["control"]: + entry["control"] += "AI" + if self.deny_adds_explicit_copies: + self.add_explicit_copies(path) + elif args[0] == "/remove:d": + trustee = args[1].lstrip("*") + entry["aces"] = [ace for ace in entry["aces"] + if not (ace.startswith("(D;") and ace.endswith(";;;" + trustee + ")"))] + elif args[0] == "/inheritance:r": + if args[1] != "/grant:r": + raise AssertionError("unmodelled icacls operation") + entry["control"] = "P" + entry["control"].replace("P", "") + entry["aces"] = [ace for ace in entry["aces"] + if "ID" not in _tokens(ace[1:-1].split(";")[1])] + for grant in args[2:]: + trustee, rights = grant.split(":", 1) + flags = "OICI" if rights == "(OI)(CI)(F)" else "" + if rights not in ("(OI)(CI)(F)", "(F)"): + raise AssertionError("unmodelled icacls rights") + alias = self.ALIASES[trustee.lstrip("*")] + ace = "(A;" + flags + ";FA;;;" + alias + ")" + entry["aces"] = [a for a in entry["aces"] + if not (a.startswith("(A;") and "ID" not in _tokens(a[1:-1].split(";")[1]) + and a.endswith(";;;" + alias + ")"))] + entry["aces"].append(ace) + else: + raise AssertionError("unmodelled icacls operation") + for hook in list(self.hooks): + hook(Path(path), args) + + def add_explicit_copies(self, path) -> None: + entry = self.ensure(path) + copies = [] + for ace in entry["aces"]: + fields = ace[1:-1].split(";") + if fields[0] == "A" and "ID" in _tokens(fields[1]): + fields[1] = "".join(t for t in _tokens(fields[1]) if t != "ID") + copy = "(" + ";".join(fields) + ")" + if copy not in entry["aces"]: + copies.append(copy) + position = self._leading_denies(entry) + entry["aces"][position:position] = copies + + def broaden(self, path) -> None: + entry = self.ensure(path) + entry["aces"].insert(self._leading_denies(entry), "(A;;FR;;;WD)") + + def duplicate_first_allow(self, path, flags: str | None = None) -> None: + """Insert one more explicit allow equal to the first allow (optionally with other flags).""" + entry = self.ensure(path) + first = next(ace for ace in entry["aces"] if ace.startswith("(A;")) + fields = first[1:-1].split(";") + if flags is not None: + fields[1] = flags + entry["aces"].insert(self._leading_denies(entry), "(" + ";".join(fields) + ")") + + def materialize_tree(self, path) -> None: + self.ensure(path) + for child in Path(path).rglob("*"): + self.ensure(child) + + def moved(self, source, destination) -> None: + old, new = str(source), str(destination) + for key in [k for k in self.entries if k == old or k.startswith(old + os.sep)]: + self.entries[new + key[len(old):]] = self.entries.pop(key) + + +class _NtOs: + """The os module with ``name == 'nt'`` and a rename that carries the simulated DACLs along.""" + + name = "nt" + + def __init__(self, model: SimulatedDacls): + self._model = model + self.renames = [] + self.at_rename = None + + def rename(self, source, destination): + self.renames.append((source, destination)) + if self.at_rename is not None: + self.at_rename() + self._model.materialize_tree(source) + os.rename(source, destination) + self._model.moved(source, destination) + + def __getattr__(self, attribute): + return getattr(os, attribute) + + +@contextmanager +def simulated_windows(model: SimulatedDacls): + with ExitStack() as stack: + proxy = _NtOs(model) + stack.enter_context(patch.object(guarded, "os", proxy)) + stack.enter_context(patch.object(guarded, "directory_dacl_sddl", model.read)) + stack.enter_context(patch.object(guarded, "_icacls", model.icacls)) + stack.enter_context(patch.object(guarded, "_stage_lock", lambda path: path.open("r+b"))) + stack.enter_context(patch.object(guarded, "_current_sid", lambda: SID)) + stack.enter_context(patch.object(guarded, "_canonical_trustee", lambda trustee: trustee)) + yield proxy + + +class SimulatedModelTests(unittest.TestCase): + def setUp(self): + self.model = SimulatedDacls() + self.temp = tempfile.TemporaryDirectory(prefix=PREFIX) + self.addCleanup(self.temp.cleanup) + self.base = Path(self.temp.name) + + def test_fixture_inheritance_matches_the_shapes_seen_in_hosted_logs(self): + child = self.base / "dir" + child.mkdir() + (child / "file").write_bytes(b"0") + self.assertEqual(self.model.read(self.base), + "D:P(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)") + self.assertEqual(self.model.read(child), + "D:(A;OICIID;FA;;;SY)(A;OICIID;FA;;;BA)(A;OICIID;FA;;;OW)") + self.assertEqual(self.model.read(child / "file"), + "D:(A;ID;FA;;;SY)(A;ID;FA;;;BA)(A;ID;FA;;;OW)") + + def test_deny_is_prepended_with_ai_and_remove_lifts_only_that_trustee(self): + child = self.base / "file" + child.write_bytes(b"0") + before = self.model.read(child) + self.model.icacls(child, "/deny", "*" + SID + ":(WD,AD)") + self.assertEqual(self.model.read(child), "D:AI(D;;DCLC;;;" + SID + ")" + before[2:]) + self.model.icacls(child, "/remove:d", "*S-1-1-0") + self.assertIn(SID, self.model.read(child)) + self.model.icacls(child, "/remove:d", "*" + SID) + self.assertEqual(self.model.read(child), "D:AI" + before[2:]) + + def test_explicit_copy_shape_matches_the_hosted_logs(self): + directory = self.base / "dir" + directory.mkdir() + self.model.read(directory) + self.model.icacls(directory, "/deny", "*BU:(AD)") + self.model.add_explicit_copies(directory) + self.assertEqual(self.model.read(directory), PARENT_COPIES) + + +class SimulatedFenceCase(unittest.TestCase): + def setUp(self): + self.model = SimulatedDacls() + self.proxy = self.enterContext(simulated_windows(self.model)) + self.temp = tempfile.TemporaryDirectory(prefix=PREFIX) + self.addCleanup(self.temp.cleanup) + self.base = Path(self.temp.name) + self.root = self.base / "old-app" + self.old = self.root / "state" + self.old.mkdir(parents=True) + self.archive = self.base / "archive" / "state" + self.archive.parent.mkdir() + self.receipts = self.base / "receipts" + self.receipts.mkdir() + self.content = {} + for name in guarded.NAMES: + body = b"0" if name == "controller.lock" else ("simulated:" + name).encode() + (self.old / name).write_bytes(body) + self.content[name] = hashlib.sha256(body).hexdigest() + for name in ("source-stage.lock", "phase-stage.lock"): + (self.root / name).write_bytes(b"0") + + def stabilize(self): + targets = [("parent", self.root), ("state", self.old)] + [ + ("file:" + name, self.old / name) for name in sorted(guarded.NAMES)] + + def run(command, **options): + self.model.icacls(Path(command[1]), *command[2:]) + return subprocess.CompletedProcess(command, 0, "", "") + windows_fixture_acl.stabilize_owned_fixture(self.base, targets, read=self.model.read, run=run) + + def pin(self): + self.original_parent = self.model.read(self.root) + self.original_files = {name: self.model.read(self.old / name) for name in guarded.NAMES} + self.parent_acl_sha = hashlib.sha256(self.original_parent.encode()).hexdigest() + self.file_acl_sha = {name: hashlib.sha256(value.encode()).hexdigest() + for name, value in self.original_files.items()} + + def invoke(self): + return guarded.fence_closed_old_phase_guarded( + old_root=self.old, archived_root=self.archive, expected_file_sha256=self.content, + source_stage_lock=self.root / "source-stage.lock", + phase_stage_lock=self.root / "phase-stage.lock", deny_sid=SID, + expected_parent_dacl_sha256=self.parent_acl_sha, + expected_file_dacl_sha256=self.file_acl_sha, + barrier_receipt_path=self.receipts / "barrier.json", + fence_receipt_path=self.receipts / "fence.json") + + def crash_after_first_deny(self, also=None): + seen = {"count": 0} + + def hook(path, args): + if args[0] == "/deny": + seen["count"] += 1 + if seen["count"] == 1: + if also is not None: + also(path) + raise RuntimeError(CRASH) + self.model.hooks.append(hook) + + def assert_failed_closed(self): + self.assertEqual(self.proxy.renames, []) + self.assertTrue(self.old.is_dir()) + self.assertFalse(self.archive.exists()) + self.assertFalse((self.receipts / "fence.json").exists()) + self.assertTrue(guarded._file_deny(self.old / "controller.lock", SID)) + + +class SimulatedGuardTests(SimulatedFenceCase): + def test_faithful_icacls_completes_and_preserves_every_original_ace(self): + self.pin() + at_rename = {} + + def observe(): + at_rename["files"] = {name: guarded._file_deny(self.old / name, SID) + for name in guarded.NAMES} + at_rename["parent"] = guarded._has_ace(self.model.read(self.root), + guarded.DENY_PARENT_ACE, SID) + self.proxy.at_rename = observe + result = self.invoke() + self.assertEqual(result["status"], "OLD_PATH_FENCED_NEW_PHASE_ALLOWED") + self.assertEqual(at_rename, {"files": {name: True for name in guarded.NAMES}, "parent": True}) + deny = "(D;;LC;;;" + SID + ")" + parent_now = guarded._aces(self.model.read(self.root)) + self.assertIn(deny, parent_now) + parent_now.remove(deny) + self.assertEqual(parent_now, guarded._aces(self.original_parent)) + for name in guarded.NAMES: + self.assertEqual(guarded._aces(self.model.read(self.archive / name)), + guarded._aces(self.original_files[name]), name) + self.assertFalse(guarded._file_deny(self.archive / name, SID)) + + def test_declared_explicit_and_inherited_source_is_preserved_exactly(self): + self.model.add_explicit_copies(self.root) + for name in guarded.NAMES: + self.model.add_explicit_copies(self.old / name) + self.pin() + self.assertEqual(len([a for a in guarded._aces(self.original_parent) if a.startswith("(A;OICI;")]), 3) + self.invoke() + parent_now = guarded._aces(self.model.read(self.root)) + parent_now.remove("(D;;LC;;;" + SID + ")") + self.assertEqual(parent_now, guarded._aces(self.original_parent)) + for name in guarded.NAMES: + self.assertEqual(guarded._aces(self.model.read(self.archive / name)), + guarded._aces(self.original_files[name]), name) + + def test_unexpected_explicit_copies_on_the_parent_fail_closed_before_the_rename(self): + self.pin() + self.model.deny_adds_explicit_copies = True + with self.assertRaisesRegex(ContractError, "old parent DACL broadened during fence"): + self.invoke() + self.assertEqual(self.proxy.renames, []) + self.assertTrue(self.old.is_dir()) + self.assertFalse(self.archive.exists()) + self.assertFalse((self.receipts / "fence.json").exists()) + self.assertTrue(guarded._file_deny(self.old / "controller.lock", SID)) + + def test_unexpected_explicit_copies_on_a_denied_file_fail_closed_on_replay(self): + self.pin() + self.model.deny_adds_explicit_copies = True + self.crash_after_first_deny() + with self.assertRaisesRegex(RuntimeError, CRASH): + self.invoke() + self.model.hooks.clear() + with self.assertRaisesRegex(ContractError, "unreviewed old file DACL drift"): + self.invoke() + self.assert_failed_closed() + + def test_real_parent_broadening_after_the_guard_deny_fails_closed(self): + self.pin() + self.model.hooks.append(lambda path, args: self.model.broaden(path) + if path == self.root and args[0] == "/deny" else None) + with self.assertRaisesRegex(ContractError, "old parent DACL broadened during fence"): + self.invoke() + self.assert_failed_closed() + + def test_real_file_broadening_is_rejected_on_replay_before_the_rename(self): + self.pin() + self.crash_after_first_deny(also=self.model.broaden) + with self.assertRaisesRegex(RuntimeError, CRASH): + self.invoke() + self.model.hooks.clear() + with self.assertRaisesRegex(ContractError, "unreviewed old file DACL drift"): + self.invoke() + self.assert_failed_closed() + + +class SimulatedStabilizedFixtureTests(SimulatedFenceCase): + """The owned fixture made explicit and protected before it is pinned (model, not Windows).""" + + def test_stabilized_fixture_has_only_the_reviewed_explicit_allows_and_no_inherited_ace(self): + self.stabilize() + self.assertEqual(self.model.read(self.root), + "D:P(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)") + self.assertEqual(self.model.read(self.old), + "D:P(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)") + for name in guarded.NAMES: + self.assertEqual(self.model.read(self.old / name), + "D:P(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)", name) + for path in (self.root, self.old, *(self.old / name for name in guarded.NAMES)): + self.assertNotIn("ID", self.model.read(path).replace("D:", ""), path) + + def test_the_unchanged_guard_rejects_the_hosted_shape_on_an_unstabilized_fixture(self): + self.pin() + self.model.deny_adds_explicit_copies = True + with self.assertRaisesRegex(ContractError, "old parent DACL broadened during fence"): + self.invoke() + self.assertEqual(self.proxy.renames, []) + + def test_the_same_switch_leaves_a_stabilized_fixture_exactly_preserved(self): + self.stabilize() + self.pin() + self.model.deny_adds_explicit_copies = True + result = self.invoke() + self.assertEqual(result["status"], "OLD_PATH_FENCED_NEW_PHASE_ALLOWED") + self.assertEqual(len(self.proxy.renames), 1) + parent_now = guarded._aces(self.model.read(self.root)) + parent_now.remove("(D;;LC;;;" + SID + ")") + self.assertEqual(parent_now, guarded._aces(self.original_parent)) + for name in guarded.NAMES: + self.assertEqual(guarded._aces(self.model.read(self.archive / name)), + guarded._aces(self.original_files[name]), name) + self.assertFalse(guarded._file_deny(self.archive / name, SID)) + + def test_an_unexpected_explicit_allow_copy_on_the_parent_still_fails_closed(self): + self.stabilize() + self.pin() + self.model.hooks.append(lambda path, args: self.model.duplicate_first_allow(path, flags="") + if path == self.root and args[0] == "/deny" else None) + with self.assertRaisesRegex(ContractError, "old parent DACL broadened during fence"): + self.invoke() + self.assertIn("(A;;FA;;;SY)", guarded._aces(self.model.read(self.root))) + self.assert_failed_closed() + + def test_an_unexpected_explicit_allow_copy_on_a_denied_file_still_fails_closed_on_replay(self): + self.stabilize() + self.pin() + self.crash_after_first_deny(also=self.model.duplicate_first_allow) + with self.assertRaisesRegex(RuntimeError, CRASH): + self.invoke() + self.model.hooks.clear() + with self.assertRaisesRegex(ContractError, "unreviewed old file DACL drift"): + self.invoke() + self.assert_failed_closed() + + def test_real_broadening_still_fails_closed_on_a_stabilized_fixture(self): + self.stabilize() + self.pin() + self.model.hooks.append(lambda path, args: self.model.broaden(path) + if path == self.root and args[0] == "/deny" else None) + with self.assertRaisesRegex(ContractError, "old parent DACL broadened during fence"): + self.invoke() + self.assert_failed_closed() + + +class SimulatedTraceTests(SimulatedFenceCase): + """The real trace_fence, run on its own fixture, against the same simulated layer.""" + + LABELS = {"temp_parent", "base", "parent", "state"} | {"file:" + n for n in guarded.NAMES} + SECRET = r"C:\Users\private-account\secret S-1-5-21-1-2-3-1001" + + def run_trace(self, *, remove_exit=0, factory=tempfile.TemporaryDirectory, timeouts=(), on_cleanup=None, + protect_exit=0): + """Cleanup-time ``icacls.exe`` calls (numbered from 1) listed in ``timeouts`` time out.""" + real_run = subprocess.run + calls = {"cleanup": 0} + + def fake_run(args, *rest, **options): + if args and args[0] == "icacls.exe" and "/inheritance:r" in args: + if protect_exit == 0: + self.model.icacls(Path(args[1]), *args[2:]) + return subprocess.CompletedProcess(args, protect_exit, "", "") + if args and args[0] == "icacls.exe": + calls["cleanup"] += 1 + if calls["cleanup"] == 1 and on_cleanup is not None: + on_cleanup() + if calls["cleanup"] in timeouts: + raise subprocess.TimeoutExpired(args, 30) + if remove_exit == 0: + self.model.icacls(Path(args[1]), *args[2:]) + return subprocess.CompletedProcess(args, remove_exit, "", "") + return real_run(args, *rest, **options) + with patch.object(diagnostics.subprocess, "run", side_effect=fake_run): + return diagnostics.trace_fence(guarded, temp_factory=factory) + + def deny_parent_and_a_file_then_fail(self): + def deny_then_fail(**request): + guarded._icacls(request["old_root"] / "controller.lock", "/deny", "*" + SID + ":(WD,AD)") + guarded._icacls(request["old_root"].parent, "/deny", "*" + SID + ":(AD)") + raise PermissionError(self.SECRET) + return patch.object(guarded, "fence_closed_old_phase_guarded", side_effect=deny_then_fail) + + def test_completed_fence_trace_records_every_phase_with_strict_verdicts(self): + trace, ok = self.run_trace() + self.assertTrue(ok, trace) + self.assertEqual(trace["outcome"], {"status": "OLD_PATH_FENCED_NEW_PHASE_ALLOWED"}) + for phase in ("setup_as_created", "setup", "at_end", "after_remove_d"): + self.assertEqual(set(trace[phase]), self.LABELS, phase) + self.assertEqual(trace["fixture_acl"], {"status": "stabilized", "targets": 2 + len(guarded.NAMES)}) + created = trace["setup_as_created"] + self.assertEqual((created["parent"]["control"], created["parent"]["inherited_allow"], + created["parent"]["explicit_allow"]), ("", 3, 0)) + self.assertEqual(created["file:controller.lock"]["inherited_allow"], 3) + setup = trace["setup"] + self.assertTrue(setup["base"]["protected"]) + for label in ("parent", "state") + tuple("file:" + name for name in guarded.NAMES): + self.assertEqual((setup[label]["protected"], setup[label]["inherited_allow"], + setup[label]["explicit_allow"], setup[label]["ace_count"], + setup[label]["explicit_copies_of_inherited"], setup[label]["deny"]), + (True, 0, 3, 3, 0, 0), label) + self.assertNotIn("strict", setup["parent"]) + + steps = {(step["target"], step["operation"]): step for step in trace["steps"]} + self.assertEqual(len(trace["steps"]), 11) + parent_deny = steps[("parent", "deny")] + self.assertEqual(parent_deny["icacls"], "ok") + self.assertEqual(set(parent_deny["before"]), self.LABELS) + self.assertEqual(set(parent_deny["after"]), self.LABELS) + self.assertEqual(parent_deny["before"]["parent"]["deny"], 0) + self.assertEqual(parent_deny["before"]["parent"]["strict"], + {"admission_only_added": True, "same_as_original": True}) + self.assertEqual(parent_deny["after"]["parent"]["strict"], + {"admission_only_added": True, "same_as_original": False}) + self.assertTrue(parent_deny["after"]["parent"]["auto_inherited"]) + self.assertEqual(parent_deny["after"]["parent"]["deny"], 1) + released = steps[("archived:controller.lock", "remove_deny")] + self.assertEqual(released["after"]["file:controller.lock"]["strict"], + {"admission_only_added": True, "same_as_original": True}) + # Nothing but the guard's own calls runs between the deny steps: each pre-state is the last post-state. + for earlier, later in zip(trace["steps"][:5], trace["steps"][1:6]): + self.assertEqual(later["before"], earlier["after"]) + + self.assertEqual(trace["at_end"]["parent"]["strict"], + {"admission_only_added": True, "same_as_original": False}) + self.assertEqual(trace["after_remove_d"]["parent"]["strict"], + {"admission_only_added": True, "same_as_original": True}) + cleanup = trace["cleanup"] + self.assertEqual({key: cleanup[key] for key in cleanup if key != "steps"}, + {"paths_checked": 6, "denies_remaining": 0, "verification_errors": 0, + "removal_failures": [], "directory_removal": "ok"}) + self.assertEqual([step["target"] for step in cleanup["steps"]], + ["parent"] + ["archived:" + n for n in sorted(guarded.NAMES)]) + parent_cleanup = cleanup["steps"][0] + self.assertEqual((parent_cleanup["before"]["parent"]["deny"], parent_cleanup["after"]["parent"]["deny"], + parent_cleanup["icacls_exit"]), (1, 0, 0)) + self.assertEqual(set(parent_cleanup["before"]), self.LABELS) + + def test_explicit_copies_are_reported_as_a_strict_failure_and_still_cleaned_up(self): + # A stand-in that leaves the fixture exactly as created (inheriting, unprotected), so the + # hosted shape can be reported; the real stabilization is covered by the other tests. + self.model.deny_adds_explicit_copies = True + with patch.object(windows_fixture_acl, "stabilize_owned_fixture", lambda *args, **kwargs: None): + trace, ok = self.run_trace() + self.assertTrue(ok, trace) + self.assertEqual(trace["outcome"], {"error": { + "type": "ContractError", "category": "PHASE_FENCE", + "detail_code": "PARENT_BROADENED_DURING_FENCE"}}) + steps = {(step["target"], step["operation"]): step for step in trace["steps"]} + self.assertEqual(len(trace["steps"]), 6) + parent = steps[("parent", "deny")]["after"]["parent"] + self.assertEqual((parent["ace_count"], parent["protected"], parent["auto_inherited"], + parent["explicit_copies_of_inherited"]), (7, False, True, 3)) + self.assertEqual(parent["strict"], + {"admission_only_added": False, "same_as_original": False}) + end = trace["at_end"] + self.assertEqual(end["state"]["explicit_copies_of_inherited"], 0) + self.assertEqual(end["file:controller.lock"]["explicit_copies_of_inherited"], 3) + self.assertEqual(end["file:controller.lock"]["strict"]["admission_only_added"], False) + # Lifting the fixture's own denies does not lift the copies: reported as it is. + after = trace["after_remove_d"] + self.assertEqual((after["parent"]["deny"], after["parent"]["explicit_copies_of_inherited"]), (0, 3)) + self.assertEqual(after["parent"]["strict"]["same_as_original"], False) + self.assertEqual(trace["cleanup"]["denies_remaining"], 0) + + def test_trace_is_sanitized(self): + for shaped in (False, True): + self.model.deny_adds_explicit_copies = shaped + trace, _ = self.run_trace() + rendered = json.dumps(trace) + self.assertNotIn(tempfile.gettempdir(), rendered) + self.assertNotIn("111-222-333", rendered) + self.assertIn("S-1-5-21--1001", rendered) + self.assertNotIn(PREFIX, rendered) + + def test_unverified_cleanup_is_reported_as_not_ok(self): + trace, ok = self.run_trace(remove_exit=5) + self.assertFalse(ok) + self.assertGreater(trace["cleanup"]["denies_remaining"], 0) + self.assertEqual({failure["target"] for failure in trace["cleanup"]["removal_failures"]}, + {step["target"] for step in trace["cleanup"]["steps"]}) + self.assertTrue(all(failure["exit_code"] == 5 for failure in trace["cleanup"]["removal_failures"])) + + def test_a_failure_raised_after_real_denies_is_reported_and_cleaned_up(self): + with self.deny_parent_and_a_file_then_fail(): + trace, ok = self.run_trace() + self.assertTrue(ok, trace) + self.assertEqual(trace["outcome"], + {"error": {"type": "PermissionError", "category": "unspecified"}}) + self.assertEqual([(s["target"], s["operation"]) for s in trace["steps"]], + [("old:controller.lock", "deny"), ("parent", "deny")]) + self.assertEqual(trace["at_end"]["parent"]["deny"], 1) + self.assertEqual(trace["after_remove_d"]["parent"]["deny"], 0) + self.assertEqual(trace["after_remove_d"]["file:controller.lock"]["deny"], 0) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_a_failing_icacls_call_is_recorded_with_its_pre_and_post_state_before_it_propagates(self): + def failing(path, *args): + raise ContractError("PHASE_FENCE", "Windows DACL update failed") + with patch.object(guarded, "_icacls", side_effect=failing): + trace, ok = self.run_trace() + self.assertTrue(ok, trace) + self.assertEqual(trace["outcome"]["error"]["detail_code"], "ICACLS_UPDATE_FAILED") + step = trace["steps"][0] + self.assertEqual(step["icacls"], "failed") + self.assertEqual(step["error"], {"type": "ContractError", "category": "PHASE_FENCE"}) + self.assertEqual(set(step["before"]), self.LABELS) + self.assertEqual(step["before"], step["after"]) + + def test_a_comparator_that_raises_is_reported_without_masking_the_trace(self): + real = guarded._only_parent_admission_added + + def raises_for_the_diagnostics_only(*args): + caller = next(frame for frame in inspect.stack()[1:] + if not frame.filename.endswith("mock.py")) + if caller.filename.endswith("windows_fence_diagnostics.py"): + raise ValueError(r"C:\Users\private-account") + return real(*args) + with patch.object(guarded, "_only_parent_admission_added", + side_effect=raises_for_the_diagnostics_only): + trace, ok = self.run_trace() + self.assertEqual(trace["outcome"], {"status": "OLD_PATH_FENCED_NEW_PHASE_ALLOWED"}) + parent = {(s["target"], s["operation"]): s for s in trace["steps"]}[("parent", "deny")] + self.assertIsNone(parent["after"]["parent"]["strict"]["admission_only_added"]) + self.assertEqual(parent["after"]["parent"]["strict"]["admission_only_added_error"], + {"type": "ValueError", "category": "unspecified"}) + self.assertNotIn("private-account", json.dumps(trace)) + self.assertTrue(ok) + + def test_a_first_target_removal_timeout_does_not_stop_the_later_targets(self): + with self.deny_parent_and_a_file_then_fail(): + trace, ok = self.run_trace(timeouts={1}) + cleanup = trace["cleanup"] + self.assertFalse(ok) + self.assertEqual(cleanup["removal_failures"], + [{"target": "parent", "error": {"type": "TimeoutExpired", "category": "unspecified"}}]) + # The first target (the parent) kept its deny and is reported as such ... + self.assertEqual(cleanup["denies_remaining"], 1) + self.assertEqual(trace["after_remove_d"]["parent"]["deny"], 1) + # ... while the later targets each still got their own removal attempt, and it worked. + self.assertEqual([step["target"] for step in cleanup["steps"]], + ["parent"] + ["old:" + n for n in sorted(guarded.NAMES)]) + self.assertEqual(trace["after_remove_d"]["file:controller.lock"]["deny"], 0) + lifted = {step["target"]: step for step in cleanup["steps"]}["old:controller.lock"] + self.assertEqual((lifted["icacls_exit"], lifted["before"]["file:controller.lock"]["deny"], + lifted["after"]["file:controller.lock"]["deny"]), (0, 1, 0)) + self.assertEqual(cleanup["steps"][0]["icacls"], "failed") + self.assertEqual(cleanup["steps"][0]["after"]["parent"]["deny"], 1) + self.assertEqual(cleanup["paths_checked"], 6) + + def test_a_removal_failure_is_not_clean_even_when_verification_finds_no_survivor(self): + with patch.object(guarded, "fence_closed_old_phase_guarded", side_effect=RuntimeError(self.SECRET)): + trace, ok = self.run_trace(timeouts={3}) + self.assertFalse(ok) + self.assertEqual(trace["cleanup"]["denies_remaining"], 0) + self.assertEqual([failure["target"] for failure in trace["cleanup"]["removal_failures"]], + ["old:" + sorted(guarded.NAMES)[1]]) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_an_unreadable_survivor_fails_verification_and_the_other_targets_are_still_verified(self): + unreadable = {"armed": False} + real_read = self.model.read + + def read(path): + if unreadable["armed"] and Path(path).name == "old-app": + raise OSError(self.SECRET) + return real_read(path) + + def arm(): + unreadable["armed"] = True + with patch.object(guarded, "directory_dacl_sddl", read): + with self.deny_parent_and_a_file_then_fail(): + trace, ok = self.run_trace(on_cleanup=arm) + cleanup = trace["cleanup"] + self.assertFalse(ok) + self.assertEqual((cleanup["verification_errors"], cleanup["denies_remaining"], + cleanup["paths_checked"]), (1, 1, 6)) + self.assertEqual(cleanup["steps"][0]["after"]["parent"], + {"error": {"type": "OSError", "category": "unspecified"}}) + self.assertEqual(trace["after_remove_d"]["parent"], {"error": {"type": "OSError", "category": "unspecified"}}) + self.assertEqual(trace["after_remove_d"]["file:controller.lock"]["deny"], 0) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_a_failed_temp_directory_removal_makes_the_status_not_clean_and_is_reported(self): + class FailsOnRemoval(tempfile.TemporaryDirectory): + def __exit__(self, *exc): + super().__exit__(*exc) + raise OSError(SimulatedTraceTests.SECRET) + trace, ok = self.run_trace(factory=FailsOnRemoval) + self.assertFalse(ok) + cleanup = trace["cleanup"] + self.assertEqual(cleanup["denies_remaining"], 0) + self.assertEqual(cleanup["directory_removal"], "failed") + self.assertEqual(cleanup["directory_removal_error"], {"type": "OSError", "category": "unspecified"}) + self.assertNotIn("error", trace) + self.assertEqual(trace["outcome"], {"status": "OLD_PATH_FENCED_NEW_PHASE_ALLOWED"}) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_a_fixture_that_cannot_be_stabilized_is_reported_and_the_fence_is_not_run(self): + with patch.object(guarded, "fence_closed_old_phase_guarded") as fence: + trace, ok = self.run_trace(protect_exit=5) + fence.assert_not_called() + self.assertFalse(ok) + self.assertEqual(trace["fixture_acl"]["status"], "failed") + self.assertEqual({problem["code"] for problem in trace["fixture_acl"]["problems"]}, {"ICACLS_FAILED"}) + self.assertEqual(trace["outcome"], {"error": {"type": "FixtureAclError", "category": "unspecified"}}) + self.assertEqual(trace["steps"], []) + self.assertEqual(trace["cleanup"]["denies_remaining"], 0) + self.assertEqual(trace["cleanup"]["directory_removal"], "ok") + + def test_an_unverifiable_stabilized_dacl_is_reported_with_fixed_codes_only(self): + real_read = self.model.read + + def inherited_again(path): + return real_read(path).replace("(A;;FA;;;SY)", "(A;ID;FA;;;SY)") + with patch.object(guarded, "directory_dacl_sddl", inherited_again): + trace, ok = self.run_trace() + self.assertFalse(ok) + problems = trace["fixture_acl"]["problems"] + self.assertTrue(problems) + self.assertTrue(all(problem["code"] in ("INHERITED_ACE",) for problem in problems), problems) + self.assertEqual({problem["target"] for problem in problems}, {"file:" + n for n in guarded.NAMES}) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_a_failed_setup_is_not_clean(self): + class FailsToCreate(tempfile.TemporaryDirectory): + def __enter__(self): + self.cleanup() + raise PermissionError(SimulatedTraceTests.SECRET) + trace, ok = self.run_trace(factory=FailsToCreate) + self.assertFalse(ok) + self.assertEqual(trace["error"], {"type": "PermissionError", "category": "unspecified"}) + self.assertIsNone(trace["cleanup"]["denies_remaining"]) + self.assertEqual(trace["cleanup"]["directory_removal"], "not_reached") + + def test_recording_is_bounded_and_the_overflow_is_counted(self): + with patch.object(diagnostics, "MAX_RECORDED_OPERATIONS", 4): + trace, ok = self.run_trace() + recorded = len(trace["steps"]) + len(trace["cleanup"]["steps"]) + self.assertEqual(recorded, 4) + self.assertEqual(trace["operations_dropped"], 11 + 6 - 4) + self.assertTrue(ok, trace) + + +class SimulatedCollectTests(SimulatedFenceCase): + """The real collect() (probe deny, probe cleanup, probe directory removal, then the trace).""" + + SECRET = SimulatedTraceTests.SECRET + + def run_collect(self, *, factory=tempfile.TemporaryDirectory, read=None): + real_run = subprocess.run + + def fake_run(args, *rest, **options): + if args and args[0] == "icacls.exe": + self.model.icacls(Path(args[1]), *args[2:]) + return subprocess.CompletedProcess(args, 0, "", "") + return real_run(args, *rest, **options) + with ExitStack() as stack: + stack.enter_context(patch.object(diagnostics.subprocess, "run", side_effect=fake_run)) + if read is not None: + stack.enter_context(patch.object(guarded, "directory_dacl_sddl", read)) + return diagnostics.collect(guarded, temp_factory=factory) + + @staticmethod + def removal_fails_for(prefix): + class FailsOnRemoval(tempfile.TemporaryDirectory): + def __init__(self, *args, prefix=None, **options): + super().__init__(*args, prefix=prefix, **options) + self.fails = prefix == fail_prefix + + def __exit__(self, *exc): + super().__exit__(*exc) + if self.fails: + raise OSError(SimulatedCollectTests.SECRET) + fail_prefix = prefix + return FailsOnRemoval + + def test_a_clean_run_reports_every_cleanup_ok(self): + report, ok = self.run_collect() + self.assertTrue(ok, report) + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertEqual(report["probe_directory_removal"], "ok") + self.assertEqual(report["fence_trace"]["fixture_acl"]["status"], "stabilized") + self.assertNotIn("diagnostic_error", report) + + def test_only_the_probe_directory_removal_failing_makes_the_status_not_clean(self): + report, ok = self.run_collect(factory=self.removal_fails_for("jraphyte-fence-diag-")) + self.assertFalse(ok) + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertEqual(report["probe_directory_removal"], "failed") + self.assertEqual(report["probe_directory_removal_error"], {"type": "OSError", "category": "unspecified"}) + self.assertNotIn("diagnostic_error", report) + # The separate fence trace was entirely clean, so it is the probe directory alone. + self.assertEqual(report["fence_trace"]["cleanup"]["directory_removal"], "ok") + self.assertEqual(report["fence_trace"]["cleanup"]["denies_remaining"], 0) + self.assertEqual(report["fence_trace"]["outcome"], {"status": "OLD_PATH_FENCED_NEW_PHASE_ALLOWED"}) + self.assertNotIn("private-account", json.dumps(report)) + + def test_main_exits_nonzero_when_only_the_probe_directory_removal_fails(self): + import io + from contextlib import redirect_stdout + for fails, expected in (("jraphyte-fence-diag-", 1), ("no-such-prefix", 0)): + factory = self.removal_fails_for(fails) + real_run = subprocess.run + + def fake_run(args, *rest, **options): + if args and args[0] == "icacls.exe": + self.model.icacls(Path(args[1]), *args[2:]) + return subprocess.CompletedProcess(args, 0, "", "") + return real_run(args, *rest, **options) + buffer = io.StringIO() + with patch.object(diagnostics.subprocess, "run", side_effect=fake_run), \ + patch.object(diagnostics.collect, "__kwdefaults__", {"temp_factory": factory}), \ + redirect_stdout(buffer): + self.assertEqual(diagnostics.main(), expected, fails) + self.assertEqual(json.loads(buffer.getvalue())["probe_directory_removal"], + "failed" if expected else "ok") + + def test_a_probe_body_failure_and_the_probe_directory_removal_failure_are_both_reported(self): + reads = {"count": 0} + real_read = self.model.read + + def fail_second_read(path): + reads["count"] += 1 + if reads["count"] == 2: + raise RuntimeError(self.SECRET) + return real_read(path) + report, ok = self.run_collect(factory=self.removal_fails_for("jraphyte-fence-diag-"), read=fail_second_read) + self.assertFalse(ok) + self.assertEqual(report["diagnostic_error"], {"type": "RuntimeError", "category": "unspecified"}) + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertEqual(report["probe_directory_removal"], "failed") + self.assertNotIn("private-account", json.dumps(report)) + + def test_the_first_dacl_read_failing_is_recorded_before_the_deny_and_the_probe_continues(self): + reads = {"count": 0} + real_read = self.model.read + + def fail_first_read(path): + reads["count"] += 1 + if reads["count"] == 1: + raise RuntimeError(self.SECRET) + return real_read(path) + report, ok = self.run_collect(read=fail_first_read) + self.assertTrue(ok, report) + self.assertEqual(report["probe_before_error"], {"type": "RuntimeError", "category": "unspecified"}) + self.assertNotIn("probe_dacl_before_runner_deny", report) + self.assertNotIn("diagnostic_error", report) + self.assertTrue(report["barrier_recognizes_deny"]) + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + + def test_a_post_deny_dacl_read_failing_is_a_diagnostic_error_and_the_deny_is_still_removed(self): + reads = {"count": 0} + real_read = self.model.read + + def fail_second_read(path): + reads["count"] += 1 + if reads["count"] == 2: + raise RuntimeError(self.SECRET) + return real_read(path) + report, ok = self.run_collect(read=fail_second_read) + self.assertTrue(ok, report) + self.assertIn("probe_dacl_before_runner_deny", report) + self.assertEqual(report["diagnostic_error"], {"type": "RuntimeError", "category": "unspecified"}) + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertNotIn("private-account", json.dumps(report)) + + +class DescribeDaclTests(unittest.TestCase): + def test_hosted_shape_is_described_structurally(self): + shape = diagnostics.describe_dacl(PARENT_COPIES) + self.assertEqual({key: shape[key] for key in shape if key != "sddl"}, + {"control": "AI", "protected": False, "auto_inherited": True, + "ace_count": 7, "deny": 1, "explicit_allow": 3, "inherited_allow": 3, + "explicit_copies_of_inherited": 3}) + + def test_protected_and_auto_inherited_control_flags_are_separate(self): + base = "(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)" + for header, protected, auto_inherited in (("D:", False, False), ("D:P", True, False), + ("D:AI", False, True), ("D:PAI", True, True)): + shape = diagnostics.describe_dacl(header + base) + self.assertEqual((shape["control"], shape["protected"], shape["auto_inherited"]), + (header[2:], protected, auto_inherited), header) + self.assertEqual(shape["explicit_copies_of_inherited"], 0) + + def test_an_explicit_allow_without_an_inherited_twin_is_not_a_copy(self): + shape = diagnostics.describe_dacl("D:(A;;FR;;;WD)(A;ID;FA;;;SY)(A;;FA;;;BA)") + self.assertEqual((shape["explicit_allow"], shape["inherited_allow"], + shape["explicit_copies_of_inherited"]), (2, 1, 0)) + + def test_machine_and_directory_sub_authorities_are_redacted(self): + text = diagnostics.describe_dacl( + "D:(D;;LC;;;S-1-5-21-1-2-3-500)(A;;FA;;;S-1-12-1-11-22-33-44)(A;;FA;;;S-1-5-18)")["sddl"] + self.assertEqual(text, "D:(D;;LC;;;S-1-5-21--500)" + "(A;;FA;;;S-1-12-1-)(A;;FA;;;S-1-5-18)") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_paper_pilot_phase_cutover_guarded_v2_strict_dacl.py b/tests/test_paper_pilot_phase_cutover_guarded_v2_strict_dacl.py new file mode 100644 index 0000000..0420c08 --- /dev/null +++ b/tests/test_paper_pilot_phase_cutover_guarded_v2_strict_dacl.py @@ -0,0 +1,337 @@ +"""Strict-DACL contract regressions for the guarded admission barrier. + +The barrier pins the original DACL of the fence parent and of every state file +and, after each of its own ``icacls`` writes, requires the remaining ACE list to +be *identical* (same ACEs, count, flags and order; only the admission deny added +and the ``AI`` control flag tolerated). An explicit allow ACE is not equivalent +to an inherited one: it survives a later revocation or removal of the parent +grant, so an unexpected explicit allow copy must fail closed, never be absorbed. + +Hosted Windows Server 2025 traces show that ``icacls /deny`` on an unprotected object that +carries inherited allows leaves explicit copies of those allows behind, which this strict +comparison rejects. Native fixtures therefore start from a deterministic owned ACL: protected, +with only the explicit allows of the CPython 0o700 base (tools/windows_fixture_acl.py), made +and verified before anything is pinned. That stabilization is not claimed sufficient until a +hosted run shows it. The one case that needs inherited allows (the declared explicit-plus- +inherited source) declares and verifies that shape itself before pinning. No case relaxes or +blesses anything: the negatives still require a changed ACE list to fail closed. +""" +from __future__ import annotations + +import hashlib +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +from src import paper_pilot_phase_cutover_guarded_v2 as guarded +from tools import windows_fixture_acl + + +ALIAS = "BU" # an already-serialized trustee keeps the text cases off ctypes +FILE_ORIGINAL = "D:(A;ID;FA;;;SY)(A;ID;FA;;;BA)(A;ID;FA;;;OW)" +FILE_WITH_COPIES = ("D:AI(D;;DCLC;;;BU)(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)" + "(A;ID;FA;;;SY)(A;ID;FA;;;BA)(A;ID;FA;;;OW)") +PARENT_ORIGINAL = "D:(A;OICIID;FA;;;SY)(A;OICIID;FA;;;BA)(A;OICIID;FA;;;OW)" +PARENT_WITH_COPIES = ("D:AI(D;;LC;;;BU)(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)" + "(A;OICIID;FA;;;SY)(A;OICIID;FA;;;BA)(A;OICIID;FA;;;OW)") +TRUSTEE_SIDS = {"SY": "S-1-5-18", "BA": "S-1-5-32-544", "OW": "S-1-3-4"} +EVERYONE_SID = "S-1-1-0" +USERS_SID = "S-1-5-32-545" +TEMP_PREFIX = "jraphyte-fence-strict-" +OPEN_FOR_WRITE = "from pathlib import Path; import sys; Path(sys.argv[1]).open('a+b').close()" + + +class StrictAclContractTests(unittest.TestCase): + def file_ok(self, original, current): + return guarded._only_admission_dacl_added(original, current, ALIAS) + + def parent_ok(self, original, current): + return guarded._only_parent_admission_added(original, current, ALIAS) + + def test_identical_aces_with_only_the_admission_deny_and_ai_are_accepted(self): + self.assertTrue(self.file_ok(FILE_ORIGINAL, "D:AI(D;;DCLC;;;BU)" + FILE_ORIGINAL[2:])) + self.assertTrue(self.parent_ok(PARENT_ORIGINAL, "D:AI(D;;LC;;;BU)" + PARENT_ORIGINAL[2:])) + self.assertTrue(guarded._same_effective_dacl(FILE_ORIGINAL, "D:AI" + FILE_ORIGINAL[2:])) + + def test_explicit_copies_of_inherited_allows_are_rejected_everywhere(self): + # The hosted-runner shape. Explicit allows outlive revocation of the + # inherited grant, so this is a different ACL, not a representation. + self.assertFalse(self.file_ok(FILE_ORIGINAL, FILE_WITH_COPIES)) + self.assertFalse(self.parent_ok(PARENT_ORIGINAL, PARENT_WITH_COPIES)) + after_remove = FILE_WITH_COPIES.replace("(D;;DCLC;;;BU)", "") + self.assertFalse(guarded._same_effective_dacl(FILE_ORIGINAL, after_remove)) + parent_after_remove = PARENT_WITH_COPIES.replace("(D;;LC;;;BU)", "") + self.assertFalse(guarded._same_effective_dacl(PARENT_ORIGINAL, parent_after_remove)) + + def test_explicit_and_inherited_spellings_of_one_allow_are_different_aces(self): + explicit, inherited = "D:(A;;FA;;;SY)", "D:(A;ID;FA;;;SY)" + self.assertNotEqual(guarded._aces(explicit), guarded._aces(inherited)) + self.assertFalse(guarded._same_effective_dacl(explicit, inherited)) + self.assertFalse(guarded._same_effective_dacl(inherited, explicit)) + self.assertFalse(self.file_ok(inherited, "D:AI(D;;DCLC;;;BU)(A;;FA;;;SY)")) + self.assertFalse(self.parent_ok("D:(A;OICIID;FA;;;SY)", "D:AI(D;;LC;;;BU)(A;OICI;FA;;;SY)")) + + def test_any_changed_count_flag_order_or_deny_is_rejected(self): + for original, now in ( + ("D:(A;ID;FR;;;SY)", "D:AI(D;;DCLC;;;BU)(A;;FA;;;SY)(A;ID;FR;;;SY)"), # extra, wider + (FILE_ORIGINAL, "D:AI(D;;DCLC;;;BU)" + FILE_ORIGINAL[2:] + "(A;;FR;;;WD)"), # new trustee + ("D:(A;ID;FA;;;SY)", "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;SY)"), # duplicated + (FILE_ORIGINAL, "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;BA)"), # lost + (FILE_ORIGINAL, + "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;BA)(A;ID;FA;;;SY)(A;ID;FA;;;OW)"), # reordered + (FILE_ORIGINAL, "D:PAI(D;;DCLC;;;BU)" + FILE_ORIGINAL[2:]), # P changed + (FILE_ORIGINAL, "D:AI(D;;DC;;;BU)" + FILE_ORIGINAL[2:]), # wrong mask + (FILE_ORIGINAL, "D:AI(D;;DCLC;;;BG)" + FILE_ORIGINAL[2:]), # wrong trustee + (FILE_ORIGINAL, "D:AI(D;OI;DCLC;;;BU)" + FILE_ORIGINAL[2:]), # wrong flags + (FILE_ORIGINAL, "D:AI(D;;DCLC;;;BU)(D;;DCLC;;;BU)" + FILE_ORIGINAL[2:])): # two denies + self.assertFalse(self.file_ok(original, now), now) + for now in (PARENT_ORIGINAL.replace("D:", "D:AI(D;;LC;;;BU)(A;OICI;FR;;;WD)", 1), + "D:AI(D;;LC;;;BG)" + PARENT_ORIGINAL[2:], + "D:AI(D;;DCLC;;;BU)" + PARENT_ORIGINAL[2:]): + self.assertFalse(self.parent_ok(PARENT_ORIGINAL, now), now) + + def test_explicit_copy_detector_flags_only_explicit_twins_of_inherited_allows(self): + self.assertEqual(len(_explicit_twins(FILE_WITH_COPIES)), 3) + self.assertEqual(len(_explicit_twins(PARENT_WITH_COPIES)), 3) + self.assertEqual(_explicit_twins(FILE_ORIGINAL), []) + self.assertEqual(_explicit_twins(PARENT_ORIGINAL), []) + # An explicit allow with no inherited twin, and a deny, are not copies. + self.assertEqual(_explicit_twins("D:AI(D;;DCLC;;;BU)(A;;FR;;;WD)(A;ID;FA;;;SY)"), []) + + +def _call(*args: str) -> subprocess.CompletedProcess: + return subprocess.run(list(args), capture_output=True, text=True, timeout=30) + + +def _icacls(*args: str) -> None: + result = _call("icacls.exe", *args) + assert result.returncode == 0, result.stderr or result.stdout + + +def _inherited_twin(ace: str) -> str: + """Inherited spelling of an explicit allow ACE (test-local, independent of the guard).""" + fields = ace[1:-1].split(";") + return "(" + ";".join(fields[:1] + [fields[1] + "ID"] + fields[2:]) + ")" if fields[0] == "A" else "" + + +def _explicit_twins(sddl: str) -> list[str]: + """Explicit ACEs whose inherited spelling is also present (test-local, independent of the guard).""" + aces = guarded._aces(sddl) + return [ace for ace in aces if _inherited_twin(ace) and _inherited_twin(ace) in aces] + + +def _grant_explicit_copies(path: Path) -> None: + """Real icacls grants that build the hosted shape on an owned fixture (a shape, not a cause).""" + # The fixture is declared before it is changed: copies must be introduced here, + # not already present, or the case would prove nothing. + assert not _explicit_twins(guarded.directory_dacl_sddl(path)), "fixture baseline already has explicit copies" + inherited = [ace.split(";")[5].rstrip(")") for ace in guarded._aces(guarded.directory_dacl_sddl(path)) + if ace.startswith("(A;") and "ID" in ace.split(";")[1]] + wanted = [alias for alias in TRUSTEE_SIDS if alias in inherited] + assert wanted, "no inherited SY/BA/OW allow to copy" + rights = "(OI)(CI)(F)" if path.is_dir() else "(F)" + # icacls prepends each grant, so pass them reversed to keep SY, BA, OW order. + _icacls(str(path), "/grant", *["*" + TRUSTEE_SIDS[alias] + ":" + rights for alias in reversed(wanted)]) + assert _explicit_twins(guarded.directory_dacl_sddl(path)), "hosted shape not reached" + + +def _grant_unexpected_explicit_allow(path: Path) -> None: + """Add one explicit allow ACE that the pinned DACL does not contain (a real icacls grant).""" + before = guarded._aces(guarded.directory_dacl_sddl(path)) + if path.is_dir(): + # Same trustee and rights as a pinned inheritable allow, minus the inheritance flags. + _icacls(str(path), "/grant", "*" + TRUSTEE_SIDS["SY"] + ":(F)") + else: + # A file has no flag variants of an allow; a new reviewed-looking trustee is the copy. + _icacls(str(path), "/grant", "*" + USERS_SID + ":(R)") + after = guarded._aces(guarded.directory_dacl_sddl(path)) + added = [ace for ace in after if ace not in before] + assert len(after) == len(before) + 1 and len(added) == 1 and added[0].startswith("(A;"), \ + "the unexpected explicit allow was not added as exactly one new ACE" + + +@unittest.skipUnless(os.name == "nt", "Windows ACL contract") +class StrictAclFenceTests(unittest.TestCase): + def setUp(self): + self.sid = guarded._current_sid() + self.temp = tempfile.TemporaryDirectory(prefix=TEMP_PREFIX) + self.addCleanup(self._cleanup) + self.base = Path(self.temp.name) + self.root = self.base / "old-app" + self.old = self.root / "state" + self.old.mkdir(parents=True) + self.archive = self.base / "archive" / "state" + self.archive.parent.mkdir() + self.receipts = self.base / "receipts" + self.receipts.mkdir() + self.content = {} + for name in guarded.NAMES: + body = b"0" if name == "controller.lock" else ("strict:" + name).encode() + (self.old / name).write_bytes(body) + self.content[name] = hashlib.sha256(body).hexdigest() + for name in ("source-stage.lock", "phase-stage.lock"): + (self.root / name).write_bytes(b"0") + windows_fixture_acl.stabilize_owned_fixture( + self.base, [("parent", self.root), ("state", self.old)] + + [("file:" + name, self.old / name) for name in sorted(guarded.NAMES)], + read=guarded.directory_dacl_sddl) + self.pin() + + def pin(self): + """Declare the source ACL: pins are taken only after the fixture ACL is final.""" + self.original_parent = guarded.directory_dacl_sddl(self.root) + self.original_files = {name: guarded.directory_dacl_sddl(self.old / name) + for name in guarded.NAMES} + self.parent_acl_sha = hashlib.sha256(self.original_parent.encode()).hexdigest() + self.file_acl_sha = {name: hashlib.sha256(value.encode()).hexdigest() + for name, value in self.original_files.items()} + + def _cleanup(self): + targets = [self.root] + [base / name for base in (self.old, self.archive) + for name in guarded.NAMES] + for target in targets: + if target.exists(): + _call("icacls.exe", str(target), "/remove:d", "*" + self.sid) + self.temp.cleanup() + self.assertFalse(self.base.exists()) + + def invoke(self): + return guarded.fence_closed_old_phase_guarded( + old_root=self.old, archived_root=self.archive, + expected_file_sha256=self.content, + source_stage_lock=self.root / "source-stage.lock", + phase_stage_lock=self.root / "phase-stage.lock", deny_sid=self.sid, + expected_parent_dacl_sha256=self.parent_acl_sha, + expected_file_dacl_sha256=self.file_acl_sha, + barrier_receipt_path=self.receipts / "barrier.json", + fence_receipt_path=self.receipts / "fence.json") + + def after_guard_deny(self, then): + """Wrap the guard's real icacls: after each real deny, apply ``then(path)``.""" + actual = guarded._icacls + + def wrapped(path, *args): + actual(path, *args) + if "/deny" in args: + then(Path(path)) + return patch.object(guarded, "_icacls", side_effect=wrapped) + + def contender_denied(self, name): + return _call(sys.executable, "-c", OPEN_FOR_WRITE, str(self.old / name)).returncode != 0 + + def contenders_denied_before_rename(self): + seen = {} + actual_rename = guarded.os.rename + + def observe(source, destination): + for name in guarded.NAMES: + seen[name] = self.contender_denied(name) + return actual_rename(source, destination) + return seen, patch.object(guarded.os, "rename", side_effect=observe) + + def assert_failed_closed(self, message, *, expect_denied_controller=True): + """Nothing moved, no completion receipt, and the real admission barrier still holds.""" + self.assertTrue(self.old.exists()) + self.assertFalse(self.archive.exists()) + self.assertFalse((self.receipts / "fence.json").exists()) + if expect_denied_controller: + self.assertTrue(guarded._file_deny(self.old / "controller.lock", self.sid)) + self.assertTrue(self.contender_denied("controller.lock")) + + def test_declared_explicit_and_inherited_source_is_preserved_exactly(self): + # Fixture ACL declared (and verified) before pinning: explicit + inherited allows. + # Reset to inheritance-only, parent first, then add the explicit copies. + for path in (self.root, self.old, *(self.old / name for name in guarded.NAMES)): + _icacls(str(path), "/reset") + _grant_explicit_copies(self.root) + for name in guarded.NAMES: + _grant_explicit_copies(self.old / name) + self.pin() + self.assertTrue(_explicit_twins(self.original_parent)) + self.assertTrue(all(_explicit_twins(value) for value in self.original_files.values())) + seen, rename = self.contenders_denied_before_rename() + with rename: + result = self.invoke() + self.assertEqual(seen, {name: True for name in guarded.NAMES}) + self.assertEqual(result["status"], "OLD_PATH_FENCED_NEW_PHASE_ALLOWED") + # Exact preservation: same ACE count, flags and order; only the parent deny added. + deny = guarded._canonical_ace(guarded.DENY_PARENT_ACE.format(sid=self.sid)) + parent_now = guarded._aces(guarded.directory_dacl_sddl(self.root)) + self.assertIn(deny, parent_now) + parent_now.remove(deny) + self.assertEqual(parent_now, guarded._aces(self.original_parent)) + for name in guarded.NAMES: + self.assertEqual(guarded._aces(guarded.directory_dacl_sddl(self.archive / name)), + guarded._aces(self.original_files[name]), name) + self.assertFalse(guarded._file_deny(self.archive / name, self.sid)) + self.assertEqual({name: hashlib.sha256((self.archive / name).read_bytes()).hexdigest() + for name in guarded.NAMES}, self.content) + + def test_unexpected_explicit_allow_copy_on_the_parent_fails_closed_before_the_rename(self): + # The deterministic baseline is protected and explicit; the copy is injected after the + # guard's own deny, so it is the only change besides the deny. + self.assertEqual(windows_fixture_acl.fixture_acl_problems(self.original_parent, True), []) + + def copy_on_parent(path): + if path == self.root: + _grant_unexpected_explicit_allow(path) + with self.after_guard_deny(copy_on_parent), \ + patch.object(guarded.os, "rename", wraps=os.rename) as rename: + with self.assertRaisesRegex(Exception, "old parent DACL broadened"): + self.invoke() + rename.assert_not_called() + self.assert_failed_closed("parent") + + def test_unexpected_explicit_allow_on_a_denied_file_fails_closed_on_replay(self): + self.assertEqual(windows_fixture_acl.fixture_acl_problems(self.original_files["controller.lock"], False), []) + calls = {"count": 0} + + def copies_then_crash(path): + _grant_unexpected_explicit_allow(path) + calls["count"] += 1 + if calls["count"] == 1: + raise RuntimeError("injected post-controller-deny crash") + with self.after_guard_deny(copies_then_crash): + with self.assertRaisesRegex(RuntimeError, "injected"): + self.invoke() + with patch.object(guarded.os, "rename", wraps=os.rename) as rename: + with self.assertRaisesRegex(Exception, "unreviewed old file DACL drift"): + self.invoke() + rename.assert_not_called() + self.assert_failed_closed("file") + + def test_real_parent_broadening_after_the_guard_deny_fails_closed(self): + def broaden_parent(path): + if path == self.root: + _icacls(str(path), "/grant", "*" + EVERYONE_SID + ":(R)") + with self.after_guard_deny(broaden_parent), \ + patch.object(guarded.os, "rename", wraps=os.rename) as rename: + with self.assertRaisesRegex(Exception, "old parent DACL broadened"): + self.invoke() + rename.assert_not_called() + self.assert_failed_closed("parent") + + def test_real_file_broadening_is_rejected_on_replay_before_the_rename(self): + calls = {"count": 0} + + def broaden_then_crash(path): + calls["count"] += 1 + if calls["count"] == 1: + _icacls(str(path), "/grant", "*" + EVERYONE_SID + ":(R)") + raise RuntimeError("injected post-controller-deny crash") + with self.after_guard_deny(broaden_then_crash): + with self.assertRaisesRegex(RuntimeError, "injected"): + self.invoke() + with patch.object(guarded.os, "rename", wraps=os.rename) as rename: + with self.assertRaisesRegex(Exception, "unreviewed old file DACL drift"): + self.invoke() + rename.assert_not_called() + self.assert_failed_closed("file") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_paper_pilot_phase_cutover_guarded_v2_trustees.py b/tests/test_paper_pilot_phase_cutover_guarded_v2_trustees.py new file mode 100644 index 0000000..7e0e7a3 --- /dev/null +++ b/tests/test_paper_pilot_phase_cutover_guarded_v2_trustees.py @@ -0,0 +1,343 @@ +"""Windows SID/SDDL trustee-alias regressions for the guarded admission barrier. + +Windows serializes some numeric SIDs in a DACL string as SDDL aliases (the +built-in Administrator, RID 500, is ``LA``; the Users group S-1-5-32-545 is +``BU``). The barrier must recognize its own deny ACE under either spelling while +still rejecting a different trustee, mask, inheritance flag, ACE order or extra +ACE. These cases are a separate hosted gate from the nine original fence cases. +""" +from __future__ import annotations + +import hashlib +import os +from pathlib import Path +import re +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +from src import paper_pilot_phase_cutover_guarded_v2 as guarded +from src.paper_pilot_phase_cutover import _current_sid +from tools import windows_fixture_acl + + +USERS_SID = "S-1-5-32-545" # serialized by Windows as the alias BU +GUESTS_SID = "S-1-5-32-546" # an unrelated trustee, serialized as BG +ADMINS_SID = "S-1-5-32-544" # an unrelated trustee, serialized as BA +TEMP_PREFIX = "jraphyte-fence-trustee-" + + +def _icacls(*args: str) -> None: + result = subprocess.run(["icacls.exe", *args], capture_output=True, text=True, timeout=30) + assert result.returncode == 0, result.stderr or result.stdout + + +FOREIGN_RID500_SID = "S-1-5-21-1111111111-2222222222-3333333333-500" # fabricated domain + + +def _account_prefix() -> str: + prefix = re.fullmatch(r"(S-1-5-21-\d+-\d+-\d+)-\d+", _current_sid()) + assert prefix is not None, "test requires a machine/domain account SID" + return prefix.group(1) + + +def _rid500_sid() -> str: + # The machine account domain's built-in Administrator; derived from the + # runner's own SID so no account lookup or account change is needed. + return _account_prefix() + "-500" + + +def _plain_account_sid() -> str: + # A same-domain account that has no SDDL alias (never RID 500). + current = _current_sid() + return current if not current.endswith("-500") else _account_prefix() + "-1001" + + +@unittest.skipUnless(os.name == "nt", "Windows SID/SDDL contract") +class TrusteeSemanticsTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(prefix=TEMP_PREFIX) + self.addCleanup(self._cleanup) + self.file = Path(self.temp.name) / "probe.bin" + self.file.write_bytes(b"0") + + def _cleanup(self): + for sid in (USERS_SID, GUESTS_SID, ADMINS_SID, _rid500_sid(), FOREIGN_RID500_SID): + subprocess.run(["icacls.exe", str(self.file), "/remove:d", "*" + sid], + capture_output=True, text=True, timeout=30) + self.temp.cleanup() + self.assertFalse(Path(self.temp.name).exists()) + + def test_canonical_trustee_uses_windows_serialization(self): + self.assertEqual(guarded._canonical_trustee(USERS_SID), "BU") + self.assertEqual(guarded._canonical_trustee("S-1-5-18"), "SY") + self.assertEqual(guarded._canonical_trustee("S-1-1-0"), "WD") + # Aliases, SIDs without an alias and non-SID text pass through unchanged. + self.assertEqual(guarded._canonical_trustee("BU"), "BU") + self.assertEqual(guarded._canonical_trustee("S-1-5-21-1-2-3-1001"), + "S-1-5-21-1-2-3-1001") + injected = "S-1-5-18)(A;;FA;;;WD" + self.assertEqual(guarded._canonical_trustee(injected), injected) + + def test_real_dacl_alias_matches_numeric_deny_and_nothing_else(self): + _icacls(str(self.file), "/deny", "*" + USERS_SID + ":(WD,AD)") + raw = guarded.directory_dacl_sddl(self.file) + # The OS really does serialize the numeric SID as the alias, so a + # literal numeric comparison cannot see this ACE. + self.assertIn("(D;;DCLC;;;BU)", raw) + self.assertNotIn(USERS_SID, raw) + self.assertNotIn(guarded.DENY_FILE_ACE.format(sid=USERS_SID), raw) + self.assertTrue(guarded._file_deny(self.file, USERS_SID)) + self.assertTrue(guarded._owner_deny_present(raw, USERS_SID)) + self.assertFalse(guarded._file_deny(self.file, ADMINS_SID)) + self.assertFalse(guarded._file_deny(self.file, GUESTS_SID)) + self.assertFalse(guarded._owner_deny_present(raw, ADMINS_SID)) + + def test_machine_administrator_rid500_deny_is_recognized(self): + sid = _rid500_sid() + self.assertFalse(guarded._file_deny(self.file, sid)) + _icacls(str(self.file), "/deny", "*" + sid + ":(WD,AD)") + raw = guarded.directory_dacl_sddl(self.file) + # Whichever spelling Windows chose (numeric, or LA on the runner whose + # own identity is RID 500) the exact barrier ACE must be recognized. + self.assertTrue(guarded._file_deny(self.file, sid), raw) + self.assertTrue(guarded._owner_deny_present(raw, sid), raw) + + def test_foreign_domain_rid500_is_not_collapsed_into_local_administrator(self): + local, foreign, plain = _rid500_sid(), FOREIGN_RID500_SID, _plain_account_sid() + self.assertNotEqual(local, foreign) + canonical = {sid: guarded._canonical_trustee(sid) for sid in (local, foreign, plain)} + # In-memory Windows conversion only: a SID from another domain that + # shares RID 500 gets no alias, and neither does a plain account. + self.assertEqual(canonical[foreign], foreign) + self.assertEqual(canonical[plain], plain) + self.assertNotIn(canonical[foreign], ("LA", canonical[local])) + self.assertEqual(len(set(canonical.values())), 3, canonical) + # Same RID 500 and same first two domain sub-authorities, different last one. + near = _account_prefix().rsplit("-", 1)[0] + "-" + str(int(_account_prefix().rsplit("-", 1)[1]) + 1) + "-500" + self.assertEqual(guarded._canonical_trustee(near), near) + self.assertNotEqual(guarded._canonical_trustee(near), canonical[local]) + + def test_ace_matching_never_crosses_local_foreign_and_plain_accounts(self): + sids = {"local": _rid500_sid(), "foreign": FOREIGN_RID500_SID, "plain": _plain_account_sid()} + for ace_name, ace_sid in sids.items(): + # Both spellings Windows may use: the serialized (alias) form and the numeric form. + for spelling in (guarded._canonical_trustee(ace_sid), ace_sid): + file_deny = "D:AI(D;;DCLC;;;" + spelling + ")(A;ID;FA;;;SY)" + parent_deny = "D:AI(D;;LC;;;" + spelling + ")(A;OICIID;FA;;;SY)" + for query_name, query_sid in sids.items(): + same = ace_name == query_name + label = (ace_name, spelling, query_name) + self.assertEqual(guarded._has_ace(file_deny, guarded.DENY_FILE_ACE, query_sid), + same, label) + self.assertEqual(guarded._has_ace(parent_deny, guarded.DENY_PARENT_ACE, query_sid), + same, label) + self.assertEqual(guarded._owner_deny_present(file_deny, query_sid), same, label) + self.assertEqual(guarded._only_admission_dacl_added( + "D:(A;ID;FA;;;SY)", file_deny, query_sid), same, label) + self.assertEqual(guarded._only_parent_admission_added( + "D:(A;OICIID;FA;;;SY)", parent_deny, query_sid), same, label) + # An effective-DACL comparison must not equate two different RID-500 trustees. + self.assertFalse(guarded._same_effective_dacl( + "D:(D;;DCLC;;;" + guarded._canonical_trustee(sids["local"]) + ")", + "D:(D;;DCLC;;;" + FOREIGN_RID500_SID + ")")) + self.assertTrue(guarded._same_effective_dacl( + "D:(D;;DCLC;;;" + guarded._canonical_trustee(sids["local"]) + ")", + "D:AI(D;;DCLC;;;" + sids["local"] + ")")) + + def test_real_local_rid500_deny_does_not_match_foreign_or_plain_accounts(self): + # icacls cannot install an unresolvable foreign-domain SID, so the real + # DACL leg uses the machine's own RID 500 and queries the other two. + sid = _rid500_sid() + _icacls(str(self.file), "/deny", "*" + sid + ":(WD,AD)") + raw = guarded.directory_dacl_sddl(self.file) + self.assertTrue(guarded._file_deny(self.file, sid), raw) + for other in (FOREIGN_RID500_SID, _plain_account_sid(), USERS_SID): + if other == sid: # the runner itself is the plain account only when not RID 500 + continue + self.assertFalse(guarded._file_deny(self.file, other), (other, raw)) + self.assertFalse(guarded._owner_deny_present(raw, other), (other, raw)) + + def test_numeric_and_alias_spellings_are_equivalent_in_sddl_text(self): + alias = "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;SY)" + numeric = "D:AI(D;;DCLC;;;" + USERS_SID + ")(A;ID;FA;;;SY)" + for text in (alias, numeric): + self.assertTrue(guarded._has_ace(text, guarded.DENY_FILE_ACE, USERS_SID)) + self.assertTrue(guarded._owner_deny_present(text, USERS_SID)) + self.assertTrue(guarded._only_admission_dacl_added( + "D:(A;ID;FA;;;SY)", alias, USERS_SID)) + self.assertTrue(guarded._only_admission_dacl_added( + "D:(A;ID;FA;;;SY)", numeric, USERS_SID)) + parent = "D:AI(D;;LC;;;BU)(A;OICIID;FA;;;SY)" + self.assertTrue(guarded._only_parent_admission_added( + "D:(A;OICIID;FA;;;SY)", parent, USERS_SID)) + self.assertTrue(guarded._same_effective_dacl( + "D:(D;;DCLC;;;BU)(A;ID;FA;;;SY)", "D:AI(D;;DCLC;;;" + USERS_SID + ")(A;ID;FA;;;SY)")) + + def test_wrong_trustee_mask_flags_or_type_are_rejected(self): + for text in ( + "D:(D;;DCLC;;;BA)", # different trustee + "D:(D;;DCLC;;;BG)", + "D:(D;;DC;;;BU)", # narrower mask + "D:(D;;DCLCWP;;;BU)", # broader mask + "D:(D;OI;DCLC;;;BU)", # inheritance flag added + "D:(D;ID;DCLC;;;BU)", + "D:(A;;DCLC;;;BU)", # allow, not deny + "D:(OD;;DCLC;;;BU)"): # object deny + self.assertFalse(guarded._has_ace(text, guarded.DENY_FILE_ACE, USERS_SID), text) + self.assertFalse(guarded._has_ace("D:(D;;LC;;;BU)", guarded.DENY_FILE_ACE, USERS_SID)) + self.assertFalse(guarded._has_ace("D:(D;;DCLC;;;BU)", guarded.DENY_PARENT_ACE, USERS_SID)) + # An allow ACE or a deny of another trustee is not an owner deny. + self.assertFalse(guarded._owner_deny_present("D:(A;;FA;;;BU)", USERS_SID)) + self.assertFalse(guarded._owner_deny_present("D:(D;;FX;;;BA)", USERS_SID)) + self.assertTrue(guarded._owner_deny_present("D:(D;;FX;;;BU)", USERS_SID)) + + def test_only_the_exact_deny_may_be_added_in_original_order(self): + original = "D:(A;ID;FA;;;SY)(A;ID;FA;;;BA)" + good = "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;BA)" + self.assertTrue(guarded._only_admission_dacl_added(original, good, USERS_SID)) + for current in ( + "D:AI(D;;DCLC;;;BG)(A;ID;FA;;;SY)(A;ID;FA;;;BA)", # wrong trustee + "D:AI(D;;DC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;BA)", # wrong mask + "D:AI(D;OI;DCLC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;BA)", # wrong flags + "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;BA)(A;ID;FA;;;SY)", # reordered + "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;BA)(A;;FR;;;WD)", # extra access + "D:AI(D;;DCLC;;;BU)(A;ID;FA;;;SY)", # removed access + "D:PAI(D;;DCLC;;;BU)(A;ID;FA;;;SY)(A;ID;FA;;;BA)"): # protection changed + self.assertFalse(guarded._only_admission_dacl_added(original, current, USERS_SID), + current) + self.assertFalse(guarded._only_parent_admission_added( + "D:(A;OICIID;FA;;;SY)", "D:AI(D;;DC;;;BU)(A;OICIID;FA;;;SY)", USERS_SID)) + + +@unittest.skipUnless(os.name == "nt", "Windows SID/SDDL contract") +class AliasIdentityFenceTests(unittest.TestCase): + """Run the real barrier as an identity whose SDDL trustee is an alias (BU).""" + + def setUp(self): + self.sid = USERS_SID + patcher = patch.object(guarded, "_current_sid", return_value=self.sid) + patcher.start() + self.addCleanup(patcher.stop) + self.temp = tempfile.TemporaryDirectory(prefix=TEMP_PREFIX) + self.addCleanup(self._cleanup) + self.base = Path(self.temp.name) + self.root = self.base / "old-app" + self.old = self.root / "state" + self.old.mkdir(parents=True) + self.archive = self.base / "archive" / "state" + self.archive.parent.mkdir() + self.receipts = self.base / "receipts" + self.receipts.mkdir() + self.content = {} + for name in guarded.NAMES: + body = b"0" if name == "controller.lock" else ("alias:" + name).encode() + (self.old / name).write_bytes(body) + self.content[name] = hashlib.sha256(body).hexdigest() + for name in ("source-stage.lock", "phase-stage.lock"): + (self.root / name).write_bytes(b"0") + # Explicit, protected and verified before the first pin (see tools/windows_fixture_acl.py). + windows_fixture_acl.stabilize_owned_fixture( + self.base, [("parent", self.root), ("state", self.old)] + + [("file:" + name, self.old / name) for name in sorted(guarded.NAMES)], + read=guarded.directory_dacl_sddl) + self._pin() + + def _pin(self): + self.parent_acl_sha = hashlib.sha256( + guarded.directory_dacl_sddl(self.root).encode()).hexdigest() + self.file_acl_sha = {name: hashlib.sha256( + guarded.directory_dacl_sddl(self.old / name).encode()).hexdigest() + for name in guarded.NAMES} + + def _cleanup(self): + for target in (self.root, *(base / name for base in (self.old, self.archive) + for name in guarded.NAMES)): + if target.exists(): + for sid in (USERS_SID, GUESTS_SID): + subprocess.run(["icacls.exe", str(target), "/remove:d", "*" + sid], + capture_output=True, text=True, timeout=30) + self.temp.cleanup() + self.assertFalse(self.base.exists()) + + def invoke(self): + return guarded.fence_closed_old_phase_guarded( + old_root=self.old, archived_root=self.archive, + expected_file_sha256=self.content, + source_stage_lock=self.root / "source-stage.lock", + phase_stage_lock=self.root / "phase-stage.lock", deny_sid=self.sid, + expected_parent_dacl_sha256=self.parent_acl_sha, + expected_file_dacl_sha256=self.file_acl_sha, + barrier_receipt_path=self.receipts / "barrier.json", + fence_receipt_path=self.receipts / "fence.json") + + def test_fence_completes_when_trustee_is_serialized_as_alias(self): + result = self.invoke() + self.assertEqual(result["status"], "OLD_PATH_FENCED_NEW_PHASE_ALLOWED") + self.assertFalse(self.old.exists()) + self.assertEqual({name: hashlib.sha256((self.archive / name).read_bytes()).hexdigest() + for name in guarded.NAMES}, self.content) + # The parent deny stays durable, spelled by Windows as the alias. + self.assertIn("(D;;LC;;;BU)", guarded.directory_dacl_sddl(self.root)) + # Archived per-file denies were restored away. + for name in guarded.NAMES: + self.assertFalse(guarded._file_deny(self.archive / name, self.sid)) + self.assertEqual(result, self.invoke()) + + def test_crash_after_first_alias_deny_replays(self): + actual = guarded._icacls + calls = {"count": 0} + + def first_deny_then_crash(path, *args): + actual(path, *args) + calls["count"] += 1 + if calls["count"] == 1: + raise RuntimeError("injected post-controller-deny crash") + with patch.object(guarded, "_icacls", side_effect=first_deny_then_crash): + with self.assertRaisesRegex(RuntimeError, "injected"): + self.invoke() + self.assertIn("(D;;DCLC;;;BU)", + guarded.directory_dacl_sddl(self.old / "controller.lock")) + self.assertTrue(guarded._file_deny(self.old / "controller.lock", self.sid)) + self.assertEqual(self.invoke()["status"], "OLD_PATH_FENCED_NEW_PHASE_ALLOWED") + + def test_preexisting_alias_file_deny_is_preserved_and_rejected(self): + path = self.old / "checkpoint.sqlite3" + _icacls(str(path), "/deny", "*" + self.sid + ":(X)") + prior = guarded.directory_dacl_sddl(path) + self.assertIn(";;;BU)", prior) + self.file_acl_sha[path.name] = hashlib.sha256(prior.encode()).hexdigest() + with self.assertRaisesRegex(Exception, "existing owner deny"): + self.invoke() + self.assertEqual(guarded.directory_dacl_sddl(path), prior) + self.assertFalse((self.receipts / "barrier.json").exists()) + self.assertTrue(self.old.exists()) + + def test_preexisting_alias_parent_deny_is_preserved_and_rejected(self): + _icacls(str(self.root), "/deny", "*" + self.sid + ":(X)") + prior = guarded.directory_dacl_sddl(self.root) + self.assertIn(";;;BU)", prior) + # Windows re-propagates inheritance to children when the parent ACL + # changes (adding AI), so re-pin them: only the owner deny may reject. + self._pin() + with self.assertRaisesRegex(Exception, "existing owner deny"): + self.invoke() + self.assertEqual(guarded.directory_dacl_sddl(self.root), prior) + self.assertFalse((self.receipts / "barrier.json").exists()) + + def test_preexisting_deny_for_a_different_trustee_is_not_an_owner_deny(self): + path = self.old / "checkpoint.sqlite3" + _icacls(str(path), "/deny", "*" + GUESTS_SID + ":(X)") + prior = guarded.directory_dacl_sddl(path) + self.assertIn(";;;BG)", prior) + self.file_acl_sha[path.name] = hashlib.sha256(prior.encode()).hexdigest() + self.assertEqual(self.invoke()["status"], "OLD_PATH_FENCED_NEW_PHASE_ALLOWED") + # The unrelated deny is preserved byte-for-byte in the archive. + self.assertTrue(guarded._same_effective_dacl( + prior, guarded.directory_dacl_sddl(self.archive / path.name))) + self.assertIn(";;;BG)", guarded.directory_dacl_sddl(self.archive / path.name)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_windows_fence_diagnostics.py b/tests/test_windows_fence_diagnostics.py new file mode 100644 index 0000000..bb7f98e --- /dev/null +++ b/tests/test_windows_fence_diagnostics.py @@ -0,0 +1,366 @@ +"""Failure-path coverage for the sanitized hosted-CI diagnostics tool.""" +from __future__ import annotations + +import importlib.util +import io +import json +import os +from pathlib import Path +import subprocess +import tempfile +import unittest +from contextlib import redirect_stdout +from unittest.mock import patch + +from trace_gc.errors import ContractError + +ROOT = Path(__file__).resolve().parents[1] +_SPEC = importlib.util.spec_from_file_location( + "windows_fence_diagnostics_under_test", ROOT / "tools" / "windows_fence_diagnostics.py") +diagnostics = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(diagnostics) + +SECRET_TEXT = r"C:\Users\private-account\secret-dir\file.bin S-1-5-21-111-222-333-1001" + + +class SanitizationTests(unittest.TestCase): + def test_redact_masks_machine_sub_authorities_only(self): + self.assertEqual(diagnostics.redact("D:(D;;DCLC;;;S-1-5-21-1-2-3-500)(A;;FA;;;SY)"), + "D:(D;;DCLC;;;S-1-5-21--500)(A;;FA;;;SY)") + + def test_error_description_never_carries_message_text(self): + for exc in (ValueError(SECRET_TEXT), OSError(2, SECRET_TEXT, SECRET_TEXT), + ContractError("PHASE_FENCE", SECRET_TEXT, path=SECRET_TEXT), + subprocess.CalledProcessError(5, ["icacls.exe", SECRET_TEXT])): + text = json.dumps(diagnostics.describe_error(exc)) + self.assertNotIn("private-account", text) + self.assertNotIn("secret-dir", text) + self.assertNotIn("S-1-5-21", text) + self.assertEqual(diagnostics.describe_error(ContractError("PHASE_FENCE", SECRET_TEXT)), + {"type": "ContractError", "category": "PHASE_FENCE"}) + self.assertEqual(diagnostics.describe_error( + subprocess.CalledProcessError(5, ["icacls.exe"])), + {"type": "CalledProcessError", "category": "unspecified", "exit_code": 5}) + + def test_error_category_is_bounded_to_a_code_shaped_string(self): + class Odd(Exception): + code = SECRET_TEXT + self.assertEqual(diagnostics.describe_error(Odd()), + {"type": "Odd", "category": "unspecified"}) + + def test_outcome_emits_only_fixed_allowlisted_codes_never_free_text(self): + for text in ("private-account internal project", "private account", "project path", + "old parent DACL broadened during fence and private-account", + "Old parent DACL broadened during fence", "", "x" * 400): + outcome = diagnostics._outcome(ContractError("PHASE_FENCE", text)) + self.assertEqual(outcome, {"type": "ContractError", "category": "PHASE_FENCE"}, text) + self.assertEqual(diagnostics._outcome(ContractError("PHASE_FENCE", "old parent DACL broadened during fence")), + {"type": "ContractError", "category": "PHASE_FENCE", + "detail_code": "PARENT_BROADENED_DURING_FENCE"}) + # A known constant message with any path attached is not trusted either. + self.assertNotIn("detail_code", diagnostics._outcome(ContractError( + "PHASE_FENCE", "old parent DACL broadened during fence", path="private-account"))) + for exc in (ValueError("old parent DACL broadened during fence"), + ContractError("PHASE_FENCE", SECRET_TEXT, path=SECRET_TEXT)): + rendered = json.dumps(diagnostics._outcome(exc)) + self.assertNotIn("detail", rendered.replace("detail_code", "")) + self.assertNotIn("private-account", rendered) + + def test_every_allowlisted_message_is_a_constant_of_the_guard_and_codes_are_fixed_shape(self): + source = (ROOT / "src" / "paper_pilot_phase_cutover_guarded_v2.py").read_text() + for message, code in diagnostics._DETAIL_CODES.items(): + self.assertIn('"' + message + '"', source, message) + self.assertRegex(code, r"[A-Z][A-Z_]{2,40}") + self.assertEqual(len(set(diagnostics._DETAIL_CODES.values())), len(diagnostics._DETAIL_CODES)) + + +@unittest.skipUnless(os.name == "nt", "Windows ACL diagnostics") +class ProbeCleanupTests(unittest.TestCase): + def setUp(self): + from src import paper_pilot_phase_cutover_guarded_v2 as guarded + self.guarded = guarded + self.real_dacl = guarded.directory_dacl_sddl + self.sid = guarded._current_sid() + self.at_exit = {} + test = self + + class Recording(tempfile.TemporaryDirectory): + def __exit__(self, *exc): + # Independent check, before the directory is removed: no owner + # deny may remain on the probe once the tool's own cleanup ran. + tree = Path(self.name) + if (tree / "old-app").exists(): + guarded = test.guarded + leftover = 0 + for target in [tree / "old-app"] + [tree / folder / name + for folder in ("old-app/state", "archive/state") + for name in guarded.NAMES]: + if target.exists(): + if guarded._owner_deny_present(test.real_dacl(target), test.sid): + leftover += 1 + subprocess.run(["icacls.exe", str(target), "/remove:d", "*" + test.sid], + capture_output=True, text=True, timeout=30) + test.at_exit["tree_denies"] = leftover + probe = Path(self.name) / "probe.bin" + if probe.exists(): + test.at_exit["deny_present"] = test.guarded._owner_deny_present( + test.real_dacl(probe), test.sid) + subprocess.run(["icacls.exe", str(probe), "/remove:d", "*" + test.sid], + capture_output=True, text=True, timeout=30) + return super().__exit__(*exc) + self.factory = Recording + + def _collect(self): + return diagnostics.collect(self.guarded, temp_factory=self.factory) + + def test_success_path_reports_and_verifies_cleanup(self): + report, cleanup_ok = self._collect() + self.assertNotIn("diagnostic_error", report, report) + self.assertTrue(report["barrier_recognizes_deny"]) + self.assertTrue(report["owner_deny_detected"]) + self.assertEqual(report["probe_cleanup"], + {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertTrue(cleanup_ok) + self.assertIs(self.at_exit["deny_present"], False) + rendered = json.dumps(report) + self.assertNotIn(tempfile.gettempdir(), rendered) + self.assertNotIn(os.environ.get("USERNAME", "\0"), rendered) + self.assertNotIn(self.sid.rsplit("-", 1)[0], rendered) # machine sub-authorities redacted + + def test_post_deny_dacl_query_failure_still_removes_probe_deny_without_leaking(self): + # collect() reads the probe DACL first before the deny (recorded separately), then after it. + calls = {"count": 0} + + def fail_second_read(path): + calls["count"] += 1 + if calls["count"] == 2: + raise RuntimeError(SECRET_TEXT) + return self.real_dacl(path) + with patch.object(self.guarded, "directory_dacl_sddl", side_effect=fail_second_read): + report, cleanup_ok = self._collect() + self.assertGreaterEqual(calls["count"], 3) + self.assertIn("probe_dacl_before_runner_deny", report) + self.assertNotIn("probe_before_error", report) + self.assertEqual(report["diagnostic_error"], + {"type": "RuntimeError", "category": "unspecified"}) + self.assertEqual(report["probe_cleanup"], + {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertTrue(cleanup_ok) + self.assertIs(self.at_exit["deny_present"], False) + self.assertNotIn("private-account", json.dumps(report)) + + def test_first_dacl_query_failure_is_recorded_separately_and_the_probe_continues(self): + calls = {"count": 0} + + def fail_first_read(path): + calls["count"] += 1 + if calls["count"] == 1: + raise RuntimeError(SECRET_TEXT) + return self.real_dacl(path) + with patch.object(self.guarded, "directory_dacl_sddl", side_effect=fail_first_read): + report, cleanup_ok = self._collect() + self.assertEqual(report["probe_before_error"], {"type": "RuntimeError", "category": "unspecified"}) + self.assertNotIn("probe_dacl_before_runner_deny", report) + self.assertNotIn("diagnostic_error", report) + self.assertTrue(report["barrier_recognizes_deny"]) + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertTrue(cleanup_ok) + self.assertIs(self.at_exit["deny_present"], False) + self.assertNotIn("private-account", json.dumps(report)) + + def test_recognition_failure_still_removes_probe_deny_without_leaking(self): + with patch.object(self.guarded, "_file_deny", + side_effect=ContractError("PHASE_FENCE", SECRET_TEXT, path=SECRET_TEXT)): + report, cleanup_ok = self._collect() + self.assertEqual(report["diagnostic_error"], + {"type": "ContractError", "category": "PHASE_FENCE"}) + self.assertTrue(cleanup_ok) + self.assertIs(self.at_exit["deny_present"], False) + self.assertNotIn("secret-dir", json.dumps(report)) + + def test_probe_directory_removal_failure_alone_is_not_clean_and_main_is_nonzero(self): + outer = self.factory + + class ProbeDirectoryFailsOnRemoval(outer): + def __init__(self, *args, prefix=None, **options): + super().__init__(*args, prefix=prefix, **options) + self.fails = prefix == "jraphyte-fence-diag-" + + def __exit__(self, *exc): + super().__exit__(*exc) + if self.fails: + raise OSError(SECRET_TEXT) + report, cleanup_ok = diagnostics.collect(self.guarded, temp_factory=ProbeDirectoryFailsOnRemoval) + self.assertFalse(cleanup_ok) + # Only the probe directory's removal failed: the probe deny and the trace are clean. + self.assertEqual(report["probe_cleanup"], {"icacls_remove_exit": 0, "deny_absent_verified": True}) + self.assertEqual(report["probe_directory_removal"], "failed") + self.assertEqual(report["probe_directory_removal_error"], {"type": "OSError", "category": "unspecified"}) + self.assertNotIn("diagnostic_error", report) + self.assertEqual(report["fence_trace"]["cleanup"]["directory_removal"], "ok") + self.assertEqual(report["fence_trace"]["cleanup"]["denies_remaining"], 0) + self.assertNotIn("private-account", json.dumps(report)) + buffer = io.StringIO() + with patch.object(diagnostics.collect, "__kwdefaults__", {"temp_factory": ProbeDirectoryFailsOnRemoval}), \ + redirect_stdout(buffer): + self.assertEqual(diagnostics.main(), 1) + self.assertEqual(json.loads(buffer.getvalue())["probe_directory_removal"], "failed") + + def test_failed_probe_cleanup_is_reported_and_nonzero(self): + real_run = subprocess.run + state = {"faked": False} + + def fake_remove_once(args, *a, **k): + if "/remove:d" in args and not state["faked"]: + state["faked"] = True + return subprocess.CompletedProcess(args, 5, "", "") + return real_run(args, *a, **k) + with patch.object(diagnostics.subprocess, "run", side_effect=fake_remove_once): + report, cleanup_ok = self._collect() + self.assertEqual(report["probe_cleanup"], + {"icacls_remove_exit": 5, "deny_absent_verified": False}) + self.assertFalse(cleanup_ok) + # The probe deny really was left for this case, and the harness removed it. + self.assertIs(self.at_exit["deny_present"], True) + + def test_trace_records_sanitized_dacl_observations_whatever_the_fence_outcome(self): + trace, ok = diagnostics.trace_fence(self.guarded, temp_factory=self.factory) + # The fence itself may legitimately fail on a host (that is what is being + # diagnosed); the observation structure and the verified cleanup may not. + self.assertTrue(ok, trace.get("cleanup")) + outcome = trace["outcome"] + self.assertTrue(outcome == {"status": "OLD_PATH_FENCED_NEW_PHASE_ALLOWED"} or "error" in outcome, + outcome) + labels = {"temp_parent", "base", "parent", "state"} | {"file:" + n for n in self.guarded.NAMES} + for phase in ("setup_as_created", "setup", "at_end", "after_remove_d"): + self.assertEqual(set(trace[phase]), labels, phase) + self.assertEqual(trace["fixture_acl"], {"status": "stabilized", "targets": 2 + len(self.guarded.NAMES)}) + for label in labels: + entry = trace["setup"][label] + self.assertTrue(entry["sddl"].startswith("D:"), label) + self.assertEqual(set(entry) & {"protected", "auto_inherited", "ace_count"}, + {"protected", "auto_inherited", "ace_count"}, label) + for label in ("parent", "state") + tuple("file:" + n for n in self.guarded.NAMES): + entry = trace["setup"][label] + self.assertEqual((entry["protected"], entry["inherited_allow"], entry["explicit_allow"], + entry["explicit_copies_of_inherited"]), (True, 0, 3, 0), label) + self.assertEqual(trace["cleanup"]["denies_remaining"], 0) + self.assertIs(self.at_exit["tree_denies"], 0) + rendered = json.dumps(trace) + self.assertNotIn(tempfile.gettempdir(), rendered) + self.assertNotIn(os.environ.get("USERNAME", ""), rendered) + self.assertNotIn(self.sid.rsplit("-", 1)[0], rendered) + + def test_trace_failure_after_real_denies_still_cleans_up_without_leaking(self): + sid = self.sid + + def real_denies_then_fail(**request): + # Real icacls denies through the tool's recording wrapper, then a failure, + # independent of whether the production fence would have got this far. + self.guarded._icacls(request["old_root"] / "controller.lock", "/deny", "*" + sid + ":(WD,AD)") + self.guarded._icacls(request["old_root"].parent, "/deny", "*" + sid + ":(AD)") + raise PermissionError(SECRET_TEXT) + with patch.object(self.guarded, "fence_closed_old_phase_guarded", + side_effect=real_denies_then_fail): + trace, ok = diagnostics.trace_fence(self.guarded, temp_factory=self.factory) + self.assertEqual(trace["outcome"], + {"error": {"type": "PermissionError", "category": "unspecified"}}) + self.assertEqual([(step["target"], step["operation"]) for step in trace["steps"]], + [("old:controller.lock", "deny"), ("parent", "deny")]) + # The real denies were in place at the failure, and the tool removed them. + for step in trace["steps"]: + self.assertEqual(set(step["before"]), set(step["after"])) + self.assertEqual(trace["steps"][1]["before"]["file:controller.lock"]["deny"], 1) + self.assertEqual(trace["steps"][1]["after"]["parent"]["deny"], 1) + self.assertEqual(trace["at_end"]["parent"]["deny"], 1) + self.assertEqual(trace["at_end"]["file:controller.lock"]["deny"], 1) + self.assertEqual(trace["after_remove_d"]["parent"]["deny"], 0) + self.assertEqual(trace["cleanup"]["denies_remaining"], 0) + self.assertEqual(trace["cleanup"]["directory_removal"], "ok") + self.assertTrue(ok) + self.assertIs(self.at_exit["tree_denies"], 0) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_trace_error_detail_is_only_a_fixed_allowlisted_code(self): + for error, expected in ( + (ContractError("PHASE_FENCE", "old parent DACL broadened during fence"), + {"type": "ContractError", "category": "PHASE_FENCE", + "detail_code": "PARENT_BROADENED_DURING_FENCE"}), + (ContractError("PHASE_FENCE", SECRET_TEXT), + {"type": "ContractError", "category": "PHASE_FENCE"}), + (ContractError("PHASE_FENCE", "private-account internal project"), + {"type": "ContractError", "category": "PHASE_FENCE"}), + (ContractError("PHASE_FENCE", "bad detail", path=SECRET_TEXT), + {"type": "ContractError", "category": "PHASE_FENCE"})): + with patch.object(self.guarded, "fence_closed_old_phase_guarded", side_effect=error): + trace, ok = diagnostics.trace_fence(self.guarded, temp_factory=self.factory) + self.assertEqual(trace["outcome"], {"error": expected}) + self.assertTrue(ok) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_trace_cleanup_continues_after_a_first_target_timeout_and_reports_it(self): + real_run = subprocess.run + state = {"timed_out": False} + + def first_removal_times_out(args, *a, **k): + if "/remove:d" in args and not state["timed_out"]: + state["timed_out"] = True + raise subprocess.TimeoutExpired(args, 30) + return real_run(args, *a, **k) + sid = self.sid + + def real_denies_then_fail(**request): + self.guarded._icacls(request["old_root"] / "controller.lock", "/deny", "*" + sid + ":(WD,AD)") + self.guarded._icacls(request["old_root"].parent, "/deny", "*" + sid + ":(AD)") + raise PermissionError(SECRET_TEXT) + with patch.object(self.guarded, "fence_closed_old_phase_guarded", side_effect=real_denies_then_fail), \ + patch.object(diagnostics.subprocess, "run", side_effect=first_removal_times_out): + trace, ok = diagnostics.trace_fence(self.guarded, temp_factory=self.factory) + cleanup = trace["cleanup"] + self.assertFalse(ok) + self.assertEqual(cleanup["removal_failures"], + [{"target": "parent", "error": {"type": "TimeoutExpired", "category": "unspecified"}}]) + self.assertEqual(cleanup["denies_remaining"], 1) + self.assertEqual(trace["after_remove_d"]["file:controller.lock"]["deny"], 0) + self.assertEqual(len(cleanup["steps"]), 6) + # The harness, not the tool, removes the one deny the injected timeout left behind. + self.assertEqual(self.at_exit["tree_denies"], 1) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_trace_directory_removal_failure_is_not_clean(self): + outer = self.factory + + class FailsOnRemoval(outer): + def __exit__(self, *exc): + super().__exit__(*exc) + raise OSError(SECRET_TEXT) + trace, ok = diagnostics.trace_fence(self.guarded, temp_factory=FailsOnRemoval) + self.assertFalse(ok) + self.assertEqual(trace["cleanup"]["denies_remaining"], 0) + self.assertEqual(trace["cleanup"]["directory_removal"], "failed") + self.assertEqual(trace["cleanup"]["directory_removal_error"], + {"type": "OSError", "category": "unspecified"}) + self.assertNotIn("private-account", json.dumps(trace)) + + def test_collect_includes_the_trace_and_its_cleanup_status(self): + report, cleanup_ok = self._collect() + self.assertIn("outcome", report["fence_trace"]) + self.assertIn("probe_dacl_before_runner_deny", report) + self.assertTrue(cleanup_ok) + with patch.object(diagnostics, "trace_fence", + return_value=({"cleanup": {"denies_remaining": 2}}, False)): + report, cleanup_ok = self._collect() + self.assertFalse(cleanup_ok) + self.assertEqual(report["fence_trace"], {"cleanup": {"denies_remaining": 2}}) + + def test_main_prints_one_json_object_and_exit_status_follows_cleanup(self): + for ok, expected in ((True, 0), (False, 1)): + buffer = io.StringIO() + with patch.object(diagnostics, "collect", return_value=({"python": "x"}, ok)), \ + redirect_stdout(buffer): + self.assertEqual(diagnostics.main(), expected) + self.assertEqual(json.loads(buffer.getvalue()), {"python": "x"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_windows_fixture_acl.py b/tests/test_windows_fixture_acl.py new file mode 100644 index 0000000..8cd444b --- /dev/null +++ b/tests/test_windows_fixture_acl.py @@ -0,0 +1,213 @@ +"""Owned-fixture ACL stabilization: verification logic (any OS) and its native effect (Windows). + +The platform-independent cases check the argument construction, the ownership boundary and +the fixed-code verification with fake ``icacls`` and DACL reads. The Windows-only cases run +the real ``icacls`` on a disposable tree and are the ones that can show, on a real host, +whether a stabilized fixture really survives the guard's own deny/remove without any change +to its ACE list. Nothing here relaxes a production comparison. +""" +from __future__ import annotations + +import os +from pathlib import Path +import subprocess +import tempfile +import unittest + +from src import paper_pilot_phase_cutover_guarded_v2 as guarded +from tools import windows_fixture_acl as fixture + +DIR_OK = "D:P(A;OICI;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)" +FILE_OK = "D:P(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)" +SECRET = r"C:\\Users\\private-account\\secret-dir" +TEMP_PREFIX = "jraphyte-fixture-acl-" + + +class ProblemCodeTests(unittest.TestCase): + def test_the_stabilized_shapes_have_no_problems_in_any_order(self): + self.assertEqual(fixture.fixture_acl_problems(DIR_OK, True), []) + self.assertEqual(fixture.fixture_acl_problems(FILE_OK, False), []) + self.assertEqual(fixture.fixture_acl_problems("D:PAI(A;;FA;;;OW)(A;;FA;;;SY)(A;;FA;;;BA)", False), []) + + def test_each_departure_has_its_own_fixed_code(self): + for sddl, is_dir, expected in ( + (DIR_OK.replace("D:P", "D:"), True, ["NOT_PROTECTED"]), + (FILE_OK.replace("D:P", "D:AI"), False, ["NOT_PROTECTED"]), + ("D:(A;ID;FA;;;SY)(A;ID;FA;;;BA)(A;ID;FA;;;OW)", False, ["NOT_PROTECTED", "INHERITED_ACE"]), + ("D:P(A;OICIID;FA;;;SY)(A;OICI;FA;;;BA)(A;OICI;FA;;;OW)", True, ["INHERITED_ACE"]), + ("D:P(D;;DCLC;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)", False, ["NON_ALLOW_ACE", "UNREVIEWED_RIGHTS"]), + ("D:P(A;;FR;;;SY)(A;;FA;;;BA)(A;;FA;;;OW)", False, ["UNREVIEWED_RIGHTS"]), + (FILE_OK.replace("(A;;FA;;;SY)", "(A;OICI;FA;;;SY)"), False, ["UNREVIEWED_FLAGS"]), + (DIR_OK.replace("(A;OICI;FA;;;SY)", "(A;;FA;;;SY)"), True, ["UNREVIEWED_FLAGS"]), + (FILE_OK + "(A;;FA;;;WD)", False, ["UNREVIEWED_TRUSTEES"]), + ("D:P(A;;FA;;;SY)(A;;FA;;;BA)", False, ["UNREVIEWED_TRUSTEES"]), + (FILE_OK + "(A;;FA;;;SY)", False, ["UNREVIEWED_TRUSTEES"]), + ("D:P(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA;;;S-1-5-21-1-2-3-500)", False, ["UNREVIEWED_TRUSTEES"]), + ("D:P(A;;FA;;;SY)(A;;FA;;;BA)(A;;FA)", False, ["MALFORMED_ACE", "UNREVIEWED_TRUSTEES"])): + self.assertEqual(fixture.fixture_acl_problems(sddl, is_dir), expected, sddl) + + def test_an_added_everyone_read_allow_reports_both_rights_and_trustee(self): + # The shape icacls /grant *S-1-1-0:(R) leaves behind: a read-only ACE for a trustee outside the trio. + self.assertEqual(fixture.fixture_acl_problems(FILE_OK + "(A;;FR;;;WD)", False), + ["UNREVIEWED_RIGHTS", "UNREVIEWED_TRUSTEES"]) + self.assertEqual(fixture.fixture_acl_problems(DIR_OK + "(A;OICI;FR;;;WD)", True), + ["UNREVIEWED_RIGHTS", "UNREVIEWED_TRUSTEES"]) + self.assertEqual(fixture.fixture_acl_problems("D:P(A;;FR;;;WD)" + FILE_OK[3:], False), + ["UNREVIEWED_RIGHTS", "UNREVIEWED_TRUSTEES"]) + + def test_a_barrier_deny_is_never_part_of_the_stabilized_shape(self): + self.assertEqual(fixture.fixture_acl_problems("D:P(D;;DCLC;;;BU)" + FILE_OK[3:], False), + ["NON_ALLOW_ACE", "UNREVIEWED_RIGHTS", "UNREVIEWED_TRUSTEES"]) + + def test_the_grant_is_the_base_trio_with_inheritance_removed_first(self): + self.assertEqual(fixture.grant_arguments(True), + ["/inheritance:r", "/grant:r", "*S-1-5-18:(OI)(CI)(F)", + "*S-1-5-32-544:(OI)(CI)(F)", "*S-1-3-4:(OI)(CI)(F)"]) + self.assertEqual(fixture.grant_arguments(False), + ["/inheritance:r", "/grant:r", "*S-1-5-18:(F)", "*S-1-5-32-544:(F)", "*S-1-3-4:(F)"]) + # No principal beyond the CPython 0o700 base trio (SYSTEM, Administrators, Owner Rights). + self.assertEqual([alias for alias, _ in fixture.REVIEWED_ALLOWS], ["SY", "BA", "OW"]) + + +class StabilizeTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(prefix=TEMP_PREFIX) + self.addCleanup(self.temp.cleanup) + self.base = Path(self.temp.name) + self.folder = self.base / "state" + self.folder.mkdir() + self.file = self.folder / "a.bin" + self.file.write_bytes(b"0") + self.targets = [("state", self.folder), ("file:a.bin", self.file)] + self.calls = [] + + def run_ok(self, command, **options): + self.calls.append((command, options)) + return subprocess.CompletedProcess(command, 0, "", "") + + def read(self, path): + return DIR_OK if Path(path).is_dir() else FILE_OK + + def test_every_target_gets_its_own_protecting_grant_in_order_and_is_verified(self): + fixture.stabilize_owned_fixture(self.base, self.targets, read=self.read, run=self.run_ok) + self.assertEqual([command for command, _ in self.calls], [ + ["icacls.exe", str(self.folder), *fixture.grant_arguments(True)], + ["icacls.exe", str(self.file), *fixture.grant_arguments(False)]]) + self.assertTrue(all(options["timeout"] == fixture.TIMEOUT_SECONDS for _, options in self.calls)) + + def test_nothing_outside_the_fixture_or_the_fixture_itself_is_ever_changed(self): + for label, path in (("base", self.base), ("parent", self.base.parent), + ("outside", self.base.parent / "elsewhere"), + ("dotdot", self.folder / ".." / ".." / "x")): + with self.assertRaises(fixture.FixtureAclError) as caught: + fixture.stabilize_owned_fixture(self.base, [("state", self.folder), (label, path)], + read=self.read, run=self.run_ok) + self.assertEqual(caught.exception.problems, [(label, "OUTSIDE_OWNED_FIXTURE")]) + self.assertEqual(self.calls, []) + + def test_a_failed_or_timed_out_icacls_is_reported_and_later_targets_are_still_attempted(self): + results = iter((subprocess.CompletedProcess([], 5, "", SECRET),)) + + def first_fails_second_times_out(command, **options): + self.calls.append(command) + try: + return next(results) + except StopIteration: + raise subprocess.TimeoutExpired(command, 30) + with self.assertRaises(fixture.FixtureAclError) as caught: + fixture.stabilize_owned_fixture(self.base, self.targets, read=self.read, + run=first_fails_second_times_out) + self.assertEqual(caught.exception.problems, [("state", "ICACLS_FAILED"), ("file:a.bin", "ICACLS_FAILED")]) + self.assertEqual(len(self.calls), 2) + + def test_a_dacl_that_is_not_the_stabilized_shape_fails_with_codes_not_text(self): + def read(path): + if Path(path).is_dir(): + return DIR_OK + raise OSError(SECRET) + with self.assertRaises(fixture.FixtureAclError) as caught: + fixture.stabilize_owned_fixture(self.base, self.targets, read=read, run=self.run_ok) + self.assertEqual(caught.exception.problems, [("file:a.bin", "DACL_UNREADABLE")]) + wide = DIR_OK + "(A;OICI;FR;;;WD)" + with self.assertRaises(fixture.FixtureAclError) as caught: + fixture.stabilize_owned_fixture(self.base, self.targets, read=lambda p: wide if Path(p).is_dir() + else FILE_OK, run=self.run_ok) + self.assertEqual(caught.exception.problems, [("state", "UNREVIEWED_RIGHTS"), ("state", "UNREVIEWED_TRUSTEES")]) + for error in (caught.exception, fixture.FixtureAclError([("x", "ICACLS_FAILED")])): + self.assertNotIn("private-account", repr(error) + str(error)) + + +def _icacls(*args: str) -> None: + result = subprocess.run(["icacls.exe", *args], capture_output=True, text=True, timeout=30) + assert result.returncode == 0, result.stderr or result.stdout + + +@unittest.skipUnless(os.name == "nt", "Windows ACL contract") +class NativeStabilizedFixtureTests(unittest.TestCase): + def setUp(self): + self.sid = guarded._current_sid() + self.temp = tempfile.TemporaryDirectory(prefix=TEMP_PREFIX) + self.addCleanup(self._cleanup) + self.base = Path(self.temp.name) + self.folder = self.base / "old-app" + self.state = self.folder / "state" + self.state.mkdir(parents=True) + self.file = self.state / "controller.lock" + self.file.write_bytes(b"0") + self.targets = [("parent", self.folder), ("state", self.state), ("file", self.file)] + + def _cleanup(self): + for target in (self.folder, self.file): + if target.exists(): + subprocess.run(["icacls.exe", str(target), "/remove:d", "*" + self.sid], + capture_output=True, text=True, timeout=30) + self.temp.cleanup() + self.assertFalse(self.base.exists()) + + def stabilize(self): + fixture.stabilize_owned_fixture(self.base, self.targets, read=guarded.directory_dacl_sddl) + + def test_the_real_dacls_are_protected_explicit_and_exactly_the_reviewed_trio(self): + self.stabilize() + for _, path in self.targets: + sddl = guarded.directory_dacl_sddl(path) + self.assertEqual(fixture.fixture_acl_problems(sddl, path.is_dir()), [], sddl) + self.assertNotIn("ID", sddl) + # The runner can still use its own fixture. + self.file.write_bytes(b"1") + self.assertEqual(self.file.read_bytes(), b"1") + + def test_the_stabilized_fixture_survives_a_real_deny_and_removal_without_gaining_any_ace(self): + self.stabilize() + original_dir = guarded.directory_dacl_sddl(self.folder) + original_file = guarded.directory_dacl_sddl(self.file) + _icacls(str(self.file), "/deny", "*" + self.sid + ":(WD,AD)") + _icacls(str(self.folder), "/deny", "*" + self.sid + ":(AD)") + now_file = guarded.directory_dacl_sddl(self.file) + now_dir = guarded.directory_dacl_sddl(self.folder) + # The unchanged production comparisons: only the admission deny was added. + self.assertTrue(guarded._only_admission_dacl_added(original_file, now_file, self.sid), now_file) + self.assertTrue(guarded._only_parent_admission_added(original_dir, now_dir, self.sid), now_dir) + self.assertEqual(len(guarded._aces(now_file)), len(guarded._aces(original_file)) + 1) + self.assertEqual(len(guarded._aces(now_dir)), len(guarded._aces(original_dir)) + 1) + _icacls(str(self.file), "/remove:d", "*" + self.sid) + _icacls(str(self.folder), "/remove:d", "*" + self.sid) + self.assertTrue(guarded._same_effective_dacl(original_file, guarded.directory_dacl_sddl(self.file))) + self.assertTrue(guarded._same_effective_dacl(original_dir, guarded.directory_dacl_sddl(self.folder))) + + def test_a_later_unreviewed_grant_is_reported_by_the_verification(self): + self.stabilize() + _icacls(str(self.file), "/grant", "*S-1-1-0:(R)") + self.assertEqual(fixture.fixture_acl_problems(guarded.directory_dacl_sddl(self.file), False), + ["UNREVIEWED_RIGHTS", "UNREVIEWED_TRUSTEES"]) + + def test_the_fixture_base_itself_is_never_touched(self): + before = guarded.directory_dacl_sddl(self.base) + with self.assertRaises(fixture.FixtureAclError): + fixture.stabilize_owned_fixture(self.base, [("base", self.base)], read=guarded.directory_dacl_sddl) + self.stabilize() + self.assertEqual(guarded.directory_dacl_sddl(self.base), before) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/windows_fence_diagnostics.py b/tools/windows_fence_diagnostics.py new file mode 100644 index 0000000..f0a4959 --- /dev/null +++ b/tools/windows_fence_diagnostics.py @@ -0,0 +1,452 @@ +#!/usr/bin/env python3 +"""Sanitized Windows trustee-serialization diagnostics for the hosted fence job. + +Prints one JSON object describing how this runner's own SID is written in a real +DACL string and whether the admission barrier recognizes it. Machine and domain +sub-authorities are redacted; no account name, path or exception text is emitted. +The owned probe deny is always removed again and the removal is verified and +reported, as is the probe directory's own removal; the exit status is nonzero only when +a cleanup is not verified. + +It also runs the real guarded fence once on a disposable owned fixture and prints +sanitized DACL descriptions (control flags P/AI, ACE counts, explicit vs. inherited +allows, and the production strict-comparison verdicts) for the fixture's temp parent, +base, fence parent, state directory and state files: as created, after the owned fixture +ACL is made explicit and protected (windows_fixture_acl), and before any pin, +immediately before and after every icacls call (including failed calls), around each +cleanup removal, at the end of the run and after cleanup. Recording is capped at +MAX_RECORDED_OPERATIONS. It only observes: it changes no production behavior and +asserts no cause. Every fixture target gets its own removal attempt; any survivor, +failed removal or failed temp-directory removal makes the run's status non-clean. +""" +from __future__ import annotations + +import hashlib +import json +import os +import platform +import re +import subprocess +import sys +import tempfile +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from tools import windows_fixture_acl # noqa: E402 + + +def redact(text: str) -> str: + text = re.sub(r"S-1-5-21-\d+-\d+-\d+-(\d+)", r"S-1-5-21--\1", text) + return re.sub(r"S-1-12-1(?:-\d+){4}", "S-1-12-1-", text) + + +_ACE = re.compile(r"\([^)]*\)") +_CONTROL = re.compile(r"AI|AR|P") +_TWIN_FLAG_TOKEN = re.compile(r"..") + + +def _ace_fields(ace: str) -> list[str]: + return ace[1:-1].split(";") + + +def _twin_key(ace: str) -> tuple: + """Type, non-inherited flag tokens, rights and trustee of an allow ACE.""" + fields = _ace_fields(ace) + if len(fields) != 6: + return () + tokens = [t for t in _TWIN_FLAG_TOKEN.findall(fields[1]) if t != "ID"] + return (fields[0], tuple(tokens), fields[2], fields[5]) + + +def describe_dacl(sddl: str) -> dict: + """Sanitized structural description of one DACL string (text only; no Windows calls).""" + control = _CONTROL.findall(sddl.split("(", 1)[0].removeprefix("D:")) + aces = _ACE.findall(sddl) + allows = [ace for ace in aces if ace.startswith("(A;") and len(_ace_fields(ace)) == 6] + inherited = [ace for ace in allows if "ID" in _TWIN_FLAG_TOKEN.findall(_ace_fields(ace)[1])] + explicit = [ace for ace in allows if ace not in inherited] + inherited_keys = {_twin_key(ace) for ace in inherited} + return {"sddl": redact(sddl), "control": "".join(control), + "protected": "P" in control, "auto_inherited": "AI" in control, + "ace_count": len(aces), + "deny": sum(1 for ace in aces if ace.startswith("(D;")), + "explicit_allow": len(explicit), "inherited_allow": len(inherited), + "explicit_copies_of_inherited": sum(1 for ace in explicit + if _twin_key(ace) in inherited_keys)} + + +def describe_error(exc: BaseException) -> dict: + """Exception type and a bounded category only; never the message (it may hold paths or names).""" + kind = type(exc).__name__ + code = getattr(exc, "code", None) + result = {"type": kind if re.fullmatch(r"\w{1,64}", kind) else "Exception", + "category": code if isinstance(code, str) and re.fullmatch(r"[A-Z][A-Z0-9_]{0,31}", code) + else "unspecified"} + exit_code = getattr(exc, "returncode", None) + if type(exit_code) is int: + result["exit_code"] = exit_code + return result + + +# Exact constant messages raised by the guarded fence, mapped to fixed diagnostic codes. +# Nothing outside this table is ever copied into the report. +_DETAIL_CODES = { + "Windows DACL update failed": "ICACLS_UPDATE_FAILED", + "historical five-file inventory differs": "INVENTORY_DIFFERS", + "historical state bytes differ": "STATE_BYTES_DIFFER", + "reviewed SID or five-file pins differ": "PINS_DIFFER", + "reviewed ACL pins absent": "ACL_PINS_ABSENT", + "ambiguous old/archive namespace": "AMBIGUOUS_NAMESPACE", + "historical barrier scope differs": "BARRIER_SCOPE_DIFFERS", + "archive lacks prior barrier": "ARCHIVE_LACKS_BARRIER", + "existing owner deny cannot be safely merged": "OWNER_DENY_NOT_MERGEABLE", + "original Windows DACL identity changed": "ORIGINAL_DACL_IDENTITY_CHANGED", + "unreviewed preexisting file deny": "PREEXISTING_FILE_DENY", + "old parent DACL changed beyond admission deny": "PARENT_CHANGED_BEYOND_DENY", + "database changed after lock acquisition": "DATABASE_CHANGED", + "unreviewed old file DACL drift": "FILE_DACL_DRIFT", + "new old-state write open not denied": "WRITE_OPEN_NOT_DENIED", + "old state recreation remains possible": "RECREATION_POSSIBLE", + "old parent DACL broadened during fence": "PARENT_BROADENED_DURING_FENCE", + "stage lock path replaced while held": "STAGE_LOCK_REPLACED", + "fixed old path not closed": "OLD_PATH_NOT_CLOSED", + "archived readback still denied": "ARCHIVED_STILL_DENIED", + "old fixed path admission reopened": "ADMISSION_REOPENED", + "old parent DACL changed after archive": "PARENT_CHANGED_AFTER_ARCHIVE", + "archived effective file DACL changed": "ARCHIVED_FILE_DACL_CHANGED", +} + + +def _outcome(exc: BaseException) -> dict: + """Error description plus a fixed code, only for a known constant message without a path.""" + result = describe_error(exc) + detail = getattr(exc, "detail", None) + if isinstance(detail, str) and not getattr(exc, "path", ""): + code = _DETAIL_CODES.get(detail) + if code is not None: + result["detail_code"] = code + return result + + +def _remove_probe_deny(guarded, probe: Path, sid: str) -> dict: + """Remove the owned probe deny and verify, independently of the checks under diagnosis.""" + outcome = {"icacls_remove_exit": None, "deny_absent_verified": False} + try: + removed = subprocess.run(["icacls.exe", str(probe), "/remove:d", "*" + sid], + capture_output=True, text=True, timeout=30) + outcome["icacls_remove_exit"] = removed.returncode + except Exception as exc: + outcome["error"] = describe_error(exc) + return outcome + try: + outcome["deny_absent_verified"] = not guarded._owner_deny_present( + guarded.directory_dacl_sddl(probe), sid) + except Exception as exc: + outcome["verify_error"] = describe_error(exc) + return outcome + + +def _snapshot(guarded, read, paths: dict, sid: str | None = None, originals: dict | None = None) -> dict: + """Describe each existing path's DACL; with originals, add the production strict verdicts.""" + result = {} + for label, path in paths.items(): + try: + present = path.exists() + except Exception as exc: + result[label] = {"error": describe_error(exc)} + continue + if not present: + result[label] = {"absent": True} + continue + try: + raw = read(path) + entry = describe_dacl(raw) + except Exception as exc: + result[label] = {"error": describe_error(exc)} + continue + if originals is not None and label in originals: + entry["strict"] = _verdicts(guarded, label, originals[label], raw, sid) + result[label] = entry + return result + + +def _verdicts(guarded, label: str, original: str, current: str, sid: str) -> dict: + """The unchanged production comparators, applied to one observed DACL.""" + admission = (guarded._only_parent_admission_added if label == "parent" + else guarded._only_admission_dacl_added) + verdict = {} + for key, check in (("admission_only_added", lambda: admission(original, current, sid)), + ("same_as_original", lambda: guarded._same_effective_dacl(original, current))): + try: + verdict[key] = bool(check()) + except Exception as exc: + verdict[key] = None + verdict[key + "_error"] = describe_error(exc) + return verdict + + +def _fixture_paths(guarded, base: Path, root: Path, old: Path, archive: Path) -> dict: + state = old if old.exists() else archive + paths = {"temp_parent": base.parent, "base": base, "parent": root, "state": state} + for name in sorted(guarded.NAMES): + paths["file:" + name] = state / name + return paths + + +MAX_RECORDED_OPERATIONS = 64 +_COMPACT = ("sddl", "protected", "auto_inherited", "ace_count", "deny", + "explicit_copies_of_inherited", "strict", "absent", "error") + + +def _compact(snapshot: dict) -> dict: + return {label: {key: entry[key] for key in _COMPACT if key in entry} + for label, entry in snapshot.items()} + + +def _label(root: Path, archive: Path, target: Path) -> str: + return ("parent" if target == root else + ("archived:" if target.parent == archive else "old:") + target.name) + + +def _cleanup_targets(guarded, root: Path, old: Path, archive: Path) -> list: + targets = [("parent", root)] + for folder, prefix in ((old, "old:"), (archive, "archived:")): + targets += [(prefix + name, folder / name) for name in sorted(guarded.NAMES)] + return targets + + +def _exists(path: Path) -> bool: + try: + return path.exists() + except Exception: + return True # unknown: still attempt and verify it + + +def _lift_and_verify(guarded, read, sid: str, targets: list, observe, record, cleanup: dict) -> None: + """Attempt every owned target's own removal, then verify every surviving target.""" + failures = [] + for label, target in targets: + if not _exists(target): + continue + step = {"target": label, "operation": "cleanup_remove_deny", "before": observe()} + try: + done = subprocess.run(["icacls.exe", str(target), "/remove:d", "*" + sid], + capture_output=True, text=True, timeout=30) + step["icacls_exit"] = done.returncode + if done.returncode != 0: + failures.append({"target": label, "exit_code": done.returncode}) + except Exception as exc: + step["icacls"] = "failed" + step["error"] = describe_error(exc) + failures.append({"target": label, "error": step["error"]}) + step["after"] = observe(strict=True) + record(cleanup["steps"], step) + remaining = verification_errors = checked = 0 + for label, target in targets: + if not _exists(target): + continue + checked += 1 + try: + if guarded._owner_deny_present(read(target), sid): + remaining += 1 + except Exception: + verification_errors += 1 + remaining += 1 # an unverifiable target is not a clean one + cleanup.update({"paths_checked": checked, "denies_remaining": remaining, + "verification_errors": verification_errors, "removal_failures": failures}) + + +def trace_fence(guarded, *, temp_factory=tempfile.TemporaryDirectory) -> tuple[dict, bool]: + """Run the real guarded fence on an owned fixture, observing sanitized DACL descriptions. + + The fixture is recorded as created, then given the verified explicit, protected + ACL of windows_fixture_acl, then pinned. Returns (trace, clean). clean is False for + a failed setup or fixture stabilization, a failed or unverified removal on any + target, or a failed temp-directory removal. + """ + trace = {"steps": [], "operations_dropped": 0} + cleanup = {"paths_checked": 0, "denies_remaining": None, "removal_failures": [], + "verification_errors": 0, "steps": [], "directory_removal": "not_reached"} + recorded = {"count": 0} + + def record(bucket: list, step: dict) -> None: + if recorded["count"] >= MAX_RECORDED_OPERATIONS: + trace["operations_dropped"] += 1 + return + recorded["count"] += 1 + bucket.append(step) + + body_done = False + try: + sid = guarded._current_sid() + with temp_factory(prefix="jraphyte-fence-diag-trace-") as temp: + base = Path(temp) + root, old = base / "old-app", base / "old-app" / "state" + archive = base / "archive" / "state" + receipts = base / "receipts" + old.mkdir(parents=True) + archive.parent.mkdir() + receipts.mkdir() + content = {} + for name in sorted(guarded.NAMES): + body = b"0" if name == "controller.lock" else name.encode() + (old / name).write_bytes(body) + content[name] = hashlib.sha256(body).hexdigest() + for name in ("source-stage.lock", "phase-stage.lock"): + (root / name).write_bytes(b"0") + read = guarded.directory_dacl_sddl + originals = None + + def observe(strict: bool = False) -> dict: + try: + return _compact(_snapshot(guarded, read, _fixture_paths(guarded, base, root, old, archive), + sid if strict else None, originals if strict else None)) + except Exception as exc: + return {"error": describe_error(exc)} + try: + # The fixture exactly as created: nothing yet applied, nothing yet pinned. + trace["setup_as_created"] = _snapshot(guarded, read, _fixture_paths(guarded, base, root, old, archive)) + targets = [("parent", root), ("state", old)] + [("file:" + name, old / name) + for name in sorted(guarded.NAMES)] + try: + windows_fixture_acl.stabilize_owned_fixture(base, targets, read=read, run=subprocess.run) + trace["fixture_acl"] = {"status": "stabilized", "targets": len(targets)} + except windows_fixture_acl.FixtureAclError as exc: + trace["fixture_acl"] = {"status": "failed", "problems": [ + {"target": label, "code": code} for label, code in exc.problems]} + raise + # The fixture as it is pinned: after stabilization, before the first fence call. + trace["setup"] = _snapshot(guarded, read, _fixture_paths(guarded, base, root, old, archive)) + originals = {"parent": read(root)} + originals.update({"file:" + name: read(old / name) for name in guarded.NAMES}) + pins = (hashlib.sha256(originals["parent"].encode()).hexdigest(), + {name: hashlib.sha256(originals["file:" + name].encode()).hexdigest() + for name in guarded.NAMES}) + actual = guarded._icacls + + def recording(path, *args): + target = Path(path) + # Only the constant historical file names and the location are reported. + step = {"target": _label(root, archive, target), + "operation": "deny" if "/deny" in args else "remove_deny", + "before": observe(strict=True)} + try: + actual(path, *args) + step["icacls"] = "ok" + except BaseException as exc: + step["icacls"] = "failed" + step["error"] = describe_error(exc) + raise + finally: + step["after"] = observe(strict=True) + record(trace["steps"], step) + guarded._icacls = recording + try: + result = guarded.fence_closed_old_phase_guarded( + old_root=old, archived_root=archive, expected_file_sha256=content, + source_stage_lock=root / "source-stage.lock", + phase_stage_lock=root / "phase-stage.lock", deny_sid=sid, + expected_parent_dacl_sha256=pins[0], expected_file_dacl_sha256=pins[1], + barrier_receipt_path=receipts / "barrier.json", + fence_receipt_path=receipts / "fence.json") + trace["outcome"] = {"status": result["status"]} + finally: + guarded._icacls = actual + except Exception as exc: + trace["outcome"] = {"error": _outcome(exc)} + finally: + try: + trace["at_end"] = _snapshot(guarded, read, _fixture_paths(guarded, base, root, old, archive), + sid, originals) + except Exception as exc: + trace["at_end"] = {"error": describe_error(exc)} + _lift_and_verify(guarded, read, sid, _cleanup_targets(guarded, root, old, archive), + observe, record, cleanup) + try: + trace["after_remove_d"] = _snapshot(guarded, read, + _fixture_paths(guarded, base, root, old, archive), + sid, originals) + except Exception as exc: + trace["after_remove_d"] = {"error": describe_error(exc)} + body_done = True + cleanup["directory_removal"] = "ok" + except Exception as exc: + if body_done: + cleanup["directory_removal"] = "failed" + cleanup["directory_removal_error"] = describe_error(exc) + else: + trace["error"] = describe_error(exc) # setup or temp creation failed + trace["cleanup"] = cleanup + clean = (cleanup["denies_remaining"] == 0 and not cleanup["removal_failures"] + and cleanup["directory_removal"] == "ok" + and trace.get("fixture_acl", {}).get("status") == "stabilized") + return trace, clean + + +def collect(guarded=None, *, temp_factory=tempfile.TemporaryDirectory) -> tuple[dict, bool]: + """Return (sanitized report, whether every owned probe/fixture cleanup is verified).""" + report = {"python": platform.python_version(), "os": platform.platform(), + "os_name": os.name} + cleanup_ok = True + try: + if guarded is None: + from src import paper_pilot_phase_cutover_guarded_v2 as guarded + sid = guarded._current_sid() + report["runner_sid"] = redact(sid) + report["runner_canonical_trustee"] = redact(guarded._canonical_trustee(sid)) + directory = temp_factory(prefix="jraphyte-fence-diag-") + temp = directory.__enter__() + try: + probe = Path(temp) / "probe.bin" + probe.write_bytes(b"0") + try: + report["probe_dacl_before_runner_deny"] = describe_dacl(guarded.directory_dacl_sddl(probe)) + except Exception as exc: + report["probe_before_error"] = describe_error(exc) + attempted = False + try: + attempted = True + subprocess.run(["icacls.exe", str(probe), "/deny", "*" + sid + ":(WD,AD)"], + capture_output=True, text=True, timeout=30, check=True) + raw = guarded.directory_dacl_sddl(probe) + report["probe_dacl_after_runner_deny"] = redact(raw) + report["probe_dacl_after_runner_deny_shape"] = describe_dacl(raw) + report["literal_numeric_ace_present"] = guarded.DENY_FILE_ACE.format(sid=sid) in raw + report["barrier_recognizes_deny"] = guarded._file_deny(probe, sid) + report["owner_deny_detected"] = guarded._owner_deny_present(raw, sid) + finally: + # Runs even if the deny, the DACL query or the recognition failed. + if attempted: + cleanup = _remove_probe_deny(guarded, probe, sid) + report["probe_cleanup"] = cleanup + cleanup_ok = (cleanup["icacls_remove_exit"] == 0 + and cleanup["deny_absent_verified"] is True) + except Exception as exc: # diagnostics must never mask the test result + report["diagnostic_error"] = describe_error(exc) + finally: + # The probe directory's own removal is part of the cleanup status, whatever the body did. + try: + directory.__exit__(None, None, None) + report["probe_directory_removal"] = "ok" + except Exception as exc: + report["probe_directory_removal"] = "failed" + report["probe_directory_removal_error"] = describe_error(exc) + cleanup_ok = False + except Exception as exc: + report["diagnostic_error"] = describe_error(exc) + if guarded is not None: + report["fence_trace"], trace_ok = trace_fence(guarded, temp_factory=temp_factory) + cleanup_ok = cleanup_ok and trace_ok + return report, cleanup_ok + + +def main() -> int: + report, cleanup_ok = collect() + print(json.dumps(report, indent=2)) + return 0 if cleanup_ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/windows_fixture_acl.py b/tools/windows_fixture_acl.py new file mode 100644 index 0000000..0a8910c --- /dev/null +++ b/tools/windows_fixture_acl.py @@ -0,0 +1,100 @@ +"""Deterministic DACLs for owned, disposable Windows test fixtures. + +Hosted Windows Server 2025 traces show that ``icacls /deny`` on an object that is not +protected and carries inherited allows leaves explicit copies of those allows behind, +which the production strict comparison rightly rejects. A test fixture therefore must not +depend on whatever ACL its temporary parent happens to pass down. + +This module gives each fixture object the same three allows the CPython ``0o700`` base +directory already carries (SYSTEM, Administrators, Owner Rights; full control, directories +inheritable), as *explicit* ACEs with inheritance removed. No principal is added or widened +relative to that base. The result is verified before anything is pinned. This only changes +what the disposable fixtures start from; it does not alter or relax any production check, +and whether it is sufficient on the hosted runner is for hosted evidence to show. +""" +from __future__ import annotations + +from pathlib import Path +import re +import subprocess + +REVIEWED_ALLOWS = (("SY", "S-1-5-18"), ("BA", "S-1-5-32-544"), ("OW", "S-1-3-4")) +TIMEOUT_SECONDS = 30 +_ACE = re.compile(r"\([^)]*\)") +_CONTROL = re.compile(r"AI|AR|P") +_FLAG = re.compile("..") + + +class FixtureAclError(Exception): + """The fixture ACL could not be stabilized; only fixed target labels and codes are kept.""" + + def __init__(self, problems: list[tuple[str, str]]): + super().__init__("owned fixture ACL not stabilized") + self.problems = problems + + +def grant_arguments(is_dir: bool) -> list[str]: + rights = "(OI)(CI)(F)" if is_dir else "(F)" + return ["/inheritance:r", "/grant:r"] + ["*" + sid + ":" + rights for _, sid in REVIEWED_ALLOWS] + + +def fixture_acl_problems(sddl: str, is_dir: bool) -> list[str]: + """Fixed problem codes for a DACL that is not the stabilized fixture shape (text only).""" + found: list[str] = [] + control = _CONTROL.findall(sddl.split("(", 1)[0].removeprefix("D:")) + if "P" not in control: + found.append("NOT_PROTECTED") + expected_flags = ["CI", "OI"] if is_dir else [] + trustees = [] + for ace in _ACE.findall(sddl): + fields = ace[1:-1].split(";") + if len(fields) != 6: + found.append("MALFORMED_ACE") + continue + kind, flags, rights, trustee = fields[0], _FLAG.findall(fields[1]), fields[2], fields[5] + trustees.append(trustee) + if "ID" in flags: + found.append("INHERITED_ACE") + if kind != "A": + found.append("NON_ALLOW_ACE") + if rights != "FA": + found.append("UNREVIEWED_RIGHTS") + if sorted(flag for flag in flags if flag != "ID") != expected_flags: + found.append("UNREVIEWED_FLAGS") + if sorted(trustees) != sorted(alias for alias, _ in REVIEWED_ALLOWS): + found.append("UNREVIEWED_TRUSTEES") + return list(dict.fromkeys(found)) + + +def stabilize_owned_fixture(base, targets, *, read, run=subprocess.run) -> None: + """Protect each target and give it exactly the reviewed explicit allows, then verify. + + ``targets`` is an ordered list of ``(label, path)``; every path must lie strictly inside + ``base`` (the fixture's own directory), so no ancestor ACL is ever touched. Raises + ``FixtureAclError`` carrying fixed codes if any target is outside, any icacls call fails, + or any resulting DACL is not the verified shape. + """ + root = Path(base).resolve() + for label, path in targets: + resolved = Path(path).resolve() + if resolved == root or not resolved.is_relative_to(root): + raise FixtureAclError([(label, "OUTSIDE_OWNED_FIXTURE")]) + problems: list[tuple[str, str]] = [] + for label, path in targets: + is_dir = Path(path).is_dir() + try: + done = run(["icacls.exe", str(path), *grant_arguments(is_dir)], + capture_output=True, text=True, timeout=TIMEOUT_SECONDS) + applied = done.returncode == 0 + except Exception: + applied = False + if not applied: + problems.append((label, "ICACLS_FAILED")) + continue + try: + codes = fixture_acl_problems(read(path), is_dir) + except Exception: + codes = ["DACL_UNREADABLE"] + problems.extend((label, code) for code in codes) + if problems: + raise FixtureAclError(problems)