From bec0acf6b5e42cc9be7b6ea85a20fa0aad31aace Mon Sep 17 00:00:00 2001 From: shawn <3226077178@qq.com> Date: Sat, 12 Sep 2026 09:28:46 +0800 Subject: [PATCH] fix(subagents): make acceptance checks portable on Windows (#5162) * fix(subagents): make acceptance checks portable on Windows * fix(subagents): reject drive-root path escapes * fix(subagents): preserve drive-root containment * fix(subagents): use Windows path casing rules * fix(subagents): reject drive-relative cd paths * fix(subagents): reject shell-dependent cd targets * fix(subagents): reject shell-dependent runner paths * fix(subagents): harden cross-shell acceptance checks * fix(subagents): reject ambiguous shell tokenization * fix(subagents): reject tokenizer segment drift * fix(subagents): reject ambiguous PowerShell syntax * fix(subagents): include all PowerShell quote delimiters * fix(subagents): fail closed on cross-family paths * fix(subagents): reject ambiguous Windows aliases * fix(subagents): reject PSDrive alias exclusions * fix(subagents): reject PSDrive-relative aliases --------- Co-authored-by: Willem Jiang --- backend/AGENTS.md | 5 +- .../harness/deerflow/subagents/AGENTS.md | 1 + .../deerflow/subagents/acceptance_checks.py | 479 ++++++++++++- backend/tests/test_acceptance_checks.py | 634 +++++++++++++++++- 4 files changed, 1093 insertions(+), 26 deletions(-) diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 04d964b18..6a914acb5 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -261,9 +261,8 @@ make test-live PYTHONPATH=. uv run pytest tests/test_.py -v ``` -Direct pytest collection or execution of `tests/test_client_live.py` remains -skipped unless `DEER_FLOW_RUN_LIVE_TESTS=1` is set. Do not add that opt-in to -default CI workflows. +Keep live tests opt-in via `DEER_FLOW_RUN_LIVE_TESTS=1`; guard POSIX-only +markers with `os.name` for Windows collection. Jina logging tests use dummy keys (`tests/test_jina_client.py`). Jina/Browserless/InfoQuest resolve URLs without rebuilding HTML. diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 3f8acffee..495c5e835 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -24,6 +24,7 @@ executions are not checked, and acceptance never changes automatic retry policy. **Flow**: Ordinary `task()` → `SubagentExecutor` → shared process slot → result polling/SSE. Explicit `batch_task()` → durable batch/item rows → lease-based batch service (`subagents/batch_service.py`, started by Gateway or an explicit direct runtime) → the same `SubagentExecutor`/process slots → bounded stored result and owner-scoped API/JSONL export. Batch mode is selected only by the explicit tool, never inferred from prompt size. Executor queue rejection/timeout occurs before model execution and therefore releases the durable lease without consuming an item attempt; real execution failure and expired leases still consume the retry budget. User cancellation terminalizes every nonterminal item immediately and clears its lease, fencing any stale worker completion. Background cancellation resolves the result/future under `_background_tasks_lock` but calls `Future.cancel()` only after releasing it, because cancellation may synchronously invoke the completion callback that reacquires the registry lock. Direct runtimes provide the tools and worker but not Gateway's HTTP/UI surface. `task_started` carries the resolved effective model name. The per-subagent `SubagentTokenCollector` publishes a cumulative usage snapshot to the shared `SubagentResult` after every completed LLM response; the next `task_running` event carries that snapshot, so collapsed workspace cards can update without re-accounting parent-run totals. Terminal ToolMessage metadata (`subagent_model_name`, `subagent_token_usage`) and the persisted `subagent.end` event retain the model/usage after reload; absent provider usage stays absent rather than being estimated as zero. The executor caches one resolved `AppConfig` snapshot (explicit or `get_app_config()` fallback) for agent assembly, deferred setup, and receipt harvesting, so `verification.receipts_enabled=false` remains authoritative on both construction paths. Terminal tool receipts are harvested before `try_set_terminal` and committed with the other payload fields under the same state lock, so status polling cannot observe a terminal result before its receipt metadata is available. Each yielded values chunk becomes the latest terminal-harvest state and immediately publishes its harvested receipts to the shared result before cooperative cancellation is checked. Tool-ended cancellation/failure evidence uses the current ToolMessage scan, but a completed result always uses the bounded ledger snapshot attached to the assistant text being returned—even when a max-turn partial ends on a later tool chunk—so omitted receipts cannot validate its citations; a missing/malformed completed snapshot fails closed with no receipts. Therefore direct task cancellation and both execution/polling timeouts retain the latest execution evidence even when cancellation interrupts before another stream boundary. **Report contract (RFC #4651 PR3)**: `report_contract.py` owns the prompt-layer text that makes Layer 1 receipt verification non-inert. `SubagentExecutor._build_initial_state` appends `build_report_contract_section(receipts_enabled=...)` to every subagent's consolidated `SystemMessage` — built-in and custom alike — requiring `[rN tool_name]` citations (from the Tool receipts ledger) for action claims, verifiable handles (absolute path, URL, ID, HTTP status) for deliverables, and explicit reporting of failures; the citation clause follows `verification.receipts_enabled`, and the citation example derives from the single-owner `format_citation`/`receipt_id` so prompt text cannot drift from the verifier. The `task` tool hands lead-supplied `acceptance_criteria` to the `SubagentExecutor` constructor, which appends them via `render_acceptance_criteria_block(...)` to the task `HumanMessage` (stripped, capped at 20 items × 500 chars, each entry neutralized) — the untrusted channel `InputSanitizationMiddleware` escapes and boundary-frames, matching their model-supplied provenance. The subagent's `SystemMessage` never carries criterion text; it gets only the framework-owned `build_acceptance_criteria_system_note(...)` pointer naming the list's location and authority, so natural-language injection inside a criterion cannot gain system-channel priority over framework instructions. Deterministic leaf checking is a separate layer. **Acceptance checklist (RFC #4651 PR4)**: `acceptance_checks.py` checks lead-supplied `acceptance_criteria` in code on the `task` tool's completed branch (offloaded via `asyncio.to_thread`, failure-isolated). Decidable leaves: `file: exists|non-empty` and `file_written:` read through `read_current_file_content` scoped to the shared thread workspace (`workspace_path`/`outputs_path`; virtual `/mnt/user-data/...` prefixes and workspace-relative spellings normalized first) — the read itself uses the sandbox-native virtual form because the local read validator and provider mount tables resolve virtual paths, not host paths; a remote provider's `"Error: ..."` return string (remote providers return instead of raise for missing files) is normalized to a failed check, never evaluated as content — provider-typed via `is_local_sandbox`, so a genuine `Error:`-prefixed file on the local sandbox stays valid content; a UnicodeDecodeError on a text read marks a binary deliverable (PDF/image) as existing and non-empty instead of dropping the verdict; reads are byte-bounded — the size is established first (`os.stat` on the validated host path locally, so the host-bash-disabled configuration needs no shell; a metadata-only `stat`/`realpath` probe in a fresh `env -i` shell on remote providers — absolute-path utilities and a marker env that routes AIO off its persistent shell, so a completed subagent's poisoned session state (functions/aliases/PATH/exported functions/locale) cannot steer it; `stat` never opens content, so a FIFO cannot block the parent for the provider's idle timeout; the file must stat as a regular non-symlink file, and containment is canonicalized — the file's realpath must stay under the mount root's realpath, which is exactly what the provider's own read path resolves (e2b and Tenki realize `/mnt/user-data` as a symlink to the home dir by default), so a final-component symlink is rejected outright and an intermediate dir-link escape under a sane root still lands outside the canonical root), leaves above `_FILE_CONTENT_READ_CAP_BYTES` answer from the size alone — `file_written` only with an added bounded one-byte open probe (stat metadata is not read-back: a mode-000 file stats fine while any open raises EACCES), and smaller files run the full read, and an unestablishable size degrades to UNVERIFIED rather than an unbounded fallback read; out-of-scope paths degrade to UNVERIFIED, never misjudge — on the local sandbox the scope decision canonicalizes with realpath, so a workspace symlink into uploads cannot satisfy a scoped leaf with upload content. `tests_passed:` anchors to a matching bash execution (newest match wins) with `status=success` and a test-summary shape in its bounded output tail — each harvested execution carries a `shell_persistent` provenance stamp — the producing sandbox's `persistent_shell_sessions` flag (AIO's legacy exec path) resolved from the state that carried the evidence, never the parent runtime, which has no `sandbox` key when the parent delegated before touching one — and a persistent stamp (or an unidentifiable or undeclared one — a custom provider that never declared its session semantics is unknown, not fresh-shell; unknown provenance fails closed) degrades the leaf to UNVERIFIED instead, because any earlier call in the shared session could have mutated the state the clean-looking run executed in and a fresh controlled session (RFC §6 verifier) would be needed to prove otherwise, harvested by the executor (`_harvest_bash_executions`, only when criteria were delegated) from the same stamped `ToolMessage`s the receipt layer reads, accumulated per streamed chunk (merged by `tool_call_id`, newest-capped) so subagent summarization compacting earlier messages cannot erase a recorded execution, with over-cap commands carrying `command_truncated` so the matcher degrades to UNVERIFIED instead of proving a match on a suffix-less prefix; the recorded status is the actual shell exit status parsed from the output's `Exit Code: N` / `Command exited with code N` marker (a nonzero bash exit returns ordinary text that `deerflow_tool_meta` still reports as success; local/e2b/opensandbox/tenki/boxlite all append the marker on nonzero exit with or without output, aio propagates the SDK's structured exit_code on both exec paths, and local timeouts append `Exit Code: 124` and signal kills parse as signed markers (`Exit Code: -9` records error), and `_truncate_bash_output` always preserves a trailing exit marker inside its budget — a 32-char floor raises any smaller configured limit, and the remote `Command exited with code N` form is accepted only as the whole trimmed output — so truncation cannot erase the failure; the matched marker text travels on the entry as `status_marker` so a `tests_passed` detail reports what was seen instead of asserting a failure the harness cannot distinguish from the command's own trailing text), falling back to the meta status only when no marker exists; matching is shell-structure aware (operator-separated segments — a physical newline separates with `;` semantics — or with the continuation operator the next line opens with (`cmd1\n&& cmd2` is `&&`; `cmd1\n|| cmd2` is `||`, which after a successful first command skips the rest while exiting 0, so flattening it to `;` would record a run that never happened) —, so a multi-line script's trailing `echo`/`seq` lines are never merged into the matched segment's arguments, status, or output attribution —, executable identity — directional: a bare criterion executable accepts any path spelling of the name, while an explicitly path-spelled criterion requires a path-spelled execution of the same normalized executable path (spelling judged on the raw token: `./pytest` names the project-local file and normpath collapsing `./` must not demote it to a PATH lookup; a `..` component on either side is unprovable outright — `link/../pytest` normalizes to `pytest` textually, but the OS follows `link` before resolving `..`, so lexical normalization cannot prove identity), so `/tmp/fake/pytest` cannot certify `/opt/project/.venv/bin/pytest` —, ordered argument subsequence whose env-assignment prefix must equal the criterion's exactly as an effective name → final-value mapping — extra, missing, or differently-valued assignments degrade the match, as does any reordering of a repeated name (`CI=0 CI=1` vs `CI=1 CI=0` are last-wins opposite environments), since no variable is provably inert across repositories (`CI`/`DEBUG` are routinely read by tests; `PATH`/`LD_PRELOAD`/`PYTEST_ADDOPTS`/`MAKEFILES` change what runs), and any assignment or argumented `export`/`unset` in a preceding segment is state pollution, and any span token carrying a runtime expansion (`$VAR`/`$( )`/backticks) or an extra token carrying glob metacharacters (crafted option-looking filenames narrow invisibly) is likewise unprovable), so a command that merely mentions the criterion string (`echo '12 passed'; # pytest x.py`) cannot anchor the leaf, and control flow is preserved — the matching span must end at the command's last segment with provable execution (`&&` needs recorded success, `||` needs recorded failure, pipelines inside the span and backgrounding are never provable), and the criterion's own connectors are preserved — an expected `&&` executed as `;` (`cd missing; pytest x` for `cd missing && pytest x`) lets a failed preceding step be bypassed, so it degrades; only the stricter direction (criterion `;` executed as `&&` with recorded success) survives), so a short-circuited segment (`false && pytest x; echo '3 passed'`) degrades to UNVERIFIED instead of a false hold; the summary shape is evaluated only when the output is attributable to the matched segment (every preceding segment provably silent by invocation form — only shape-free `cd dir` and pure assignments qualify; `pushd`/`umask`/`ulimit`, any `export`/`unset` (an invalid identifier prints bash's `not a valid identifier` error carrying subagent-chosen text — `export 'all tests passed'; make test` — and valid forms are state pollution), and any `source`/`.`, whose `*/bin/activate` path shape says nothing about what a crafted script prints, are not — so neither `echo '12 passed'; make test` nor a sourced forge can lend the shape; the `cd` print channel is closed too — CDPATH makes `cd` print the subagent-chosen resolved path and the pass shapes match as substrings, so a `cd` argument carrying a summary shape or runtime expansion is non-silent, and any `CDPATH=` assignment degrades the match as state pollution), pass shapes require a nonzero passed count (`0 passed`, go `[no test files]`, `Ran 0 tests` veto), argument matching treats tokens consumed by negating options (`--ignore`/`--deselect`/`--exclude`/`--skip` families) as ineligible evidence (`pytest --ignore tests/security tests` cannot affirm the excluded tests) and a target negated anywhere in the command degrades the match (`pytest tests/security tests/unit --ignore tests/security` cannot certify the excluded target), and a bare criterion with no positional selection target (bare `pytest`, `make test`) treats ANY negating option as a different selection (`pytest --ignore tests/security` never ran the default selection the criterion means), and accepts extra executed flags only when provably selection-preserving (verbosity/output/xdist/coverage forms; `-k`/`-m`/`--lf`/`--collect-only`/`-c`/`-p` and friends degrade to UNVERIFIED — `pytest -k smoke tests/security` cannot certify the full selection; extra positionals widen only after a path-scoped criterion — scoping reads positional targets by option arity, so a path embedded in an option (`--basetemp=/tmp/p`, `--junitxml=/tmp/r.xml`) never counts as a target, and an option of unknown arity (absent from the value-taking table, no glued `=`) followed by a path-like token (`--rootdir /tmp/project`) fails closed — the token may be the option's value, so the criterion keeps no scoped-selection proof —, so `python -m unittest pkg.OneTest` cannot certify bare `python -m unittest`), and criterion text is neutralized with `neutralize_untrusted_tags` before storage/rendering so a blocked tag in a criterion can never reach the lead-visible checklist section. Any other criterion is UNVERIFIED, never silently passed. The verdict (`subagent_acceptance_verdict` in `additional_kwargs`, validated on read) feeds the delegation ledger's `acceptance:` segment and the model-visible checklist section appended to the result text; the gateway strips caller-forged verdicts from both ledger entries and message metadata like the citation verdict. Leaf read errors catch only `OSError`/`SandboxError` — unexpected errors propagate to the task-tool-level isolation so the blocking-IO gate keeps teeth (`tests/blocking_io/test_task_tool_acceptance_checklist.py`). Known accepted boundaries (pinned by `TestKnownBoundaries` — conscious, do not re-raise): a bare criterion executable trusts PATH/filesystem spelling (a subagent-crafted same-name script is textually indistinguishable; spelling an absolute criterion path restricts evidence to that normalized path); relative criterion targets resolve in the wrapper's cwd, so `cd` prefixes must stay in scope (relative without `..`, or absolute under the thread data roots / virtual prefix) while a symlink inside an allowed root pointing out is a filesystem-layer concern; runner semantics are trusted (a Makefile swallowing failures, a runner exiting 0 on failure) — Layer 2 is execution evidence only, claim correctness belongs to the PR5 judge / RFC §6 re-execution; evidence is bounded (500-char command, 1000-char tail) and truncation degrades to UNVERIFIED rather than proving. +**Acceptance checklist path portability**: Paths are host-independent and raw `..` fails closed. Drive/UNC absolutes retain their class across `ntpath` normalization and cannot cross the root; drive-relative, shell-dependent provider/PSDrive, and POSIX-rooted `cd` on Windows are unprovable. Selection overlap separates pytest node IDs, normalizes safe `.`/duplicate separators, rejects `..`, and applies Windows casing to drive, UNC, PSDrive, or Windows-context paths; provider-qualified drive/UNC roots remain paths until a later node-ID split. POSIX paths and all node IDs keep their required case sensitivity. Ambiguous trailing-dot/space or 8.3 components, different volume IDs, cross-family pairs, and absolute/relative PSDrive pairs fail closed without execution/filesystem provenance; only same-volume, same-root-form paths compare lexically. Without shell provenance, raw commands reject backslashes; cmd `%VAR%`/`!VAR!`, `^`, `#`; Bash braces/tilde; and PowerShell splatting, typographic quotes, or unquoted parentheses. A lone `%` remains eligible. POSIX-only markers must short-circuit on Windows because pytest evaluates them at import. **Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out` **Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol. **Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.) diff --git a/backend/packages/harness/deerflow/subagents/acceptance_checks.py b/backend/packages/harness/deerflow/subagents/acceptance_checks.py index 65989ffe3..34a088607 100644 --- a/backend/packages/harness/deerflow/subagents/acceptance_checks.py +++ b/backend/packages/harness/deerflow/subagents/acceptance_checks.py @@ -55,10 +55,13 @@ the async caller offloads the whole check with ``asyncio.to_thread``. from __future__ import annotations +import ntpath import os +import posixpath import re import shlex import stat +import unicodedata from collections.abc import Callable, Mapping from typing import Any, TypedDict @@ -480,6 +483,39 @@ def _check_file_leaf( _SHELL_OPERATORS = ";&|" #: Leading ``VAR=value`` assignments are environment setup, not the executable. _ENV_ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=") +_WINDOWS_DRIVE_QUALIFIED_RE = re.compile(r"^[A-Za-z]:") +_WINDOWS_DRIVE_ABSOLUTE_RE = re.compile(r"^[A-Za-z]:/") +_WINDOWS_UNC_ABSOLUTE_RE = re.compile(r"^//[^/]+/[^/]+(?:/|$)") +#: A generated 8.3 component has at most six legal stem characters before a +#: one-digit ``~N`` tail and may retain an extension of at most three legal +#: characters. Lexical normalization cannot prove that it differs from the +#: corresponding long name on a volume where short-name generation is active. +_WINDOWS_SHORT_NAME_COMPONENT_RE = re.compile(r"^[A-Za-z0-9$%_'@~`!(){}^#&-]{1,6}~[1-9](?:\.[A-Za-z0-9$%_'@~`!(){}^#&-]{1,3})?$") +#: PowerShell paths may name a provider/PSDrive before ``:`` (for example, +#: ``FileSystem::C:/tmp`` or ``External:/tmp``). Those forms are absolute in +#: PowerShell but look relative to POSIX path normalization. +_POWERSHELL_DRIVE_QUALIFIED_RE = re.compile(r"^[^/\\:]+:") +#: A provider-qualified filesystem path carries either a drive designator or +#: a UNC root after PowerShell's ``Provider::`` prefix. Keep that delimiter +#: distinct from pytest's later ``::nodeid`` separator. +_POWERSHELL_PROVIDER_QUALIFIED_RE = re.compile(r"^[^/\\:]+::(?P(?:[^/\\:]+:|//))") +#: Unlike drive-letter paths, a named PSDrive may have a multi-character name. +#: A separator after the colon distinguishes root-anchored from drive-relative +#: spellings whose resolution depends on that PSDrive's remembered location. +_POWERSHELL_DRIVE_ABSOLUTE_RE = re.compile(r"^[^/\\:]+:/") +#: ``cmd.exe`` expands paired-percent environment references before running +#: the command. Without shell provenance, a token such as ``%TEMP%`` cannot +#: be treated as the literal relative path seen by the POSIX parser. +_CMD_ENV_EXPANSION_RE = re.compile(r"%[^%\r\n]+%") +_CMD_DELAYED_ENV_EXPANSION_RE = re.compile(r"![^!\r\n]+!") +#: Shells supported by ``LocalSandbox`` do not agree on Bash brace expansion. +#: The evidence currently records no shell kind, so matching must reject it +#: before a POSIX parser can turn the expression into one harmless-looking +#: token. +_BASH_BRACE_EXPANSION_RE = re.compile(r"\{[^{}\r\n]*(?:,|\.\.)[^{}\r\n]*\}") +#: PowerShell treats seven typographic single and double quotes as string +#: delimiters, while POSIX ``shlex`` retains them as ordinary token characters. +_POWERSHELL_QUOTE_DELIMITERS = frozenset("‘’‚‛“”„") def _carries_summary_shape(text: str) -> bool: @@ -490,6 +526,58 @@ def _carries_summary_shape(text: str) -> bool: return bool(_TEST_PASS_SHAPE_RE.search(text) or _TEST_FAIL_SHAPE_RE.search(text) or _TEST_ZERO_SHAPE_RE.search(text)) +def _normalize_cd_scope_path(path: str) -> tuple[str, bool] | None: + """Normalize a ``cd`` path and retain Windows-absolute provenance. + + ``posixpath.normpath`` drops the slash from a bare Windows drive root and + collapses traversal above that root into a relative-looking path. Handle + drive-qualified absolute and UNC paths component by component so those + spellings remain absolute, and fail closed when ``..`` would cross the + drive or share root. Drive-relative forms such as ``C:tmp`` also fail + closed because their resolution depends on the process's remembered + directory for that drive. + """ + slash_path = path.replace("\\", "/") + if slash_path.startswith(("//?/", "//./")): + # Win32 device namespaces can alias ordinary drive/UNC paths, but the + # lexical checker has no filesystem provenance to resolve them. + return None + if _POWERSHELL_DRIVE_QUALIFIED_RE.match(slash_path) and not _WINDOWS_DRIVE_QUALIFIED_RE.match(slash_path): + return None + if _WINDOWS_DRIVE_QUALIFIED_RE.match(slash_path) and not _WINDOWS_DRIVE_ABSOLUTE_RE.match(slash_path): + return None + if _WINDOWS_UNC_ABSOLUTE_RE.match(slash_path): + unc_parts = slash_path[2:].split("/") + share_root = unc_parts[:2] + parts: list[str] = [] + for part in unc_parts[2:]: + if not part or part == ".": + continue + if part == "..": + if not parts: + return None + parts.pop() + else: + parts.append(part) + normalized = "//" + "/".join((*share_root, *parts)) + return normalized, True + if not _WINDOWS_DRIVE_ABSOLUTE_RE.match(slash_path): + return posixpath.normpath(slash_path), False + + parts: list[str] = [] + for part in slash_path[3:].split("/"): + if not part or part == ".": + continue + if part == "..": + if not parts: + return None + parts.pop() + else: + parts.append(part) + normalized = slash_path[:2] + "/" + "/".join(parts) + return normalized, True + + def _cd_target_in_scope(target: str, thread_data: Mapping[str, Any] | None) -> bool: """Whether a preceding ``cd`` target provably keeps the criterion's relative path-like targets resolving inside the thread's data roots. @@ -508,16 +596,38 @@ def _cd_target_in_scope(target: str, thread_data: Mapping[str, Any] | None) -> b """ if not target or target == "-" or target.startswith("~"): return False - normalized = os.path.normpath(target.replace("\\", "/")) - if normalized.startswith("/"): + if ".." in target.replace("\\", "/").split("/"): + # Lexical cleanup is not proof of containment: the OS follows a + # directory symlink before resolving the following ``..`` component. + return False + # After shell-dependent spellings have failed closed, keep deterministic + # POSIX lexical semantics even when this checker itself runs on Windows. + normalized_target = _normalize_cd_scope_path(target) + if normalized_target is None: + return False + normalized, is_windows_absolute = normalized_target + if _thread_uses_windows_paths(thread_data) and normalized.startswith("/") and not is_windows_absolute: + # ``/mnt/...`` is absolute under Bash but drive-rooted under + # PowerShell. Without the executing shell, a Windows workspace cannot + # prove that the virtual POSIX spelling stayed inside its data root. + return False + if normalized.startswith("/") or is_windows_absolute: roots = [VIRTUAL_PATH_PREFIX] for key in ("workspace_path", "outputs_path", "uploads_path"): value = (thread_data or {}).get(key) if isinstance(value, str) and value: roots.append(value) for root in roots: - normalized_root = os.path.normpath(root.replace("\\", "/")) - if normalized == normalized_root or normalized.startswith(normalized_root + "/"): + normalized_root_result = _normalize_cd_scope_path(root) + if normalized_root_result is None: + continue + normalized_root, root_is_windows_absolute = normalized_root_result + use_windows_comparison = is_windows_absolute and root_is_windows_absolute + candidate_for_comparison = ntpath.normcase(normalized) if use_windows_comparison else normalized + root_for_comparison = ntpath.normcase(normalized_root) if use_windows_comparison else normalized_root + separator = "\\" if use_windows_comparison else "/" + root_prefix = root_for_comparison if root_for_comparison.endswith(separator) else root_for_comparison + separator + if candidate_for_comparison == root_for_comparison or candidate_for_comparison.startswith(root_prefix): return True return False return ".." not in normalized.split("/") @@ -605,26 +715,175 @@ def _negated_value(token: str) -> str: return token -def _negation_overlaps(criterion_token: str, negated_value: str) -> bool: +def _thread_uses_windows_paths(thread_data: Mapping[str, Any] | None) -> bool: + """Whether the thread roots establish Windows filesystem semantics.""" + for key in ("workspace_path", "outputs_path", "uploads_path"): + value = (thread_data or {}).get(key) + if not isinstance(value, str): + continue + slash_path = value.replace("\\", "/") + if _WINDOWS_DRIVE_ABSOLUTE_RE.match(slash_path) or _WINDOWS_UNC_ABSOLUTE_RE.match(slash_path): + return True + return False + + +def _selection_path_parts(value: str) -> tuple[str, str | None]: + """Split a runner selection into its filesystem path and pytest nodeid.""" + provider_match = _POWERSHELL_PROVIDER_QUALIFIED_RE.match(value) + nodeid_start = provider_match.end() if provider_match is not None else 0 + marker_index = value.find("::", nodeid_start) + if marker_index < 0: + return value, None + return value[:marker_index], value[marker_index + 2 :] + + +def _without_powershell_provider(path: str) -> str: + """Return the rooted portion of a provider-qualified filesystem path.""" + provider_match = _POWERSHELL_PROVIDER_QUALIFIED_RE.match(path) + if provider_match is None: + return path + return path[provider_match.start("path") :] + + +def _has_parent_path_component(value: str) -> bool: + path, _nodeid = _selection_path_parts(value) + return ".." in path.replace("\\", "/").split("/") + + +def _uses_windows_selection_semantics(value: str, *, windows_path_context: bool) -> bool: + path, _nodeid = _selection_path_parts(value) + slash_path = _without_powershell_provider(path).replace("\\", "/") + return bool(windows_path_context or _POWERSHELL_DRIVE_QUALIFIED_RE.match(slash_path) or _WINDOWS_UNC_ABSOLUTE_RE.match(slash_path)) + + +def _selection_path_kind(value: str) -> str: + path, _nodeid = _selection_path_parts(value) + slash_path = _without_powershell_provider(path).replace("\\", "/") + if _WINDOWS_DRIVE_ABSOLUTE_RE.match(slash_path): + return "windows_drive_absolute" + if _WINDOWS_UNC_ABSOLUTE_RE.match(slash_path): + return "windows_unc_absolute" + if slash_path.startswith("/"): + return "posix_absolute" + if _POWERSHELL_DRIVE_QUALIFIED_RE.match(slash_path) and not _WINDOWS_DRIVE_QUALIFIED_RE.match(slash_path): + if _POWERSHELL_DRIVE_ABSOLUTE_RE.match(slash_path): + return "powershell_drive_absolute" + return "powershell_drive_relative" + return "relative" + + +def _windows_volume_identifier(value: str) -> tuple[str, str] | None: + """Return a comparable drive, PSDrive, or UNC-root identifier.""" + path, _nodeid = _selection_path_parts(value) + slash_path = _without_powershell_provider(path).replace("\\", "/") + if _WINDOWS_DRIVE_QUALIFIED_RE.match(slash_path): + return "drive", ntpath.normcase(slash_path[:2]) + if _WINDOWS_UNC_ABSOLUTE_RE.match(slash_path): + server, share, *_rest = slash_path[2:].split("/") + return "unc", ntpath.normcase(f"//{server}/{share}") + if _POWERSHELL_DRIVE_QUALIFIED_RE.match(slash_path): + drive, _separator, _rest = slash_path.partition(":") + return "psdrive", ntpath.normcase(drive) + return None + + +def _has_ambiguous_windows_component(value: str, *, windows_path_context: bool) -> bool: + """Whether ordinary Win32 cleanup may alias a textual path component. + + Windows APIs normally discard trailing spaces and periods from path + components, while extended-length paths can preserve them. Generated 8.3 + short names can also identify a longer component without any lexical + relationship. Without shell and filesystem provenance, these spellings + must make an overlap decision fail closed. + """ + if not _uses_windows_selection_semantics(value, windows_path_context=windows_path_context): + return False + path, _nodeid = _selection_path_parts(value) + path = _without_powershell_provider(path) + return any(component not in {"", ".", ".."} and (component.endswith((" ", ".")) or _WINDOWS_SHORT_NAME_COMPONENT_RE.fullmatch(component)) for component in path.replace("\\", "/").split("/")) + + +def _normalize_selection_path(value: str, *, windows_path_context: bool) -> tuple[str, str | None, bool]: + """Normalize safe lexical aliases without erasing pytest nodeid case. + + Parent traversal is rejected by the caller because resolving ``..`` + lexically can cross symlinks. ``.`` and duplicate separators are safe to + collapse. Drive-qualified and UNC paths always use ``ntpath`` semantics; + when the thread roots establish a Windows workspace, every runner path + does, including drive-rooted ``/tests`` and mapped virtual paths. + """ + path, nodeid = _selection_path_parts(value) + slash_path = _without_powershell_provider(path).replace("\\", "/") + is_windows_path = _uses_windows_selection_semantics(value, windows_path_context=windows_path_context) + if is_windows_path: + normalized = ntpath.normcase(ntpath.normpath(slash_path)) + else: + normalized = posixpath.normpath(slash_path) + return normalized, nodeid, is_windows_path + + +def _nodeids_overlap(a: str | None, b: str | None) -> bool: + if a is None or b is None: + return True + return a == b or a.startswith(b + "::") or b.startswith(a + "::") + + +def _negation_overlaps(criterion_token: str, negated_value: str, *, windows_path_context: bool = False) -> bool: """Whether a negated value overlaps a matched criterion target: equal, or one nested under the other at a path boundary (``tests`` vs ``tests/unit/test_auth.py``) or a pytest nodeid boundary (``tests/x.py`` vs ``tests/x.py::test_y``). Overlap means part of the criterion's selection never ran, so the passing summary may not cover it; unrelated exclusions (``--ignore tests/slow`` against ``pytest tests/unit``) do not - overlap and keep matching.""" - a = criterion_token.replace("\\", "/").removeprefix("./").rstrip("/") - b = negated_value.replace("\\", "/").removeprefix("./").rstrip("/") - if not a or not b: + overlap and keep matching. Safe lexical aliases are normalized, while + mixed path families fail closed because evidence does not carry enough + filesystem provenance to prove that their spellings are distinct.""" + a_kind = _selection_path_kind(criterion_token) + b_kind = _selection_path_kind(negated_value) + if a_kind != b_kind: + # Relative selections resolve against an unrecorded cwd, and Windows + # can resolve POSIX-rooted spellings against the current drive or map + # a drive onto a UNC share. Cross-family spellings can therefore alias + # even when their lexical prefixes differ. + return True + a_volume = _windows_volume_identifier(criterion_token) + b_volume = _windows_volume_identifier(negated_value) + if a_volume is not None and b_volume is not None and a_volume != b_volume: + # Distinct drive letters can alias through SUBST or mapped drives, and + # distinct UNC roots can alias through DFS, DNS, or share mappings. + # The execution evidence records none of that volume provenance. + return True + if a_kind == "relative": + a_path, _a_nodeid = _selection_path_parts(criterion_token) + b_path, _b_nodeid = _selection_path_parts(negated_value) + a_is_drive_relative = bool(_WINDOWS_DRIVE_QUALIFIED_RE.match(a_path.replace("\\", "/"))) + b_is_drive_relative = bool(_WINDOWS_DRIVE_QUALIFIED_RE.match(b_path.replace("\\", "/"))) + if a_is_drive_relative != b_is_drive_relative: + # ``tests/x`` and ``D:tests/x`` can name the same path when the + # process cwd and D:'s remembered cwd coincide. Neither value is + # absolute, and the execution evidence carries neither cwd. + return True + a, a_nodeid, a_is_windows = _normalize_selection_path(criterion_token, windows_path_context=windows_path_context) + b, b_nodeid, b_is_windows = _normalize_selection_path(negated_value, windows_path_context=windows_path_context) + if not a or not b or a == "." or b == ".": return False - return a == b or a.startswith((b + "/", b + "::")) or b.startswith((a + "/", a + "::")) + if a == b: + return _nodeids_overlap(a_nodeid, b_nodeid) + if a_is_windows != b_is_windows: + # A drive-relative spelling can resolve to the same path as an + # ordinary relative token, but the per-drive cwd is not recorded. + return True + separator = "\\" if a_is_windows else "/" + a_prefix = a if a.endswith(separator) else a + separator + b_prefix = b if b.endswith(separator) else b + separator + return a.startswith(b_prefix) or b.startswith(a_prefix) def _normalize_command(command: str) -> str: return " ".join(command.split()) -def _shell_parse_line(line: str) -> tuple[str | None, list[list[str]], list[str]] | None: +def _shell_parse_line(line: str, *, posix: bool = True) -> tuple[str | None, list[list[str]], list[str]] | None: """Tokenize one physical line into segments plus the operators joining them. Returns ``(leading_op, segments, ops)``: ``ops[i]`` is the operator @@ -636,9 +895,11 @@ def _shell_parse_line(line: str) -> tuple[str | None, list[list[str]], list[str] ``;`` would overstate what provably ran. Comments are stripped (a ``# pytest ...`` remark executes nothing) and quotes are honored, so an operator inside an argument cannot split a segment. Returns ``None`` on - malformed shell (unbalanced quotes). + malformed shell (unbalanced quotes). ``posix=False`` is reserved for the + raw-token safety pass: it keeps backslashes and surrounding quotes visible + before the normal POSIX parse can consume them as escaping syntax. """ - lexer = shlex.shlex(line, posix=True, punctuation_chars=_SHELL_OPERATORS) + lexer = shlex.shlex(line, posix=posix, punctuation_chars=_SHELL_OPERATORS) lexer.whitespace_split = True lexer.commenters = "#" try: @@ -670,7 +931,7 @@ def _shell_parse_line(line: str) -> tuple[str | None, list[list[str]], list[str] return leading_op, segments, ops -def _shell_parse(command: str) -> tuple[list[list[str]], list[str]] | None: +def _shell_parse(command: str, *, posix: bool = True) -> tuple[list[list[str]], list[str]] | None: """Tokenize a shell command into segments plus the operators joining them. Physical newlines are command separators with ``;`` semantics — bash @@ -692,7 +953,7 @@ def _shell_parse(command: str) -> tuple[list[list[str]], list[str]] | None: segments: list[list[str]] = [] ops: list[str] = [] for line in command.split("\n"): - parsed = _shell_parse_line(line) + parsed = _shell_parse_line(line, posix=posix) if parsed is None: return None leading_op, line_segments, line_ops = parsed @@ -709,6 +970,142 @@ def _shell_parse(command: str) -> tuple[list[list[str]], list[str]] | None: return segments, ops +def _has_shell_ambiguous_whitespace(command: str) -> bool: + """Whether whitespace can split differently across supported shells. + + ASCII space, tab, LF, and CRLF are shared separators. PowerShell also + separates on bare CR, vertical tab, form feed, NEL, and Unicode separator + characters, while Python's POSIX ``shlex`` can retain them inside a token. + """ + command_without_crlf = command.replace("\r\n", "\n") + return any(character in "\r\v\f\x85" or (character != " " and unicodedata.category(character) in {"Zs", "Zl", "Zp"}) for character in command_without_crlf) + + +def _has_cmd_control_operator_in_single_quotes(command: str) -> bool: + """Whether POSIX single quotes hide cmd.exe control syntax. + + Cmd does not use single quotes for grouping, so metacharacters within them + remain active there. Ordinary single-quoted text stays verifiable for the + POSIX execution path. Shell-specific ways to escape quotes are already + rejected by the other provenance checks below. + """ + in_double_quotes = False + in_single_quotes = False + for character in command: + if character == '"': + in_double_quotes = not in_double_quotes + elif character == "'" and not in_double_quotes: + in_single_quotes = not in_single_quotes + elif in_single_quotes and character in "&|<>()": + return True + return False + + +def _has_unquoted_parenthesis(command: str) -> bool: + """Whether a parenthesis appears outside a quoted string. + + PowerShell evaluates an unquoted parenthesized command expression and + expands its output into native arguments. POSIX tokenization instead + leaves the parentheses attached to ordinary tokens, which can hide an + injected runner option. Parentheses inside double- or single-quoted + strings are data; cmd-specific control syntax inside single quotes is + rejected separately because cmd does not honor those quotes. + """ + in_double_quotes = False + in_single_quotes = False + for character in command: + if character == '"' and not in_single_quotes: + in_double_quotes = not in_double_quotes + elif character == "'" and not in_double_quotes: + in_single_quotes = not in_single_quotes + elif character in "()" and not in_double_quotes and not in_single_quotes: + return True + return False + + +def _has_unquoted_single_quote(text: str) -> bool: + """Whether *text* contains a single quote outside double quotes. + + The raw ``posix=False`` token stream retains surrounding quotes. Cmd.exe + passes a POSIX single quote to the child process instead of using it for + grouping, so such a token cannot be compared with its quote-stripped + POSIX form without knowing which shell executed it. + """ + in_double_quotes = False + for character in text: + if character == '"': + in_double_quotes = not in_double_quotes + elif character == "'" and not in_double_quotes: + return True + return False + + +def _raw_segment_has_unquoted_single_quote(segment: list[str]) -> bool: + return any(_has_unquoted_single_quote(token) for token in segment) + + +def _matched_span_has_ambiguous_single_quotes( + expected_raw: list[list[str]], + actual_raw: list[list[str]], + actual: list[list[str]], + *, + start: int, + span: int, + thread_data: Mapping[str, Any] | None, +) -> bool: + """Whether cmd.exe single-quote semantics can change a candidate match. + + Every expected segment and the matching actual span define the invocation + being certified, so a POSIX single quote in any of them is ambiguous. + Before the span, only segments accepted as provably silent can contribute + to a successful match; a quoted ``cd`` or assignment can change the state + in which the runner executes and therefore also requires shell provenance. + Non-silent prefixes remain governed by output-attribution checks, preserving + authoritative failure results such as ``echo '12 passed'; make test``. + """ + if any(_raw_segment_has_unquoted_single_quote(segment) for segment in expected_raw): + return True + if any(_raw_segment_has_unquoted_single_quote(segment) for segment in actual_raw[start : start + span]): + return True + return any(_raw_segment_has_unquoted_single_quote(actual_raw[index]) and _is_silent_segment(actual[index], thread_data) for index in range(start)) + + +def _command_requires_shell_provenance(command: str) -> bool: + """Whether raw command tokens have shell-dependent Windows semantics. + + The acceptance evidence does not currently record which shell executed a + command. Inspect a non-POSIX tokenization before the authoritative POSIX + parser can discard backslashes anywhere in the candidate command: PowerShell + and native Windows runners preserve them as path separators while POSIX + shells treat them as escapes. Cmd percent/bang references and ``^``, Bash + tilde/brace expansion, and PowerShell splatting, typographic quotes, or + unquoted parentheses can all rewrite the native argv differently. ``#`` + starts a comment for the POSIX parser but is an ordinary argument to + cmd.exe. Cmd also does not treat single quotes as quoting, so cmd control + syntax hidden inside POSIX single quotes is unsafe. PowerShell recognizes + more separators than POSIX ``shlex``, including bare carriage returns and + Unicode separator characters; normal CRLF remains unambiguous. These forms + fail closed rather than certifying arguments that depend on an unknown + shell. + """ + if ( + _has_shell_ambiguous_whitespace(command) + or _has_cmd_control_operator_in_single_quotes(command) + or "#" in command + or "^" in command + or _CMD_DELAYED_ENV_EXPANSION_RE.search(command) + or _BASH_BRACE_EXPANSION_RE.search(command) + or any(quote in command for quote in _POWERSHELL_QUOTE_DELIMITERS) + or _has_unquoted_parenthesis(command) + ): + return True + parsed = _shell_parse(command, posix=False) + if parsed is None: + return "\\" in command or bool(_CMD_ENV_EXPANSION_RE.search(command)) + segments, _ops = parsed + return any("\\" in token or token.startswith(("~", "@")) or (token.startswith("(") and token.endswith(")")) or _CMD_ENV_EXPANSION_RE.search(token) for segment in segments for token in segment) + + def _strip_env_assignments(tokens: list[str]) -> list[str]: index = 0 while index < len(tokens) and _ENV_ASSIGNMENT_RE.match(tokens[index]): @@ -878,7 +1275,7 @@ def _normalize_executable(token: str) -> str: return os.path.normpath(token.replace("\\", "/")) -def _segment_matches(expected: list[str], actual: list[str]) -> str: +def _segment_matches(expected: list[str], actual: list[str], *, windows_path_context: bool = False) -> str: """Match one segment against the criterion's, classifying extra flags. Returns ``"match"`` when the executable agrees — directional: a bare @@ -1003,7 +1400,14 @@ def _segment_matches(expected: list[str], actual: list[str]) -> str: # before resolving ``..``, so a textually unrelated value can name # the criterion's target). return "unprovable" - if any(_negation_overlaps(actual[position], value) for position in consumed if position != 0 for value in negated_values): + compared_values = [actual[position] for position in consumed if position != 0] + negated_values + if negated_values and any(_has_ambiguous_windows_component(value, windows_path_context=windows_path_context) for value in compared_values): + return "unprovable" + if negated_values and any(_has_parent_path_component(actual[position]) for position in consumed if position != 0): + # The matched target itself may normalize onto an excluded path, but + # collapsing its ``..`` components would be unsound across symlinks. + return "unprovable" + if any(_negation_overlaps(actual[position], value, windows_path_context=windows_path_context) for position in consumed if position != 0 for value in negated_values): return "unprovable" for position, token in enumerate(actual): if position in consumed or position in negated or position in option_positions: @@ -1096,7 +1500,13 @@ def _criterion_connectors_preserved(expected_ops: list[str], executed_within_ops return True -def _commands_match(criterion_command: str, executed_command: str, *, executed_success: bool) -> str: +def _commands_match( + criterion_command: str, + executed_command: str, + *, + executed_success: bool, + thread_data: Mapping[str, Any] | None = None, +) -> str: """Shell-structure match with control-flow attribution. Returns ``"match"`` when the criterion's segment sequence appears as @@ -1110,23 +1520,35 @@ def _commands_match(criterion_command: str, executed_command: str, *, executed_s """ expected_parsed = _shell_parse(criterion_command) actual_parsed = _shell_parse(executed_command) + expected_raw_parsed = _shell_parse(criterion_command, posix=False) + actual_raw_parsed = _shell_parse(executed_command, posix=False) + needs_shell_provenance = _command_requires_shell_provenance(criterion_command) or _command_requires_shell_provenance(executed_command) if expected_parsed is None or actual_parsed is None: # Malformed shell: only exact normalized equality survives. expected_norm = _normalize_command(criterion_command) - return "match" if expected_norm and expected_norm == _normalize_command(executed_command) else "no_match" + if not expected_norm or expected_norm != _normalize_command(executed_command): + return "no_match" + has_ambiguous_single_quote = _has_unquoted_single_quote(criterion_command) or _has_unquoted_single_quote(executed_command) + return "unprovable" if needs_shell_provenance or has_ambiguous_single_quote else "match" expected, expected_ops = expected_parsed actual, ops = actual_parsed if not expected or not actual or len(expected) > len(actual): return "no_match" + raw_segments_align = expected_raw_parsed is not None and actual_raw_parsed is not None and len(expected_raw_parsed[0]) == len(expected) and len(actual_raw_parsed[0]) == len(actual) + if not raw_segments_align: + needs_shell_provenance = True + expected_raw = expected_raw_parsed[0] if raw_segments_align and expected_raw_parsed is not None else [] + actual_raw = actual_raw_parsed[0] if raw_segments_align and actual_raw_parsed is not None else [] span = len(expected) saw_unprovable = False + windows_path_context = _thread_uses_windows_paths(thread_data) for start in range(len(actual) - span + 1): if any(_segment_pollutes_state(segment) for segment in actual[:start]): # A preceding segment mutated shell state the matcher cannot see # (PATH/exports): nothing later is provable. saw_unprovable = True continue - outcomes = [_segment_matches(expected[i], actual[start + i]) for i in range(span)] + outcomes = [_segment_matches(expected[i], actual[start + i], windows_path_context=windows_path_context) for i in range(span)] if any(outcome == "no_match" for outcome in outcomes): continue if any(outcome == "unprovable" for outcome in outcomes): @@ -1143,6 +1565,16 @@ def _commands_match(criterion_command: str, executed_command: str, *, executed_s saw_unprovable = True continue if _span_attributable(ops[:start], ops[start : start + span - 1], ops[start + span - 1 :], executed_success): + if needs_shell_provenance or _matched_span_has_ambiguous_single_quotes( + expected_raw, + actual_raw, + actual, + start=start, + span=span, + thread_data=thread_data, + ): + saw_unprovable = True + continue return "match" saw_unprovable = True return "unprovable" if saw_unprovable else "no_match" @@ -1190,7 +1622,12 @@ def _check_tests_passed_leaf(command: str, bash_executions: list[dict[str, Any]] matches: list[tuple[str, dict[str, Any]]] = [] for execution in bash_executions or []: status = str(execution.get("status") or "") - outcome = _commands_match(command, str(execution.get("command") or ""), executed_success=status == "success") + outcome = _commands_match( + command, + str(execution.get("command") or ""), + executed_success=status == "success", + thread_data=thread_data, + ) if outcome == "match" and execution.get("command_truncated"): # The recorded command lost its suffix to the evidence cap; a # selection-changing tail (``-k smoke``) may have been cut away. diff --git a/backend/tests/test_acceptance_checks.py b/backend/tests/test_acceptance_checks.py index 011b7f4a9..1a6c920aa 100644 --- a/backend/tests/test_acceptance_checks.py +++ b/backend/tests/test_acceptance_checks.py @@ -24,6 +24,12 @@ THREAD_DATA = { "outputs_path": "/ws/thread/user-data/outputs", } +WINDOWS_THREAD_DATA = { + "workspace_path": "D:/WS", + "uploads_path": "D:/WS/uploads", + "outputs_path": "D:/WS/outputs", +} + def _reader(files: dict[str, str]): def read(_runtime, path: str) -> str: @@ -124,7 +130,7 @@ class TestFileLeaves: from deerflow.sandbox.tools import _resolve_local_read_path - assert Path(_resolve_local_read_path(seen[0], THREAD_DATA)) == Path("/ws/thread/user-data/outputs/report.md") # type: ignore[arg-type] + assert Path(_resolve_local_read_path(seen[0], THREAD_DATA)) == Path("/ws/thread/user-data/outputs/report.md").resolve() # type: ignore[arg-type] def test_non_empty_fails_on_empty_file(self): files = {"/mnt/user-data/outputs/report.md": ""} @@ -656,7 +662,7 @@ class TestProbeInnerScriptRealLayouts: (outputs / "report.md").write_text("hello", encoding="utf-8") assert self._run_read_probe(str(outputs / "report.md"), str(outputs)) == "READABLE" - @pytest.mark.skipif(os.geteuid() == 0, reason="root reads through mode-000") + @pytest.mark.skipif(os.name == "nt" or os.geteuid() == 0, reason="mode-000 readability needs POSIX permissions and a non-root euid") def test_read_probe_mode_000_is_unreadable(self, tmp_path): outputs = tmp_path / "outputs" outputs.mkdir() @@ -1156,6 +1162,120 @@ class TestTestsPassedLeaf: assert verdict["leaves"][0]["holds"] is True + def test_crlf_execution_with_test_command_last_still_matches(self): + """A normal Windows CRLF command record has the same boundaries as + its LF equivalent and must not require shell provenance.""" + executions = [_bash_execution("cd backend\r\npytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_bare_carriage_return_is_unprovable_before_powershell_line_split(self): + """PowerShell treats a standalone CR as a command boundary, while a + POSIX parser may absorb it as whitespace and attribute a forged + passing summary to the preceding test run.""" + executions = [_bash_execution("pytest tests/security\rWrite-Output '3 passed'", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize("separator", ("\v", "\f", "\x85", "\u00a0", "\u2028", "\u2029")) + def test_powershell_only_whitespace_cannot_hide_runner_arguments(self, separator): + """PowerShell separates arguments on these characters, while POSIX + shlex can absorb them into one apparently harmless token.""" + command = f"pytest tests/security tests/unit{separator}--ignore{separator}tests/security" + executions = [_bash_execution(command, output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cmd_single_quote_cannot_hide_a_control_operator(self): + """cmd.exe does not quote ``&`` with single quotes, even though POSIX + shlex would hide the operator inside one argument.""" + executions = [_bash_execution("pytest tests/security '& echo 3 passed'", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cmd_single_quotes_cannot_change_matched_runner_argv(self): + """POSIX removes ordinary single quotes, while cmd.exe passes them + through as part of the runner argument.""" + executions = [_bash_execution("pytest 'tests/security'", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cmd_single_quotes_in_criterion_are_unprovable(self): + executions = [_bash_execution("pytest tests/security", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest 'tests/security'"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cmd_single_quotes_in_preceding_cd_are_unprovable(self): + """A quoted POSIX ``cd`` target is not the same path under cmd.exe, + so it cannot establish the runner's working directory.""" + executions = [_bash_execution("cd '/mnt/user-data/workspace' && pytest tests/security", output_tail="3 passed")] + verdict = check_acceptance_criteria( + ["tests_passed:pytest tests/security"], + thread_data=THREAD_DATA, + bash_executions=executions, + ) + + assert verdict["leaves"][0]["checked"] is False + + def test_mismatched_posix_and_raw_segments_are_unprovable(self): + """A tokenizer disagreement must fail closed instead of indexing a + raw segment list that could not be produced.""" + executions = [_bash_execution('cd foo";"bar && pytest tests/', output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], bash_executions=executions) + + leaf = verdict["leaves"][0] + assert leaf["checked"] is False + assert leaf["holds"] is False + + def test_powershell_parenthesized_expression_cannot_inject_runner_arguments(self): + """PowerShell expands a parenthesized command expression into native + arguments, while POSIX tokenization leaves the closing parenthesis on + the exclusion path and can miss that the required target was skipped.""" + command = 'pytest tests/security tests/unit (Write-Output "--ignore" "tests/security")' + executions = [_bash_execution(command, output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + leaf = verdict["leaves"][0] + assert leaf["checked"] is False + assert leaf["holds"] is False + + @pytest.mark.parametrize( + ("opening_quote", "closing_quote"), + [("“", "”"), ("‘", "’"), ("‚", "‚"), ("‛", "‛"), ("„", "„")], + ) + def test_powershell_quote_delimiters_cannot_hide_runner_exclusions(self, opening_quote: str, closing_quote: str): + command = f"pytest tests/security tests/unit {opening_quote}--deselect=tests/security/test_auth.py::test_required{closing_quote}" + executions = [_bash_execution(command, output_tail="3 passed, 1 deselected")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + leaf = verdict["leaves"][0] + assert leaf["checked"] is False + assert leaf["holds"] is False + + def test_shared_whitespace_and_double_quotes_remain_verifiable(self): + executions = [_bash_execution('pytest\t"tests/security" -q', output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_parentheses_inside_double_quoted_path_remain_verifiable(self): + executions = [_bash_execution('pytest "tests/(security)"', output_tail="3 passed")] + verdict = check_acceptance_criteria(['tests_passed:pytest "tests/(security)"'], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_apostrophe_inside_double_quotes_remains_verifiable(self): + executions = [_bash_execution('pytest "tests/O\'Brien" -q', output_tail="3 passed")] + verdict = check_acceptance_criteria(['tests_passed:pytest "tests/O\'Brien"'], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + def test_multiline_background_operator_stays_unprovable(self): """A trailing ``&`` at end of a line still separates (and backgrounds) the next line's command.""" @@ -1190,6 +1310,151 @@ class TestTestsPassedLeaf: assert verdict["leaves"][0]["checked"] is False + @pytest.mark.parametrize( + "target", + ( + r"..\..\tmp", + r"'\mnt\user-data\workspace\fake'", + ), + ) + def test_cd_with_backslashes_is_unprovable_before_posix_tokenization(self, target): + """Shell provenance is absent, so a backslash-bearing ``cd`` target + cannot be interpreted safely as either POSIX escaping or a Windows + path separator.""" + executions = [_bash_execution(f"cd {target} && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_runner_backslash_negation_is_unprovable_before_posix_tokenization(self): + """PowerShell preserves the separator in ``tests\\security``, while + POSIX ``shlex`` consumes it and could hide that the required target was + excluded from the recorded run.""" + executions = [_bash_execution(r"pytest tests/security tests/unit --ignore tests\security", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cmd_hash_argument_is_unprovable_before_comment_stripping(self): + """``cmd.exe`` passes ``#`` as an ordinary argument, while POSIX + parsing treats it as the start of a comment and can hide a following + exclusion from the acceptance matcher.""" + executions = [_bash_execution("pytest tests/security tests/unit # --ignore tests/security", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cmd_caret_escape_is_unprovable_before_posix_tokenization(self): + """``cmd.exe`` removes ``^`` escaping before invoking the runner, so + the literal token seen by the matcher may hide an exclusion alias.""" + executions = [_bash_execution("pytest tests/security tests/unit --ignore tests^/security", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize( + "command", + ( + "pytest tests/security tests/unit {--ignore,tests/security}", + 'pytest tests/security tests/unit @("--ignore","tests/security")', + 'pytest tests/security tests/unit ("--ignore","tests/security")', + 'pytest tests/security tests/unit @("--ignore=tests/security")', + 'pytest tests/security tests/unit ("--ignore=tests/security")', + "pytest tests/security tests/unit @pytestArgs", + "pytest tests/security tests/unit --ignore ~/repo/tests/security", + "pytest tests/security tests/unit --ignore !TARGET!", + ), + ) + def test_shell_expansion_syntax_is_unprovable_before_tokenization(self, command): + """Bash brace/tilde expansion and PowerShell splatting/arrays can + inject or rewrite runner arguments before the process is launched.""" + executions = [_bash_execution(command, output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize("target", ("%TEMP%", "%USERPROFILE%/fake")) + def test_cd_with_cmd_environment_expansion_is_unprovable(self, target): + executions = [_bash_execution(f"cd {target} && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cd_with_unpaired_percent_remains_a_literal_relative_path(self): + executions = [_bash_execution("cd reports/100% && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_cd_to_out_of_scope_windows_drive_path_is_unprovable(self): + thread_data = { + "workspace_path": "D:/ws/thread/user-data/workspace", + "outputs_path": "D:/ws/thread/user-data/outputs", + } + executions = [_bash_execution("cd D:/tmp/fake && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize("target", ("FileSystem::C:/tmp", "External:/tmp")) + def test_cd_to_powershell_provider_or_psdrive_path_is_unprovable(self, target): + """Provider-qualified and named-PSDrive paths are absolute in + PowerShell but look relative to the POSIX path normalizer.""" + thread_data = { + "workspace_path": "D:/ws/thread/user-data/workspace", + "outputs_path": "D:/ws/thread/user-data/outputs", + } + executions = [_bash_execution(f"cd {target} && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize("target", ("C:", "C:tmp", "C:../tmp")) + def test_cd_to_windows_drive_relative_path_is_unprovable(self, target): + thread_data = { + "workspace_path": "D:/ws/thread/user-data/workspace", + "outputs_path": "D:/ws/thread/user-data/outputs", + } + executions = [_bash_execution(f"cd {target} && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize("target", ("C:/", "C:/../tmp", "D:/ws/../../tmp")) + def test_cd_to_windows_drive_root_or_above_root_is_unprovable(self, target): + thread_data = { + "workspace_path": "D:/ws/thread/user-data/workspace", + "outputs_path": "D:/ws/thread/user-data/outputs", + } + executions = [_bash_execution(f"cd {target} && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cd_to_in_scope_windows_drive_path_is_case_insensitive(self): + thread_data = { + "workspace_path": "D:/ws/thread/user-data/workspace", + "outputs_path": "D:/ws/thread/user-data/outputs", + } + executions = [_bash_execution("cd d:/WS/thread/user-data/workspace/project && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_cd_to_windows_unicode_casefold_collision_is_unprovable(self): + thread_data = {"workspace_path": "D:/ws/Straße/workspace"} + executions = [_bash_execution("cd d:/WS/STRASSE/workspace/fake && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_cd_to_child_of_windows_drive_root_is_in_scope(self): + thread_data = {"workspace_path": "D:/"} + executions = [_bash_execution("cd d:/project && pytest tests/", output_tail="3 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + @pytest.mark.parametrize( "wrapped", ( @@ -1197,6 +1462,7 @@ class TestTestsPassedLeaf: "cd && pytest tests/", # bare cd goes HOME "cd - && pytest tests/", # prints OLDPWD "cd backend/../../x && pytest tests/", # lexical walk-out + "cd linked/../safe && pytest tests/", # ``..`` can cross a directory symlink "cd /mnt/user-data/../etc && pytest tests/", # normalized escape "cd /ws/thread/user-data/workspace2 && pytest tests/", # sibling of an allowed root ), @@ -1557,6 +1823,370 @@ class TestTestsPassedLeaf: assert leaf["checked"] is False assert leaf["detail"] == "matching segment cannot be proven to have executed" + def test_windows_case_alias_of_target_that_is_excluded_is_unprovable(self): + """Windows drive paths are case-insensitive for overlap purposes, so + a differently-cased exclusion can still remove the required target.""" + executions = [_bash_execution("pytest D:/WS/tests/security D:/WS/tests/unit --ignore d:/ws/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest D:/WS/tests/security"], bash_executions=executions) + + leaf = verdict["leaves"][0] + assert leaf["checked"] is False + assert leaf["detail"] == "matching segment cannot be proven to have executed" + + def test_windows_case_alias_exclusion_nested_under_target_is_unprovable(self): + executions = [_bash_execution("pytest D:/WS/tests --ignore d:/ws/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest D:/WS/tests"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_unrelated_windows_exclusion_keeps_matching(self): + target = "D:/WS/tests/security" + executions = [_bash_execution(f"pytest {target} D:/WS/tests/unit --ignore d:/ws/tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_different_absolute_drive_exclusion_fails_closed_without_thread_context(self): + executions = [_bash_execution("pytest D:/ws/tests/security D:/ws/tests/unit --ignore E:/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest D:/ws/tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_different_unc_root_exclusion_fails_closed_without_thread_context(self): + target = "//server1/share/tests/security" + executions = [_bash_execution(f"pytest {target} //server1/share/tests/unit --ignore //server2/share/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_unrelated_exclusion_on_same_unc_root_keeps_matching(self): + target = "//server/share/tests/security" + executions = [_bash_execution(f"pytest {target} //server/share/tests/unit --ignore //server/share/tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_different_psdrive_exclusion_fails_closed_without_thread_context(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore Mirror:/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_psdrive_exclusion_fails_closed_without_thread_context(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore FileSystem::Mirror:/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_psdrive_relative_exclusion_fails_closed_without_thread_context(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore Data:tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_psdrive_relative_exclusion_fails_closed(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore FileSystem::data:tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_psdrive_absolute_exclusion_fails_closed_for_relative_criterion(self): + target = "Data:tests/security" + executions = [_bash_execution(f"pytest {target} Data:tests/unit --ignore Data:/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_psdrive_relative_deselect_fails_closed_for_absolute_nodeid(self): + target = "Data:/tests/security/test_x.py::test_a" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --deselect Data:tests/security/test_x.py", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_unrelated_exclusion_on_same_relative_psdrive_keeps_matching(self): + target = "Data:tests/security" + executions = [_bash_execution(f"pytest {target} Data:tests/unit --ignore data:tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_unrelated_exclusion_on_same_psdrive_keeps_matching(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore data:/tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_same_psdrive_case_alias_exclusion_is_unprovable(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore data:/TESTS/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_path_on_same_psdrive_compares_lexically(self): + target = "Data:/tests/security" + executions = [_bash_execution(f"pytest {target} Data:/tests/unit --ignore FileSystem::data:/tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_provider_qualified_psdrive_preserves_pytest_nodeid_boundary(self): + target = "Data:/tests/x.py::TestA" + executions = [_bash_execution(f"pytest {target} Data:/tests/y.py --deselect FileSystem::data:/tests/x.py::TestA", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_psdrive_keeps_pytest_nodeid_case_sensitive(self): + target = "Data:/tests/x.py::TestA" + executions = [_bash_execution(f"pytest {target} Data:/tests/y.py --deselect FileSystem::data:/tests/X.PY::testa", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_provider_qualified_unc_case_alias_exclusion_fails_closed(self): + target = "FileSystem:://srv/share/tests/security" + executions = [_bash_execution(f"pytest {target} FileSystem:://srv/share/tests/unit --ignore FileSystem:://srv/SHARE/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_distinct_unc_root_exclusion_fails_closed(self): + target = "FileSystem:://srv/share/tests/security" + executions = [_bash_execution(f"pytest {target} FileSystem:://srv/share/tests/unit --ignore FileSystem:://srv/other/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_unrelated_unc_exclusion_keeps_matching(self): + target = "FileSystem:://srv/share/tests/security" + executions = [_bash_execution(f"pytest {target} FileSystem:://srv/share/tests/unit --ignore FileSystem:://srv/share/tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_provider_qualified_unc_preserves_pytest_nodeid_boundary(self): + target = "FileSystem:://srv/share/tests/x.py::TestA" + executions = [_bash_execution(f"pytest {target} FileSystem:://srv/share/tests/y.py --deselect FileSystem:://srv/SHARE/tests/x.py::TestA", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_provider_qualified_unc_keeps_pytest_nodeid_case_sensitive(self): + target = "FileSystem:://srv/share/tests/x.py::TestA" + executions = [_bash_execution(f"pytest {target} FileSystem:://srv/share/tests/y.py --deselect FileSystem:://srv/SHARE/tests/X.PY::testa", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_windows_drive_root_exclusion_overlap_is_unprovable(self): + executions = [_bash_execution("pytest D:/ D:/WS/tests/unit --ignore d:/ws/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest D:/"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize( + ("criterion_target", "ignored_target"), + ( + ("D:/WS/tests/./security", "d:/ws/tests/security"), + ("D:/WS/tests/security", "d:/ws/tests/./security"), + ("D:/WS/tests//security", "d:/ws/tests/security"), + ), + ) + def test_windows_lexical_alias_exclusion_is_unprovable(self, criterion_target, ignored_target): + executions = [_bash_execution(f"pytest {criterion_target} D:/WS/tests/unit --ignore {ignored_target}", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {criterion_target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_windows_relative_case_alias_exclusion_is_unprovable(self): + executions = [_bash_execution("pytest tests/security tests/unit --ignore TESTS/SECURITY", output_tail="12 passed")] + verdict = check_acceptance_criteria( + ["tests_passed:pytest tests/security"], + thread_data=WINDOWS_THREAD_DATA, + bash_executions=executions, + ) + + assert verdict["leaves"][0]["checked"] is False + + def test_windows_rooted_path_case_alias_exclusion_is_unprovable(self): + target = "/Tests/Security" + executions = [_bash_execution(f"pytest {target} /Tests/Unit --ignore /tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_windows_thread_rejects_shell_ambiguous_virtual_cd_target(self): + command = "cd /mnt/user-data/workspace && pytest tests/" + executions = [_bash_execution(command, output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:{command}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_unc_cd_target_uses_windows_case_rules(self): + thread_data = { + "workspace_path": "//SERVER/Share/Workspace", + "uploads_path": "//SERVER/Share/Workspace/uploads", + "outputs_path": "//SERVER/Share/Workspace/outputs", + } + command = "cd //server/share/workspace && pytest tests/" + executions = [_bash_execution(command, output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:{command}"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + @pytest.mark.parametrize("ignored", ("D:tests/security", "C:WS/tests/security")) + def test_drive_relative_exclusion_fails_closed(self, ignored): + target = "D:/WS/tests/security" + executions = [_bash_execution(f"pytest {target} D:/WS/tests/unit --ignore {ignored}", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_drive_relative_case_alias_uses_windows_semantics_without_thread_context(self): + target = "D:tests/security" + executions = [_bash_execution(f"pytest {target} D:tests/unit --ignore d:TESTS/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_different_drive_relative_exclusion_fails_closed_without_thread_context(self): + executions = [_bash_execution("pytest C:tests/security C:tests/unit --ignore D:tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest C:tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_unrelated_exclusion_on_same_drive_relative_root_keeps_matching(self): + target = "C:tests/security" + executions = [_bash_execution(f"pytest {target} C:tests/unit --ignore c:tests/slow", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_drive_relative_and_plain_relative_exclusion_fail_closed_without_thread_context(self): + executions = [_bash_execution("pytest tests/security tests/unit --ignore D:tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_drive_relative_and_plain_relative_exclusion_fail_closed_in_windows_context(self): + executions = [_bash_execution("pytest tests/security tests/unit --ignore D:tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria( + ["tests_passed:pytest tests/security"], + thread_data=WINDOWS_THREAD_DATA, + bash_executions=executions, + ) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize( + ("thread_data", "target", "ignored"), + ( + (WINDOWS_THREAD_DATA, "tests/security", "D:/WS/tests/security"), + (WINDOWS_THREAD_DATA, "/mnt/user-data/workspace/tests/security", "D:/WS/tests/security"), + (THREAD_DATA, "tests/security", "/ws/thread/user-data/workspace/tests/security"), + ), + ) + def test_mixed_path_forms_in_exclusion_fail_closed(self, thread_data, target, ignored): + executions = [_bash_execution(f"pytest {target} tests/unit --ignore {ignored}", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=thread_data, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize( + ("criterion_target", "additional_target", "ignored_target"), + ( + ("/tests/security", "/tests/unit", "D:/tests/security"), + ("D:/WS/tests/security", "D:/WS/tests/unit", "/WS/tests/security"), + ("//srv/share/tests/security", "//srv/share/tests/unit", "D:/tests/security"), + ), + ) + def test_cross_family_absolute_exclusion_fails_closed(self, criterion_target, additional_target, ignored_target): + executions = [_bash_execution(f"pytest {criterion_target} {additional_target} --ignore {ignored_target}", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {criterion_target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_unc_case_alias_exclusion_is_unprovable(self): + target = "//SERVER/Share/tests/security" + executions = [_bash_execution(f"pytest {target} //SERVER/Share/tests/unit --ignore //server/share/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + @pytest.mark.parametrize("criterion_target", ("tests/./security", "tests//security")) + def test_posix_lexical_alias_exclusion_is_unprovable(self, criterion_target): + executions = [_bash_execution(f"pytest {criterion_target} tests/unit --ignore tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {criterion_target}"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_windows_trailing_dot_alias_exclusion_is_unprovable(self): + target = "D:/WS/tests/security." + executions = [_bash_execution(f"pytest {target} D:/WS/tests/unit --ignore d:/ws/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_windows_short_name_alias_exclusion_is_unprovable_without_thread_context(self): + target = "C:/longdirectoryname/tests/security" + executions = [_bash_execution(f"pytest {target} C:/longdirectoryname/tests/unit --ignore C:/LONGDI~1/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_relative_windows_short_name_alias_exclusion_is_unprovable_in_windows_context(self): + target = "longdirectoryname/tests/security" + executions = [_bash_execution(f"pytest {target} longdirectoryname/tests/unit --ignore LONGDI~1/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_non_short_name_tilde_component_keeps_matching(self): + target = "C:/LONGDI~X/tests/security" + executions = [_bash_execution(f"pytest {target} C:/LONGDI~X/tests/unit --ignore C:/other/tests/security", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_parent_traversal_in_consumed_target_with_exclusion_is_unprovable(self): + executions = [_bash_execution("pytest tests/../security tests/unit --ignore security", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/../security"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["checked"] is False + + def test_posix_relative_case_aliases_remain_distinct(self): + executions = [_bash_execution("pytest tests/security tests/unit --ignore TESTS/SECURITY", output_tail="12 passed")] + verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], thread_data=THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + def test_windows_nodeid_case_remains_distinct(self): + target = "D:/WS/tests/x.py::TestA" + executions = [_bash_execution(f"pytest {target} D:/WS/tests/y.py --deselect d:/ws/tests/X.PY::testa", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {target}"], thread_data=WINDOWS_THREAD_DATA, bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + + @pytest.mark.parametrize( + ("criterion_target", "ignored_target"), + ( + ("/opt/WS/tests/security", "/opt/ws/tests/security"), + ("D:/WS/Straße/tests/security", "d:/ws/STRASSE/tests/security"), + ), + ) + def test_distinct_case_sensitive_exclusion_paths_do_not_overlap(self, criterion_target, ignored_target): + executions = [_bash_execution(f"pytest {criterion_target} tests/unit --ignore {ignored_target}", output_tail="12 passed")] + verdict = check_acceptance_criteria([f"tests_passed:pytest {criterion_target}"], bash_executions=executions) + + assert verdict["leaves"][0]["holds"] is True + def test_unrelated_exclusion_does_not_block_the_match(self): executions = [_bash_execution("pytest tests/security tests/unit --ignore tests/slow", output_tail="12 passed")] verdict = check_acceptance_criteria(["tests_passed:pytest tests/security"], bash_executions=executions)