diff --git a/CHANGELOG.md b/CHANGELOG.md index 734f227c..7ec70f69 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,12 @@ ## [Unreleased] +## [v0.51.506] — 2026-06-18 — Release RQ (harden destructive workspace Git paths against repo-local execution) + +### Security + +- **Opening or operating on an untrusted Git repository in the workspace can no longer execute arbitrary commands the repo planted in its own config or hooks (#3777).** Workspace Git operations now run with a hardened configuration that neutralizes repo-local execution vectors — `core.fsmonitor`, `core.sshCommand`, `core.gitProxy`, `credential.helper`, `core.askPass`, `protocol.ext.allow`, GPG/signing programs, `core.alternateRefsCommand`, submodule recursion, and `filter`/`merge`/remote-helper programs — and redirects Git hooks to an empty directory on every ref-mutating and destructive operation (including Fetch, which fires the `reference-transaction` hook). Repo-derived config overrides are delivered exclusively via the `GIT_CONFIG_COUNT`/`GIT_CONFIG_KEY_n`/`GIT_CONFIG_VALUE_n` environment form so an attacker-controlled filter/section name containing `=` cannot inject configuration. When a repository defines local content filters, the content-rewriting actions (stage, discard, pull, checkout/switch, stash restore, selected-commit staging) fail closed with a clear message rather than silently writing unfiltered bytes. Thanks @rodboev. + ## [v0.51.505] — 2026-06-18 — Release RP (allow macOS Mail `message:` links in chat markdown) ### Fixed diff --git a/api/workspace_git.py b/api/workspace_git.py index 3194eaa9..a33fd316 100644 --- a/api/workspace_git.py +++ b/api/workspace_git.py @@ -8,15 +8,20 @@ pathspecs, and keeps all Git subprocess calls shell-free and bounded. from __future__ import annotations import difflib +import logging import os +import shutil import subprocess import tempfile import threading +import re from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path from typing import Iterable +logger = logging.getLogger(__name__) + from api.workspace import rmtree_anchored, safe_resolve_ws, unlink_anchored @@ -55,13 +60,41 @@ _GIT_HARDENED_CONFIG = ( # Neutralize repo-local core.gitProxy, which specifies an external proxy # command reachable on `git fetch` against a git:// remote. ("core.gitProxy", ""), + # Prevent submodule operations from recursing into nested repos, which + # could trigger hooks or fetch from attacker-controlled submodule URLs. + ("submodule.recurse", "false"), + ("fetch.recurseSubmodules", "false"), +) +_GIT_DESTRUCTIVE_HARDENED_CONFIG = ( + # Disable signing helper command resolution while performing destructive + # Git operations. Hooks are redirected to a temporary empty directory in + # _run_git() so Git never falls back to .git/hooks. + ("commit.gpgSign", "false"), + ("push.gpgSign", "false"), + ("gpg.program", ""), + ("gpg.ssh.program", ""), + ("gpg.x509.program", ""), + ("core.alternateRefsCommand", ""), ) -def _hardened_git_argv(args: list[str]) -> list[str]: +def _hardened_git_argv( + args: list[str], + *, + destructive: bool = False, + attributes_file: str | None = None, + hooks_path: str | None = None, +) -> list[str]: argv = ["git"] for key, value in _GIT_HARDENED_CONFIG: argv.extend(["-c", f"{key}={value}"]) + if destructive: + for key, value in _GIT_DESTRUCTIVE_HARDENED_CONFIG: + argv.extend(["-c", f"{key}={value}"]) + if hooks_path: + argv.extend(["-c", f"core.hooksPath={hooks_path}"]) + if attributes_file: + argv.extend(["-c", f"core.attributesFile={attributes_file}"]) argv.extend(args) return argv @@ -164,12 +197,51 @@ def _run_git( timeout: int = GIT_TIMEOUT, check: bool = False, env: dict[str, str] | None = None, + destructive: bool = False, + force_destructive_hardening: bool = False, + disable_filter_attributes: bool = False, + neutralize_filter_programs: bool = False, + neutralize_remote_helpers: bool = False, ) -> subprocess.CompletedProcess[str]: cwd = ctx_or_cwd.repo_root if isinstance(ctx_or_cwd, GitContext) else ctx_or_cwd run_env = _clean_git_env(env) + effective_destructive = destructive and workspace_git_destructive_enabled() + hardened_destructive_path = effective_destructive or force_destructive_hardening + attributes_file = None + hooks_path = None + extra_configs: list[tuple[str, str]] = [] + temporary_attributes: list[str] = [] + temporary_dirs: list[str] = [] try: + if disable_filter_attributes: + fd, attributes_path = tempfile.mkstemp(prefix="hermes-webui-git-attrs-") + os.close(fd) + attributes_file = attributes_path + temporary_attributes = [attributes_path] + if disable_filter_attributes or neutralize_filter_programs: + # Read/status/fetch paths treat repo-local filter programs as + # untrusted code. Prefer raw-byte visibility over executing them. + extra_configs.extend(_destructive_filter_overrides(cwd, run_env)) + if effective_destructive: + extra_configs.extend(_destructive_merge_driver_overrides(cwd, run_env)) + if effective_destructive or neutralize_remote_helpers: + extra_configs.extend(_destructive_remote_helper_overrides(cwd, run_env)) + args = _destructive_remote_command_args(args, cwd, run_env) + if hardened_destructive_path: + hooks_path = tempfile.mkdtemp(prefix="hermes-webui-git-hooks-") + temporary_dirs = [hooks_path] + if extra_configs: + run_env["GIT_CONFIG_COUNT"] = str(len(extra_configs)) + for i, (key, value) in enumerate(extra_configs): + run_env[f"GIT_CONFIG_KEY_{i}"] = key + run_env[f"GIT_CONFIG_VALUE_{i}"] = value result = subprocess.run( - _hardened_git_argv(args), + _hardened_git_argv( + args, + destructive=hardened_destructive_path, + attributes_file=attributes_file, + hooks_path=hooks_path, + ), cwd=str(cwd), shell=False, capture_output=True, @@ -183,12 +255,202 @@ def _run_git( raise GitWorkspaceError("Git is not installed or not available on PATH", "missing_git") from exc except OSError as exc: raise GitWorkspaceError(str(exc), _classify_git_error(str(exc), args)) from exc + finally: + for path in temporary_attributes: + Path(path).unlink(missing_ok=True) + for path in temporary_dirs: + shutil.rmtree(path, ignore_errors=True) if check and result.returncode != 0: message = (result.stderr or result.stdout or "Git command failed").strip() raise GitWorkspaceError(message, _classify_git_error(message, args)) return result +_FILTER_CONFIG_RE = re.compile(r"^filter\.(.+)\.(clean|smudge|process|required)$") + + +_MERGE_DRIVER_CONFIG_RE = re.compile(r"^merge\.(.+)\.driver$") +_REMOTE_HELPER_CONFIG_RE = re.compile(r"^remote\.(.+)\.(uploadpack|receivepack)$") + + +def _config_names_for_scope( + scope: str, + cwd: Path, + env: dict[str, str], + config_pattern: str, + name_re: re.Pattern[str], + *, + ignore_unsupported: bool = False, +) -> set[str]: + result = subprocess.run( + ["git", "config", "--includes", scope, "--name-only", "--get-regexp", config_pattern], + cwd=str(cwd), + shell=False, + text=True, + capture_output=True, + timeout=GIT_TIMEOUT, + env=env, + ) + if result.returncode not in {0, 1}: + if ignore_unsupported: + return set() + message = (result.stderr or result.stdout or "Git command failed").strip() + raise GitWorkspaceError(message, _classify_git_error(message, ["config"])) + names: set[str] = set() + for line in (result.stdout or "").splitlines(): + match = name_re.match(line.strip()) + if match: + names.add(match.group(1)) + return names + + +def _filter_names_for_scope( + scope: str, + cwd: Path, + env: dict[str, str], + *, + ignore_unsupported: bool = False, +) -> set[str]: + return _config_names_for_scope( + scope, + cwd, + env, + r"^filter\..*\.(clean|smudge|process|required)$", + _FILTER_CONFIG_RE, + ignore_unsupported=ignore_unsupported, + ) + + +def _merge_driver_names_for_scope( + scope: str, + cwd: Path, + env: dict[str, str], + *, + ignore_unsupported: bool = False, +) -> set[str]: + return _config_names_for_scope( + scope, + cwd, + env, + r"^merge\..*\.driver$", + _MERGE_DRIVER_CONFIG_RE, + ignore_unsupported=ignore_unsupported, + ) + + +def _remote_helper_names_for_scope( + scope: str, + cwd: Path, + env: dict[str, str], + *, + ignore_unsupported: bool = False, +) -> set[str]: + return _config_names_for_scope( + scope, + cwd, + env, + r"^remote\..*\.(uploadpack|receivepack)$", + _REMOTE_HELPER_CONFIG_RE, + ignore_unsupported=ignore_unsupported, + ) + + +def _destructive_filter_overrides(cwd: Path, env: dict[str, str]) -> list[tuple[str, str]]: + names = _filter_names_for_scope("--local", cwd, env) + names |= _filter_names_for_scope( + "--worktree", + cwd, + env, + ignore_unsupported=True, + ) + overrides: list[tuple[str, str]] = [] + for name in sorted(names): + if "\n" in name or "\0" in name: + logger.warning("Skipping filter name with illegal characters: %r", name) + continue + overrides.extend( + [ + (f"filter.{name}.clean", "cat"), + (f"filter.{name}.smudge", "cat"), + (f"filter.{name}.process", ""), + (f"filter.{name}.required", "false"), + ] + ) + return overrides + + +def _destructive_merge_driver_overrides(cwd: Path, env: dict[str, str]) -> list[tuple[str, str]]: + names = _merge_driver_names_for_scope("--local", cwd, env) + names |= _merge_driver_names_for_scope( + "--worktree", + cwd, + env, + ignore_unsupported=True, + ) + # Replace repo-defined merge drivers with Git's trusted three-way merge + # binary so stash restores cannot invoke workspace-controlled helpers. + overrides: list[tuple[str, str]] = [] + for name in sorted(names): + if "\n" in name or "\0" in name: + logger.warning("Skipping merge driver name with illegal characters: %r", name) + continue + overrides.append((f"merge.{name}.driver", 'git merge-file "%A" "%O" "%B"')) + return overrides + + +def _destructive_remote_helper_overrides(cwd: Path, env: dict[str, str]) -> list[tuple[str, str]]: + names = _remote_helper_names_for_scope("--local", cwd, env) + names |= _remote_helper_names_for_scope( + "--worktree", + cwd, + env, + ignore_unsupported=True, + ) + overrides: list[tuple[str, str]] = [] + for name in sorted(names): + if "\n" in name or "\0" in name: + logger.warning("Skipping remote helper name with illegal characters: %r", name) + continue + overrides.extend( + [ + (f"remote.{name}.uploadpack", "git-upload-pack"), + (f"remote.{name}.receivepack", "git-receive-pack"), + ] + ) + return overrides + + +def _destructive_remote_command_args(args: list[str], cwd: Path, env: dict[str, str]) -> list[str]: + if not args: + return args + names = _remote_helper_names_for_scope("--local", cwd, env) + names |= _remote_helper_names_for_scope( + "--worktree", + cwd, + env, + ignore_unsupported=True, + ) + if not names: + return args + command = args[0] + if command in {"fetch", "pull"}: + return [command, "--upload-pack=git-upload-pack", *args[1:]] + if command == "push": + return [command, "--receive-pack=git-receive-pack", *args[1:]] + return args + + +def _has_repo_local_filters(cwd: Path, env: dict[str, str]) -> bool: + names = _filter_names_for_scope("--local", cwd, env) + names |= _filter_names_for_scope("--worktree", cwd, env, ignore_unsupported=True) + return bool(names) + + +def _block_filtered_destructive_write(ctx: GitContext, message: str) -> None: + if workspace_git_destructive_enabled() and _has_repo_local_filters(ctx.repo_root, _clean_git_env()): + raise GitWorkspaceError(message, "filtered_path") + + def resolve_git_context(workspace: str | Path) -> GitContext | None: ws = Path(workspace).expanduser().resolve() result = _run_git(ws, ["rev-parse", "--show-toplevel"], check=False) @@ -288,12 +550,19 @@ def _parse_path_list(text: str, ctx: GitContext) -> set[str]: def _collect_diff_paths(ctx: GitContext, cached: bool, *, ignore_cr_at_eol: bool = True) -> set[str] | None: args = ["diff", "--name-only", "-z"] + args.append("--no-textconv") if ignore_cr_at_eol: args.append("--ignore-cr-at-eol") if cached: args.append("--cached") args.extend(["--", _workspace_pathspec(ctx)]) - result = _run_git(ctx, args, check=False) + result = _run_git( + ctx, + args, + check=False, + disable_filter_attributes=workspace_git_destructive_enabled(), + neutralize_filter_programs=True, + ) if result.returncode != 0: return None return _parse_path_list(result.stdout, ctx) @@ -306,12 +575,19 @@ def _collect_numstat( ignore_cr_at_eol: bool = True, ) -> dict[str, tuple[int, int, bool]]: args = ["diff", "--numstat"] + args.append("--no-textconv") if ignore_cr_at_eol: args.append("--ignore-cr-at-eol") if cached: args.append("--cached") args.extend(["--", _workspace_pathspec(ctx)]) - result = _run_git(ctx, args, check=False) + result = _run_git( + ctx, + args, + check=False, + disable_filter_attributes=workspace_git_destructive_enabled(), + neutralize_filter_programs=True, + ) if result.returncode != 0: return {} return _parse_numstat(result.stdout, ctx) @@ -354,6 +630,8 @@ def git_status(workspace: str | Path) -> dict: _workspace_pathspec(ctx), ], check=True, + disable_filter_attributes=workspace_git_destructive_enabled(), + neutralize_filter_programs=True, ) staged_stats = _collect_numstat(ctx, cached=True) unstaged_stats = _collect_numstat(ctx, cached=False) @@ -661,7 +939,8 @@ def _validate_new_branch_name(ctx: GitContext, name: str) -> str: def _dirty_worktree(ctx: GitContext) -> bool: - result = _run_git(ctx, ["status", "--porcelain=v2", "--untracked-files=all"], check=True) + result = _run_git(ctx, ["status", "--porcelain=v2", "--untracked-files=all"], + check=True, neutralize_filter_programs=True) return bool(result.stdout.strip()) @@ -703,12 +982,23 @@ def _hermes_branch_switch_stashes(ctx: GitContext) -> list[dict]: def _restore_branch_switch_stash_locked(ctx: GitContext, branch: str) -> dict: + if workspace_git_destructive_enabled() and _has_repo_local_filters(ctx.repo_root, _clean_git_env()): + return { + "restore_blocked": True, + "restore_reason": "Repository defines local filter programs", + } if _dirty_worktree(ctx): return {} for item in _hermes_branch_switch_stashes(ctx): if item.get("branch") != branch: continue - result = _run_git(ctx, ["stash", "pop", "--index", item["ref"]], check=False) + result = _run_git( + ctx, + ["stash", "pop", "--index", item["ref"]], + check=False, + destructive=True, + disable_filter_attributes=True, + ) if result.returncode == 0: return {"restored_stash": item} return { @@ -753,31 +1043,66 @@ def _perform_checkout_locked( new_branch: str | None, track: bool, ) -> subprocess.CompletedProcess[str]: + if workspace_git_destructive_enabled() and _has_repo_local_filters(ctx.repo_root, _clean_git_env()): + raise GitWorkspaceError( + "Cannot checkout: repository defines local filter programs that would alter file content", + "filtered_path", + ) if mode == "local": target = _validate_local_branch(ctx, ref) - return _run_git(ctx, ["switch", target], check=True) + return _run_git( + ctx, + ["switch", "--recurse-submodules=no", target], + check=True, + destructive=True, + disable_filter_attributes=True, + ) if mode in {"new", "create"}: branch = _validate_new_branch_name(ctx, new_branch or ref) start_ref = _validate_checkout_start(ctx, ref if (new_branch and ref and ref != new_branch) else "HEAD") - return _run_git(ctx, ["switch", "-c", branch, start_ref], check=True) + return _run_git( + ctx, + ["switch", "--recurse-submodules=no", "-c", branch, start_ref], + check=True, + destructive=True, + disable_filter_attributes=True, + ) if mode == "remote": remote_ref = _validate_remote_branch(ctx, ref) branch_name = str(new_branch or remote_ref.split("/", 1)[-1]).strip() exists = _run_git(ctx, ["show-ref", "--verify", f"refs/heads/{branch_name}"], check=False) if exists.returncode == 0: - result = _run_git(ctx, ["switch", branch_name], check=True) + result = _run_git( + ctx, + ["switch", "--recurse-submodules=no", branch_name], + check=True, + destructive=True, + disable_filter_attributes=True, + ) if track: _run_git(ctx, ["branch", "--set-upstream-to", remote_ref, branch_name], check=False) return result branch = _validate_new_branch_name(ctx, branch_name) - args = ["switch", "-c", branch] + args = ["switch", "--recurse-submodules=no", "-c", branch] if track: args.append("--track") args.append(remote_ref) - return _run_git(ctx, args, check=True) + return _run_git( + ctx, + args, + check=True, + destructive=True, + disable_filter_attributes=True, + ) if mode in {"detached", "detach"}: target = _validate_checkout_start(ctx, ref) - return _run_git(ctx, ["switch", "--detach", target], check=True) + return _run_git( + ctx, + ["switch", "--recurse-submodules=no", "--detach", target], + check=True, + destructive=True, + disable_filter_attributes=True, + ) raise GitWorkspaceError("Unsupported checkout mode", "invalid_ref") @@ -831,16 +1156,33 @@ def git_stash_and_checkout( restored: dict = {} with _git_mutation_lock(ctx): _validate_checkout_request_locked(ctx, ref, mode, new_branch) + if workspace_git_destructive_enabled() and _has_repo_local_filters(ctx.repo_root, _clean_git_env()): + raise GitWorkspaceError( + "Cannot stash: repository defines local filter programs that would alter file content", + "filtered_path", + ) stashed = False if _dirty_worktree(ctx): - stash_result = _run_git(ctx, ["stash", "push", "-u", "-m", stash_name], check=True) + stash_result = _run_git( + ctx, + ["stash", "push", "-u", "-m", stash_name], + check=True, + destructive=True, + disable_filter_attributes=True, + ) stash_text = _remote_message(stash_result) stashed = "No local changes to save" not in stash_text try: result = _perform_checkout_locked(ctx, workspace, ref, mode, new_branch, track) except Exception: if stashed: - _run_git(ctx, ["stash", "pop", "--index", "stash@{0}"], check=False) + _run_git( + ctx, + ["stash", "pop", "--index", "stash@{0}"], + check=False, + destructive=True, + disable_filter_attributes=True, + ) raise current_branch = _current_checkout_label(ctx) restored = _restore_branch_switch_stash_locked(ctx, current_branch) @@ -930,11 +1272,11 @@ def git_diff(workspace: str | Path, path: str, kind: str = "unstaged") -> dict: payload = _synthetic_untracked_diff(ctx.workspace / workspace_rel, workspace_rel) return {"path": workspace_rel, "kind": kind, **payload} - args = ["diff", "--no-ext-diff", "--unified=3"] + args = ["diff", "--no-ext-diff", "--no-textconv", "--unified=3"] if kind == "staged": args.append("--cached") args.extend(["--", repo_rel]) - result = _run_git(ctx, args, check=True) + result = _run_git(ctx, args, check=True, neutralize_filter_programs=True) diff = result.stdout binary = "Binary files " in diff or "GIT binary patch" in diff too_large = len(diff.encode("utf-8", errors="replace")) > DIFF_SIZE_LIMIT @@ -972,7 +1314,17 @@ def git_stage(workspace: str | Path, paths: Iterable[str]) -> dict: if ctx is None: raise GitWorkspaceError("Workspace is not a Git repository", "not_a_repo") with _git_mutation_lock(ctx): - _run_git(ctx, ["add", "--", *_pathspecs(ctx, paths)], check=True) + _block_filtered_destructive_write( + ctx, + "Repository uses local Git filters; stage may corrupt index content. Use the terminal to stage manually.", + ) + _run_git( + ctx, + ["add", "--", *_pathspecs(ctx, paths)], + check=True, + destructive=True, + disable_filter_attributes=True, + ) return git_status(workspace) @@ -982,9 +1334,14 @@ def git_unstage(workspace: str | Path, paths: Iterable[str]) -> dict: raise GitWorkspaceError("Workspace is not a Git repository", "not_a_repo") specs = _pathspecs(ctx, paths) with _git_mutation_lock(ctx): - result = _run_git(ctx, ["restore", "--staged", "--", *specs], check=False) + result = _run_git( + ctx, + ["restore", "--staged", "--", *specs], + check=False, + destructive=True, + ) if result.returncode != 0: - _run_git(ctx, ["reset", "HEAD", "--", *specs], check=True) + _run_git(ctx, ["reset", "HEAD", "--", *specs], check=True, destructive=True) return git_status(workspace) @@ -993,6 +1350,11 @@ def git_discard(workspace: str | Path, paths: Iterable[str], *, delete_untracked if ctx is None: raise GitWorkspaceError("Workspace is not a Git repository", "not_a_repo") with _git_mutation_lock(ctx): + _block_filtered_destructive_write( + ctx, + "Repository uses local Git filters; discard may corrupt working-tree content. " + "Use the terminal to discard manually.", + ) status = git_status(workspace) by_path = {f["path"]: f for f in status.get("files", [])} for path in _clean_paths(paths): @@ -1017,7 +1379,13 @@ def git_discard(workspace: str | Path, paths: Iterable[str], *, delete_untracked # reported it but before this discard reaches unlink. pass continue - _run_git(ctx, ["restore", "--worktree", "--", repo_rel], check=True) + _run_git( + ctx, + ["restore", "--worktree", "--", repo_rel], + check=True, + destructive=True, + disable_filter_attributes=True, + ) return git_status(workspace) @@ -1042,11 +1410,13 @@ def _staged_diff_text(ctx: GitContext) -> tuple[str, bool]: "diff", "--cached", "--no-ext-diff", + "--no-textconv", "--unified=3", "--", _workspace_pathspec(ctx), ], check=True, + neutralize_filter_programs=True, ) diff = result.stdout or "" encoded = diff.encode("utf-8", errors="replace") @@ -1056,17 +1426,35 @@ def _staged_diff_text(ctx: GitContext) -> tuple[str, bool]: def _selected_temp_index_env(ctx: GitContext, specs: list[str]) -> tuple[dict[str, str], str]: + _block_filtered_destructive_write( + ctx, + "Repository uses local Git filters; selected commit staging may corrupt index content. " + "Use the terminal to commit manually.", + ) fd, index_path = tempfile.mkstemp(prefix="hermes-webui-git-index-") os.close(fd) Path(index_path).unlink(missing_ok=True) env = {"GIT_INDEX_FILE": index_path} try: - head = _run_git(ctx, ["rev-parse", "--verify", "HEAD"], check=False, env=env) + head = _run_git( + ctx, + ["rev-parse", "--verify", "HEAD"], + check=False, + env=env, + destructive=True, + ) if head.returncode == 0: - _run_git(ctx, ["read-tree", "HEAD"], check=True, env=env) + _run_git(ctx, ["read-tree", "HEAD"], check=True, env=env, destructive=True) else: - _run_git(ctx, ["read-tree", "--empty"], check=True, env=env) - _run_git(ctx, ["add", "-A", "--", *specs], check=True, env=env) + _run_git(ctx, ["read-tree", "--empty"], check=True, env=env, destructive=True) + _run_git( + ctx, + ["add", "-A", "--", *specs], + check=True, + env=env, + destructive=True, + disable_filter_attributes=True, + ) return env, index_path except Exception: Path(index_path).unlink(missing_ok=True) @@ -1102,9 +1490,11 @@ def _selected_diff_text(ctx: GitContext, specs: list[str]) -> tuple[str, bool]: try: result = _run_git( ctx, - ["diff", "--cached", "--no-ext-diff", "--unified=3", "--", *specs], + ["diff", "--cached", "--no-ext-diff", "--no-textconv", "--unified=3", "--", *specs], check=True, env=env, + destructive=True, + disable_filter_attributes=True, ) diff = result.stdout or "" encoded = diff.encode("utf-8", errors="replace") @@ -1223,7 +1613,14 @@ def git_commit(workspace: str | Path, message: str) -> dict: if ctx is None: raise GitWorkspaceError("Workspace is not a Git repository", "not_a_repo") with _git_mutation_lock(ctx): - _run_git(ctx, ["commit", "-m", msg], timeout=10, check=True) + _run_git( + ctx, + ["commit", "-m", msg], + timeout=10, + check=True, + destructive=True, + disable_filter_attributes=True, + ) sha = _run_git(ctx, ["rev-parse", "--short", "HEAD"], check=True).stdout.strip() return {"ok": True, "commit": sha, "status": git_status(workspace)} @@ -1239,11 +1636,31 @@ def git_commit_selected(workspace: str | Path, message: str, paths: Iterable[str specs, workspace_paths, _selected_files_list = _selected_files(ctx, paths) env, index_path = _selected_temp_index_env(ctx, specs) try: - quiet = _run_git(ctx, ["diff", "--cached", "--quiet", "--", *specs], check=False, env=env) + quiet = _run_git( + ctx, + ["diff", "--cached", "--quiet", "--no-textconv", "--", *specs], + check=False, + env=env, + destructive=True, + disable_filter_attributes=True, + ) if quiet.returncode == 0: raise GitWorkspaceError("Selected paths have no committable changes") - _run_git(ctx, ["commit", "-m", msg], timeout=10, check=True, env=env) - _run_git(ctx, ["reset", "-q", "HEAD", "--", *specs], check=True) + _run_git( + ctx, + ["commit", "-m", msg], + timeout=10, + check=True, + env=env, + destructive=True, + disable_filter_attributes=True, + ) + _run_git( + ctx, + ["reset", "-q", "HEAD", "--", *specs], + check=True, + destructive=True, + ) finally: Path(index_path).unlink(missing_ok=True) sha = _run_git(ctx, ["rev-parse", "--short", "HEAD"], check=True).stdout.strip() @@ -1266,7 +1683,16 @@ def git_fetch(workspace: str | Path) -> dict: if ctx is None: raise GitWorkspaceError("Workspace is not a Git repository", "not_a_repo") with _git_mutation_lock(ctx): - result = _run_git(ctx, ["fetch", "--prune"], timeout=GIT_REMOTE_TIMEOUT, check=True) + result = _run_git( + ctx, + ["fetch", "--prune", "--no-recurse-submodules"], + timeout=GIT_REMOTE_TIMEOUT, + check=True, + force_destructive_hardening=True, + disable_filter_attributes=workspace_git_destructive_enabled(), + neutralize_filter_programs=True, + neutralize_remote_helpers=True, + ) return {"ok": True, "message": _remote_message(result), "status": git_status(workspace)} @@ -1275,7 +1701,20 @@ def git_pull(workspace: str | Path) -> dict: if ctx is None: raise GitWorkspaceError("Workspace is not a Git repository", "not_a_repo") with _git_mutation_lock(ctx): - result = _run_git(ctx, ["pull", "--ff-only"], timeout=GIT_REMOTE_TIMEOUT, check=True) + _block_filtered_destructive_write( + ctx, + "Repository uses local Git filters; pull may corrupt working-tree content. Use the terminal to pull manually.", + ) + result = _run_git( + ctx, + ["pull", "--ff-only", "--no-recurse-submodules"], + timeout=GIT_REMOTE_TIMEOUT, + check=True, + destructive=True, + disable_filter_attributes=True, + neutralize_filter_programs=True, + neutralize_remote_helpers=True, + ) return {"ok": True, "message": _remote_message(result), "status": git_status(workspace)} @@ -1292,5 +1731,14 @@ def git_push(workspace: str | Path) -> dict: if "origin" not in remotes: raise GitWorkspaceError("No upstream branch or origin remote is configured", "no_upstream") args.extend(["-u", "origin", branch]) - result = _run_git(ctx, args, timeout=GIT_REMOTE_TIMEOUT, check=True) + result = _run_git( + ctx, + args, + timeout=GIT_REMOTE_TIMEOUT, + check=True, + destructive=True, + disable_filter_attributes=True, + neutralize_filter_programs=True, + neutralize_remote_helpers=True, + ) return {"ok": True, "message": _remote_message(result), "status": git_status(workspace)} diff --git a/tests/test_issue2540_models_endpoint_error.py b/tests/test_issue2540_models_endpoint_error.py index d98f7493..ff93501b 100644 --- a/tests/test_issue2540_models_endpoint_error.py +++ b/tests/test_issue2540_models_endpoint_error.py @@ -12,10 +12,16 @@ class _ConfigState: self.old_cache = config._available_models_cache self.old_cache_ts = config._available_models_cache_ts self.old_cache_fp = config._available_models_cache_source_fingerprint + self.old_live_budget = config._LIVE_REBUILD_BUDGET_SECONDS config._cfg_mtime = 0.0 config._available_models_cache = None config._available_models_cache_ts = 0.0 config._available_models_cache_source_fingerprint = None + # This file pins custom-endpoint error shaping, not the separate async + # provider-catalog budget fallback. Force the synchronous rebuild path + # so slower Python 3.11 shards cannot fall onto the global 4s fallback + # and mask the endpoint-specific error contract under test. + config._LIVE_REBUILD_BUDGET_SECONDS = 0.0 return self def __exit__(self, exc_type, exc, tb): @@ -24,6 +30,7 @@ class _ConfigState: config._available_models_cache = self.old_cache config._available_models_cache_ts = self.old_cache_ts config._available_models_cache_source_fingerprint = self.old_cache_fp + config._LIVE_REBUILD_BUDGET_SECONDS = self.old_live_budget return False diff --git a/tests/test_workspace_git.py b/tests/test_workspace_git.py index 2e68a819..f9e28ac3 100644 --- a/tests/test_workspace_git.py +++ b/tests/test_workspace_git.py @@ -275,6 +275,73 @@ def test_git_diff_generates_untracked_text_diff_and_blocks_escape(tmp_path): git_diff(repo, "../outside.txt", "unstaged") +def test_git_diff_skips_repo_local_textconv(tmp_path): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted textconv helper setup is POSIX-only") + + from api.workspace_git import git_diff + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt diff=demo\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "git-diff-textconv-ran" + helper = tmp_path / "git_diff_textconv_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('textconv ran', encoding='utf-8')\n" + "print('converted')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "diff.demo.textconv", f'"{sys.executable}" "{helper}" "{marker}"') + + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + diff = git_diff(repo, "tracked.txt", "unstaged") + + assert "+two" in diff["diff"] + assert "converted" not in diff["diff"] + assert not marker.exists() + + +def test_git_diff_skips_repo_local_clean_filter_without_destructive_mode(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted clean filter setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_diff + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "git-diff-clean-filter-ran" + helper = tmp_path / "git_diff_clean_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('clean filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f'"{sys.executable}" "{helper}" "{marker}"') + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + diff = git_diff(repo, "tracked.txt", "unstaged") + + assert "+two" in diff["diff"] + assert not marker.exists() + + def test_git_status_reports_untracked_files_inside_directories(tmp_path): from api.workspace_git import git_discard, git_status @@ -553,6 +620,49 @@ def test_staged_commit_message_prompt_uses_only_staged_diff(tmp_path): assert clean_generated_commit_message("```text\nSubject\n\n- Body\n```") == "Subject\n\n- Body" +def test_commit_message_prompts_skip_repo_local_textconv_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted textconv helper setup is POSIX-only") + + from api.workspace_git import ( + WORKSPACE_GIT_DESTRUCTIVE_ENV, + selected_commit_message_prompt, + staged_commit_message_prompt, + ) + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt diff=demo\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "textconv-ran" + helper = tmp_path / "textconv_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('textconv ran', encoding='utf-8')\n" + "print('converted')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "diff.demo.textconv", f'"{sys.executable}" "{helper}" "{marker}"') + + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + _git(repo, "add", "tracked.txt") + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + + staged = staged_commit_message_prompt(repo) + selected = selected_commit_message_prompt(repo, ["tracked.txt"]) + + assert "+two" in staged["user_prompt"] + assert "+two" in selected["user_prompt"] + assert "converted" not in staged["user_prompt"] + assert "converted" not in selected["user_prompt"] + assert not marker.exists() + + def test_git_fetch_pull_and_push_with_upstream(tmp_path): from api.workspace_git import git_fetch, git_pull, git_push, git_status @@ -590,6 +700,114 @@ def test_git_fetch_pull_and_push_with_upstream(tmp_path): assert pushed["status"]["ahead"] == 0 +def test_git_fetch_pull_and_push_skip_repo_local_remote_helpers_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted remote helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_fetch, git_pull, git_push + + remote = _init_bare_repo(tmp_path / "remote.git") + + origin = _init_repo(tmp_path / "origin") + (origin / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(origin) + _git(origin, "branch", "-M", "main") + _git(origin, "remote", "add", "origin", str(remote)) + _git(origin, "push", "-u", "origin", "main") + _git(remote, "symbolic-ref", "HEAD", "refs/heads/main") + + clone = tmp_path / "clone" + _git(tmp_path, "clone", str(remote), str(clone)) + _git(clone, "config", "user.email", "hermes-tests@example.invalid") + _git(clone, "config", "user.name", "Hermes Tests") + + marker = tmp_path / "remote-helper-ran" + helper = tmp_path / "remote_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('remote helper ran', encoding='utf-8')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + helper.chmod(0o755) + helper_cmd = f'"{sys.executable}" "{helper}" "{marker}"' + _git(clone, "config", "remote.origin.uploadpack", helper_cmd) + _git(clone, "config", "remote.origin.receivepack", helper_cmd) + + (origin / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + _commit_all(origin, "Remote update") + _git(origin, "push") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + fetched = git_fetch(clone) + pulled = git_pull(clone) + + assert fetched["status"]["behind"] == 1 + assert pulled["status"]["behind"] == 0 + assert not marker.exists() + + (clone / "tracked.txt").write_text("one\ntwo\nthree\n", encoding="utf-8") + _git(clone, "add", "tracked.txt") + _git(clone, "commit", "-m", "Local update") + + pushed = git_push(clone) + + assert pushed["status"]["ahead"] == 0 + assert not marker.exists() + + +def test_git_fetch_skips_repo_local_remote_helpers_without_destructive_mode(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted remote helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_fetch + + remote = _init_bare_repo(tmp_path / "remote.git") + + origin = _init_repo(tmp_path / "origin") + (origin / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(origin) + _git(origin, "branch", "-M", "main") + _git(origin, "remote", "add", "origin", str(remote)) + _git(origin, "push", "-u", "origin", "main") + _git(remote, "symbolic-ref", "HEAD", "refs/heads/main") + + clone = tmp_path / "clone" + _git(tmp_path, "clone", str(remote), str(clone)) + _git(clone, "config", "user.email", "hermes-tests@example.invalid") + _git(clone, "config", "user.name", "Hermes Tests") + + marker = tmp_path / "remote-helper-default-fetch-ran" + helper = tmp_path / "remote_helper_default_fetch.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('remote helper ran', encoding='utf-8')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + helper.chmod(0o755) + helper_cmd = f'"{sys.executable}" "{helper}" "{marker}"' + _git(clone, "config", "remote.origin.uploadpack", helper_cmd) + + (origin / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + _commit_all(origin, "Remote update") + _git(origin, "push") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + fetched = git_fetch(clone) + + assert fetched["status"]["behind"] == 1 + assert not marker.exists() + + def test_git_branches_lists_local_remote_and_upstream(tmp_path): from api.workspace_git import git_branches @@ -930,6 +1148,40 @@ def test_git_status_ignores_repo_local_fsmonitor_command(tmp_path): assert not marker.exists() +def test_git_status_skips_repo_local_clean_filter_without_destructive_mode(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted clean filter setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_status + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "git-status-clean-filter-ran" + helper = tmp_path / "git_status_clean_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('clean filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f'"{sys.executable}" "{helper}" "{marker}"') + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + status = git_status(repo) + + assert status["is_git"] is True + assert status["totals"]["unstaged"] == 1 + assert not marker.exists() + + def test_git_fetch_blocks_repo_local_ext_transport_execution(tmp_path): import os import sys @@ -1063,6 +1315,907 @@ def test_git_fetch_blocks_repo_local_askpass_execution(tmp_path): assert not marker.exists() +def test_git_commit_skips_repo_local_hooks_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("hook script setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_commit, git_stage + + repo = _init_repo(tmp_path / "repo") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + hooks = tmp_path / "hooks" + hooks.mkdir() + marker = tmp_path / "pre-commit-ran" + helper = tmp_path / "pre_commit_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('pre-commit executed', encoding='utf-8')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + helper.chmod(0o755) + pre_commit = hooks / "pre-commit" + pre_commit.write_text( + "#!/bin/sh\n" + f"\"{sys.executable}\" \"{helper}\" \"{marker}\"\n", + encoding="utf-8", + ) + pre_commit.chmod(0o755) + _git(repo, "config", "core.hooksPath", str(hooks)) + + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + git_stage(repo, ["tracked.txt"]) + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + git_commit(repo, "Commit with hooks disabled") + + assert not marker.exists() + + +def test_git_commit_skips_default_repo_hooks_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("hook script setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_commit, git_stage + + repo = _init_repo(tmp_path / "repo") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "default-pre-commit-ran" + helper = tmp_path / "default_pre_commit_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('default pre-commit executed', encoding='utf-8')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + helper.chmod(0o755) + pre_commit = repo / ".git" / "hooks" / "pre-commit" + pre_commit.write_text( + "#!/bin/sh\n" + f"\"{sys.executable}\" \"{helper}\" \"{marker}\"\n", + encoding="utf-8", + ) + pre_commit.chmod(0o755) + + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + git_stage(repo, ["tracked.txt"]) + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + git_commit(repo, "Commit with default hooks disabled") + + assert not marker.exists() + + +def test_git_checkout_skips_repo_local_hooks_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("hook script setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_checkout + + repo = _init_repo(tmp_path / "repo") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + _git(repo, "branch", "-M", "main") + _git(repo, "checkout", "-b", "feature") + _git(repo, "checkout", "main") + hooks = tmp_path / "hooks" + hooks.mkdir() + marker = tmp_path / "post-checkout-ran" + helper = tmp_path / "post_checkout_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('post-checkout executed', encoding='utf-8')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + post_checkout = hooks / "post-checkout" + post_checkout.write_text( + "#!/bin/sh\n" + f"\"{sys.executable}\" \"{helper}\" \"{marker}\"\n", + encoding="utf-8", + ) + post_checkout.chmod(0o755) + _git(repo, "config", "core.hooksPath", str(hooks)) + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + result = git_checkout(repo, "feature", "local") + + assert result["ok"] is True + assert result["current_branch"] == "feature" + assert not marker.exists() + + +def test_git_pull_skips_repo_local_hooks_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("hook script setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_pull + + remote = _init_bare_repo(tmp_path / "remote.git") + origin = _init_repo(tmp_path / "origin") + (origin / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(origin) + _git(origin, "branch", "-M", "main") + _git(origin, "remote", "add", "origin", str(remote)) + _git(origin, "push", "-u", "origin", "main") + _git(remote, "symbolic-ref", "HEAD", "refs/heads/main") + + clone = tmp_path / "clone" + _git(tmp_path, "clone", str(remote), str(clone)) + _git(clone, "config", "user.email", "hermes-tests@example.invalid") + _git(clone, "config", "user.name", "Hermes Tests") + + (origin / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + _commit_all(origin, "Remote update") + _git(origin, "push") + + hooks = tmp_path / "hooks" + hooks.mkdir() + marker = tmp_path / "post-merge-ran" + helper = tmp_path / "post_merge_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('post-merge executed', encoding='utf-8')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + post_merge = hooks / "post-merge" + post_merge.write_text( + "#!/bin/sh\n" + f"\"{sys.executable}\" \"{helper}\" \"{marker}\"\n", + encoding="utf-8", + ) + post_merge.chmod(0o755) + _git(clone, "config", "core.hooksPath", str(hooks)) + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + git_pull(clone) + + assert not marker.exists() + + +def test_git_stage_skips_repo_local_filters_when_destructive_mode_disabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_stage + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "filter-ran" + helper = tmp_path / "filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "config", "filter.demo.smudge", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker.unlink(missing_ok=True) + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + staged = git_stage(repo, ["tracked.txt"]) + + assert staged["totals"]["staged"] == 1 + assert not marker.exists() + + +def test_git_checkout_blocks_repo_local_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import ( + WORKSPACE_GIT_DESTRUCTIVE_ENV, + GitWorkspaceError, + git_checkout, + ) + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "filter-smudge-ran" + helper = tmp_path / "filter_checkout_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "config", "filter.demo.smudge", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "branch", "-M", "main") + _git(repo, "add", ".gitattributes") + _git(repo, "commit", "-m", "Initial") + _git(repo, "branch", "feature") + _git(repo, "checkout", "feature") + (repo / "tracked.txt").write_text("from feature\n", encoding="utf-8") + _git(repo, "add", "tracked.txt") + _git(repo, "commit", "-m", "Feature update") + _git(repo, "checkout", "main") + marker.unlink(missing_ok=True) + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_checkout(repo, "feature", "local") + assert exc.value.code == "filtered_path" + assert not marker.exists() + + +def test_git_stage_blocks_repo_local_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, GitWorkspaceError, git_stage + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "filter-stage-ran" + helper = tmp_path / "filter_stage_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "config", "filter.demo.smudge", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker.unlink(missing_ok=True) + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_stage(repo, ["tracked.txt"]) + + assert exc.value.code == "filtered_path" + assert not marker.exists() + + +def test_git_discard_blocks_repo_local_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, GitWorkspaceError, git_discard + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "filter-discard-ran" + helper = tmp_path / "filter_discard_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "config", "filter.demo.smudge", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker.unlink(missing_ok=True) + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_discard(repo, ["tracked.txt"]) + assert exc.value.code == "filtered_path" + assert (repo / "tracked.txt").read_text(encoding="utf-8") == "one\ntwo\n" + assert not marker.exists() + + +def test_destructive_filter_overrides_include_worktree_scope(tmp_path): + import os + + from api.workspace_git import _destructive_filter_overrides + + repo = _init_repo(tmp_path / "repo") + _git(repo, "config", "extensions.worktreeConfig", "true") + _git(repo, "config", "--worktree", "filter.demo.clean", "cat") + _git(repo, "config", "--worktree", "filter.demo.required", "true") + + overrides = dict(_destructive_filter_overrides(repo, os.environ.copy())) + + assert overrides["filter.demo.clean"] == "cat" + assert overrides["filter.demo.smudge"] == "cat" + assert overrides["filter.demo.process"] == "" + assert overrides["filter.demo.required"] == "false" + + +def test_destructive_filter_overrides_include_included_scope(tmp_path): + import os + + from api.workspace_git import _destructive_filter_overrides + + repo = _init_repo(tmp_path / "repo") + included = tmp_path / "included-filter.cfg" + included.write_text( + "[filter \"demo\"]\n" + "\tclean = cat\n" + "\trequired = true\n", + encoding="utf-8", + ) + _git(repo, "config", "include.path", str(included)) + + overrides = dict(_destructive_filter_overrides(repo, os.environ.copy())) + + assert overrides["filter.demo.clean"] == "cat" + assert overrides["filter.demo.smudge"] == "cat" + assert overrides["filter.demo.process"] == "" + assert overrides["filter.demo.required"] == "false" + + +def test_destructive_merge_driver_overrides_include_local_worktree_and_included_scope(tmp_path): + import os + + from api.workspace_git import _destructive_merge_driver_overrides + + repo = _init_repo(tmp_path / "repo") + included = tmp_path / "included-merge.cfg" + included.write_text( + "[merge \"included\"]\n" + "\tdriver = cat\n", + encoding="utf-8", + ) + _git(repo, "config", "include.path", str(included)) + _git(repo, "config", "merge.local.driver", "cat") + _git(repo, "config", "extensions.worktreeConfig", "true") + _git(repo, "config", "--worktree", "merge.worktree.driver", "cat") + + overrides = dict(_destructive_merge_driver_overrides(repo, os.environ.copy())) + + trusted_driver = 'git merge-file "%A" "%O" "%B"' + assert overrides["merge.included.driver"] == trusted_driver + assert overrides["merge.local.driver"] == trusted_driver + assert overrides["merge.worktree.driver"] == trusted_driver + + +def test_git_checkout_blocks_worktree_scope_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import ( + WORKSPACE_GIT_DESTRUCTIVE_ENV, + GitWorkspaceError, + git_checkout, + ) + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "filter-worktree-smudge-ran" + helper = tmp_path / "filter_worktree_checkout_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "extensions.worktreeConfig", "true") + _git(repo, "config", "--worktree", "filter.demo.clean", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "config", "--worktree", "filter.demo.smudge", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "branch", "-M", "main") + _git(repo, "add", ".gitattributes") + _git(repo, "commit", "-m", "Initial") + _git(repo, "branch", "feature") + _git(repo, "checkout", "feature") + (repo / "tracked.txt").write_text("from feature\n", encoding="utf-8") + _git(repo, "add", "tracked.txt") + _git(repo, "commit", "-m", "Feature update") + _git(repo, "checkout", "main") + marker.unlink(missing_ok=True) + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_checkout(repo, "feature", "local") + assert exc.value.code == "filtered_path" + assert not marker.exists() + + +def test_git_fetch_and_pull_disable_submodule_recursion(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + calls = [] + + def fake_run_git(ctx_or_cwd, args, **kwargs): + calls.append(list(args)) + return types.SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(workspace_git, "resolve_git_context", lambda workspace: ctx) + monkeypatch.setattr(workspace_git, "_run_git", fake_run_git) + monkeypatch.setattr(workspace_git, "git_status", lambda workspace: {"is_git": True}) + + workspace_git.git_fetch(tmp_path) + workspace_git.git_pull(tmp_path) + + assert ["fetch", "--prune", "--no-recurse-submodules"] in calls + assert ["pull", "--ff-only", "--no-recurse-submodules"] in calls + + +def test_git_fetch_forces_destructive_hardening_without_flag(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + captured = {} + + def fake_run_git(ctx_or_cwd, args, **kwargs): + captured["args"] = list(args) + captured["kwargs"] = dict(kwargs) + return types.SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(workspace_git, "resolve_git_context", lambda workspace: ctx) + monkeypatch.setattr(workspace_git, "_run_git", fake_run_git) + monkeypatch.setattr(workspace_git, "git_status", lambda workspace: {"is_git": True}) + + workspace_git.git_fetch(tmp_path) + + assert captured["args"] == ["fetch", "--prune", "--no-recurse-submodules"] + assert captured["kwargs"]["force_destructive_hardening"] is True + + +def test_run_git_force_destructive_hardening_applies_hook_redirect_without_flag(monkeypatch, tmp_path): + from api import workspace_git + + captured = {} + hooks_dir = tmp_path / "hooks" + hooks_dir.mkdir() + + def fake_subprocess_run(argv, **kwargs): + captured["argv"] = list(argv) + captured["kwargs"] = dict(kwargs) + return types.SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(workspace_git, "workspace_git_destructive_enabled", lambda: False) + monkeypatch.setattr(workspace_git.subprocess, "run", fake_subprocess_run) + monkeypatch.setattr(workspace_git.tempfile, "mkdtemp", lambda prefix: str(hooks_dir)) + + workspace_git._run_git(tmp_path, ["fetch"], force_destructive_hardening=True) + + assert any(str(arg).startswith("core.hooksPath=") for arg in captured["argv"]) + assert "core.alternateRefsCommand=" in captured["argv"] + + +def test_git_pull_blocks_repo_local_filters_before_run_when_destructive_mode_enabled(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + + monkeypatch.setattr(workspace_git, "resolve_git_context", lambda workspace: ctx) + monkeypatch.setattr(workspace_git, "workspace_git_destructive_enabled", lambda: True) + monkeypatch.setattr(workspace_git, "_has_repo_local_filters", lambda cwd, env: True) + + with pytest.raises(workspace_git.GitWorkspaceError) as exc: + workspace_git.git_pull(tmp_path) + + assert exc.value.code == "filtered_path" + + +def test_git_stage_blocks_repo_local_filters_before_run_when_destructive_mode_enabled(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + + monkeypatch.setattr(workspace_git, "resolve_git_context", lambda workspace: ctx) + monkeypatch.setattr(workspace_git, "workspace_git_destructive_enabled", lambda: True) + monkeypatch.setattr(workspace_git, "_has_repo_local_filters", lambda cwd, env: True) + + with pytest.raises(workspace_git.GitWorkspaceError) as exc: + workspace_git.git_stage(tmp_path, ["tracked.txt"]) + + assert exc.value.code == "filtered_path" + + +def test_git_fetch_skips_repo_local_reference_transaction_hook_without_destructive_mode(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("hook script setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_fetch + + remote = _init_bare_repo(tmp_path / "remote.git") + origin = _init_repo(tmp_path / "origin") + (origin / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(origin) + _git(origin, "branch", "-M", "main") + _git(origin, "remote", "add", "origin", str(remote)) + _git(origin, "push", "-u", "origin", "main") + _git(remote, "symbolic-ref", "HEAD", "refs/heads/main") + + clone = tmp_path / "clone" + _git(tmp_path, "clone", str(remote), str(clone)) + _git(clone, "config", "user.email", "hermes-tests@example.invalid") + _git(clone, "config", "user.name", "Hermes Tests") + + (origin / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + _commit_all(origin, "Remote update") + _git(origin, "push") + + hooks = tmp_path / "hooks" + hooks.mkdir() + marker = tmp_path / "reference-transaction-ran" + helper = tmp_path / "reference_transaction_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('reference-transaction executed', encoding='utf-8')\n" + "sys.stdin.read()\n", + encoding="utf-8", + ) + helper.chmod(0o755) + hook = hooks / "reference-transaction" + hook.write_text( + "#!/bin/sh\n" + f"\"{sys.executable}\" \"{helper}\" \"{marker}\" \"$@\"\n", + encoding="utf-8", + ) + hook.chmod(0o755) + _git(clone, "config", "core.hooksPath", str(hooks)) + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + fetched = git_fetch(clone) + + assert fetched["status"]["behind"] == 1 + assert not marker.exists() + + +def test_git_checkout_disables_submodule_recursion(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + calls = [] + + def fake_run_git(ctx_or_cwd, args, **kwargs): + calls.append(list(args)) + return types.SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(workspace_git, "resolve_git_context", lambda workspace: ctx) + monkeypatch.setattr(workspace_git, "_run_git", fake_run_git) + monkeypatch.setattr(workspace_git, "_dirty_worktree", lambda ctx: False) + monkeypatch.setattr(workspace_git, "_current_checkout_label", lambda ctx: "main") + monkeypatch.setattr(workspace_git, "_restore_branch_switch_stash_locked", lambda ctx, branch: {}) + monkeypatch.setattr(workspace_git, "git_status", lambda workspace: {"is_git": True, "branch": "feature"}) + monkeypatch.setattr(workspace_git, "git_branches", lambda workspace: {"current": "feature"}) + + result = workspace_git.git_checkout(tmp_path, "feature", "local") + + assert result["current_branch"] == "feature" + assert ["switch", "--recurse-submodules=no", "feature"] in calls + + +def test_perform_checkout_locked_disables_submodule_recursion_across_modes(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + calls = [] + + def fake_run_git(ctx_or_cwd, args, **kwargs): + calls.append(list(args)) + if args[:2] == ["show-ref", "--verify"]: + return types.SimpleNamespace(stdout="", stderr="", returncode=1) + return types.SimpleNamespace(stdout="", stderr="", returncode=0) + + monkeypatch.setattr(workspace_git, "_run_git", fake_run_git) + monkeypatch.setattr(workspace_git, "_validate_local_branch", lambda ctx, ref: ref) + monkeypatch.setattr(workspace_git, "_validate_new_branch_name", lambda ctx, name: name) + monkeypatch.setattr(workspace_git, "_validate_checkout_start", lambda ctx, ref: ref) + monkeypatch.setattr(workspace_git, "_validate_remote_branch", lambda ctx, ref: ref) + + workspace_git._perform_checkout_locked(ctx, tmp_path, "feature", "new", "topic", False) + workspace_git._perform_checkout_locked(ctx, tmp_path, "origin/topic", "remote", "topic", True) + workspace_git._perform_checkout_locked(ctx, tmp_path, "deadbeef", "detach", None, False) + + assert ["switch", "--recurse-submodules=no", "-c", "topic", "feature"] in calls + assert ["switch", "--recurse-submodules=no", "-c", "topic", "--track", "origin/topic"] in calls + assert ["switch", "--recurse-submodules=no", "--detach", "deadbeef"] in calls + + +def test_git_stage_skips_included_repo_local_filters_when_destructive_mode_disabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_stage + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "included-filter-ran" + helper = tmp_path / "included_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + included = tmp_path / "included-filter.cfg" + included.write_text( + "[filter \"demo\"]\n" + f"\tclean = \"{sys.executable}\" \"{helper}\" \"{marker}\"\n" + f"\tsmudge = \"{sys.executable}\" \"{helper}\" \"{marker}\"\n", + encoding="utf-8", + ) + _git(repo, "config", "include.path", str(included)) + + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker.unlink(missing_ok=True) + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + staged = git_stage(repo, ["tracked.txt"]) + + assert staged["totals"]["staged"] == 1 + assert not marker.exists() + + +def test_selected_temp_index_env_blocks_repo_local_filters_when_destructive_mode_enabled(monkeypatch, tmp_path): + from api import workspace_git + + ctx = workspace_git.GitContext(tmp_path, tmp_path, "") + + monkeypatch.setattr(workspace_git, "workspace_git_destructive_enabled", lambda: True) + monkeypatch.setattr(workspace_git, "_has_repo_local_filters", lambda cwd, env: True) + + with pytest.raises(workspace_git.GitWorkspaceError) as exc: + workspace_git._selected_temp_index_env(ctx, ["tracked.txt"]) + + assert exc.value.code == "filtered_path" + + +def test_git_stage_blocks_included_repo_local_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, GitWorkspaceError, git_stage + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "included-filter-block-ran" + helper = tmp_path / "included_filter_block_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + included = tmp_path / "included-filter-block.cfg" + included.write_text( + "[filter \"demo\"]\n" + f"\tclean = \"{sys.executable}\" \"{helper}\" \"{marker}\"\n" + f"\tsmudge = \"{sys.executable}\" \"{helper}\" \"{marker}\"\n", + encoding="utf-8", + ) + _git(repo, "config", "include.path", str(included)) + + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker.unlink(missing_ok=True) + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_stage(repo, ["tracked.txt"]) + + assert exc.value.code == "filtered_path" + assert not marker.exists() + + +def test_git_stash_restore_skips_repo_local_merge_drivers_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted merge driver setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_stash_and_checkout + + repo = _init_repo(tmp_path / "repo") + _git(repo, "branch", "-M", "main") + (repo / ".gitattributes").write_text("tracked.txt merge=demo\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + _git(repo, "checkout", "-b", "feature") + _git(repo, "checkout", "main") + (repo / "tracked.txt").write_text("main dirty\n", encoding="utf-8") + + marker = tmp_path / "merge-driver-ran" + helper = tmp_path / "merge_driver_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('merge driver ran', encoding='utf-8')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "merge.demo.driver", f'"{sys.executable}" "{helper}" "{marker}"') + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + git_stash_and_checkout(repo, "feature", "local") + _git(repo, "checkout", "main") + (repo / "tracked.txt").write_text("main changed while parked\n", encoding="utf-8") + _commit_all(repo, "advance main") + _git(repo, "checkout", "feature") + + result = git_stash_and_checkout(repo, "main", "local") + + assert result["ok"] is True + assert result["current_branch"] == "main" + assert result["restore_failed"] is True + assert result["restore_stash"]["branch"] == "main" + assert not marker.exists() + + +def test_git_commit_skips_repo_local_gpg_program_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("non-interactive gpg helper test is POSIX-only") + + from api.workspace_git import ( + WORKSPACE_GIT_DESTRUCTIVE_ENV, + git_commit, + git_stage, + ) + + repo = _init_repo(tmp_path / "repo") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "gpg-ran" + helper = tmp_path / "gpg_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('gpg executed', encoding='utf-8')\n" + "raise SystemExit(1)\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "commit.gpgsign", "true") + _git(repo, "config", "gpg.program", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + git_stage(repo, ["tracked.txt"]) + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + result = git_commit(repo, "Signed commit path blocked") + + assert result["ok"] is True + assert not marker.exists() + + +def test_git_pull_blocks_repo_local_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, GitWorkspaceError, git_pull + + remote = _init_bare_repo(tmp_path / "remote.git") + origin = _init_repo(tmp_path / "origin") + (origin / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + (origin / "tracked.txt").write_text("one\n", encoding="utf-8") + _git(origin, "config", "filter.demo.clean", "cat") + _git(origin, "config", "filter.demo.smudge", "cat") + _commit_all(origin) + _git(origin, "branch", "-M", "main") + _git(origin, "remote", "add", "origin", str(remote)) + _git(origin, "push", "-u", "origin", "main") + _git(remote, "symbolic-ref", "HEAD", "refs/heads/main") + + clone = tmp_path / "clone" + _git(tmp_path, "clone", str(remote), str(clone)) + _git(clone, "config", "user.email", "hermes-tests@example.invalid") + _git(clone, "config", "user.name", "Hermes Tests") + _git(clone, "config", "filter.demo.clean", f"\"{sys.executable}\" -c \"import sys; print(sys.stdin.read(), end='')\"") + _git(clone, "config", "filter.demo.smudge", f"\"{sys.executable}\" -c \"import sys; print(sys.stdin.read(), end='')\"") + + (origin / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + _commit_all(origin, "Remote update") + _git(origin, "push") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_pull(clone) + + assert exc.value.code == "filtered_path" + assert (clone / "tracked.txt").read_text(encoding="utf-8") == "one\n" + + +def test_git_commit_selected_blocks_repo_local_filters_when_destructive_mode_enabled(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter helper setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, GitWorkspaceError, git_commit_selected + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + marker = tmp_path / "selected-filter-ran" + helper = tmp_path / "selected_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + _git(repo, "config", "filter.demo.smudge", f"\"{sys.executable}\" \"{helper}\" \"{marker}\"") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker.unlink(missing_ok=True) + (repo / "tracked.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_commit_selected(repo, "Commit selected path", ["tracked.txt"]) + + assert exc.value.code == "filtered_path" + assert not marker.exists() + + def test_git_env_scrub_removes_redirecting_vars_and_preserves_temp_index(monkeypatch): from api.workspace_git import _clean_git_env @@ -1192,3 +2345,173 @@ def test_git_commit_route_rejects_active_stream(monkeypatch, tmp_path): payload = handler.payload() assert payload["code"] == "active_stream" assert "active" in payload["error"].lower() + + +def test_selected_commit_message_prompt_skips_clean_filter_without_destructive_mode(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted clean filter setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, selected_commit_message_prompt + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + (repo / "selected.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "selected-clean-filter-ran" + helper = tmp_path / "selected_clean_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('clean filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f'"{sys.executable}" "{helper}" "{marker}"') + (repo / "selected.txt").write_text("one\ntwo\n", encoding="utf-8") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + prompt = selected_commit_message_prompt(repo, ["selected.txt"]) + + assert "+two" in prompt["user_prompt"] + assert not marker.exists() + + +def test_staged_commit_message_prompt_skips_clean_filter_without_destructive_mode(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted clean filter setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, staged_commit_message_prompt + + repo = _init_repo(tmp_path / "repo") + # Install .gitattributes without a filter program so git add runs as identity + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + (repo / "staged.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + + # Stage the modification before the malicious filter program is registered, + # so the git add step itself does not trigger the payload. + (repo / "staged.txt").write_text("one\ntwo\n", encoding="utf-8") + _git(repo, "add", "staged.txt") + + # Now install the malicious filter — staged_commit_message_prompt's git diff + # --cached must not invoke clean filters on index-vs-HEAD comparisons. + marker = tmp_path / "staged-clean-filter-ran" + helper = tmp_path / "staged_clean_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + "pathlib.Path(sys.argv[1]).write_text('clean filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f'"{sys.executable}" "{helper}" "{marker}"') + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + prompt = staged_commit_message_prompt(repo) + + assert "+two" in prompt["user_prompt"] + assert not marker.exists() + + +def test_git_hardened_config_blocks_submodule_recursion(): + from api.workspace_git import _GIT_HARDENED_CONFIG + + config = dict(_GIT_HARDENED_CONFIG) + assert config.get("submodule.recurse") == "false" + assert config.get("fetch.recurseSubmodules") == "false" + + +def test_git_status_blocks_injection_via_filter_name_with_equals(tmp_path, monkeypatch): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, git_status + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=evil=core.sshCommand\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "injected-command-ran" + helper = tmp_path / "injected_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + f"pathlib.Path('{marker}').write_text('injected', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.evil=core.sshCommand.clean", f'"{sys.executable}" "{helper}"') + _git(repo, "config", "core.sshCommand", f'"{sys.executable}" "{helper}"') + (repo / "tracked.txt").write_text("two\n", encoding="utf-8") + + monkeypatch.delenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, raising=False) + status = git_status(repo) + + assert status["is_git"] is True + assert not marker.exists() + + +def test_git_discard_blocks_when_repo_local_filters_present(tmp_path, monkeypatch): + import os + + if os.name == "nt": + pytest.skip("scripted filter setup is POSIX-only") + + from api.workspace_git import WORKSPACE_GIT_DESTRUCTIVE_ENV, GitWorkspaceError, git_discard + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=keyword\n", encoding="utf-8") + (repo / "tracked.txt").write_text("original\n", encoding="utf-8") + _commit_all(repo) + _git(repo, "config", "filter.keyword.clean", "cat") + _git(repo, "config", "filter.keyword.smudge", "tr a-z A-Z") + (repo / "tracked.txt").write_text("modified\n", encoding="utf-8") + + monkeypatch.setenv(WORKSPACE_GIT_DESTRUCTIVE_ENV, "1") + with pytest.raises(GitWorkspaceError) as exc: + git_discard(repo, ["tracked.txt"]) + assert exc.value.code == "filtered_path" + assert (repo / "tracked.txt").read_text(encoding="utf-8") == "modified\n" + + +def test_dirty_worktree_uses_filter_neutralization(tmp_path): + import os + import sys + + if os.name == "nt": + pytest.skip("scripted filter setup is POSIX-only") + + from api.workspace_git import _dirty_worktree, resolve_git_context + + repo = _init_repo(tmp_path / "repo") + (repo / ".gitattributes").write_text("*.txt filter=demo\n", encoding="utf-8") + (repo / "tracked.txt").write_text("one\n", encoding="utf-8") + _commit_all(repo) + marker = tmp_path / "dirty-worktree-filter-ran" + helper = tmp_path / "dirty_filter_helper.py" + helper.write_text( + "#!/usr/bin/env python3\n" + "import pathlib, sys\n" + f"pathlib.Path('{marker}').write_text('filter ran', encoding='utf-8')\n" + "print(sys.stdin.read(), end='')\n", + encoding="utf-8", + ) + helper.chmod(0o755) + _git(repo, "config", "filter.demo.clean", f'"{sys.executable}" "{helper}"') + + ctx = resolve_git_context(repo) + _dirty_worktree(ctx) + + assert not marker.exists()