mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-09-10 02:25:56 +00:00
Scrub unproven hook Git environment
This commit is contained in:
@@ -158,6 +158,75 @@ class RegistryAuditTest(unittest.TestCase):
|
||||
self.assertEqual("", self.git_stdout(caller_worktree, "status", "--porcelain=v1"))
|
||||
self.assertFalse((caller_worktree / "internal").exists())
|
||||
|
||||
def test_tracked_workspace_files_scrubs_linked_worktree_index_env_without_work_tree(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
tmp_root = Path(tmpdir)
|
||||
caller_workspace = tmp_root / "caller" / "workspace"
|
||||
caller_repo = caller_workspace / "repos" / "pulse"
|
||||
caller_worktree = caller_workspace / ".worktrees" / "pulse-first-session-onboarding-parity"
|
||||
caller_repo.mkdir(parents=True)
|
||||
caller_worktree.parent.mkdir(parents=True)
|
||||
|
||||
self.git(caller_repo, "init")
|
||||
(caller_repo / "README.md").write_text("caller\n", encoding="utf-8")
|
||||
self.git(caller_repo, "add", "README.md")
|
||||
self.git(
|
||||
caller_repo,
|
||||
"-c",
|
||||
"user.name=Pulse Test",
|
||||
"-c",
|
||||
"user.email=pulse-test@example.invalid",
|
||||
"commit",
|
||||
"-m",
|
||||
"initial",
|
||||
)
|
||||
self.git(caller_repo, "worktree", "add", "--detach", str(caller_worktree), "HEAD")
|
||||
|
||||
hook_env = self.hook_env_for_worktree(caller_worktree)
|
||||
hook_env.pop("GIT_WORK_TREE")
|
||||
|
||||
fixture_workspace = tmp_root / "fixture" / "workspace"
|
||||
repo_root = fixture_workspace / "repos" / "pulse"
|
||||
linked_worktree = fixture_workspace / ".worktrees" / "pulse-first-session-onboarding-parity"
|
||||
repo_root.mkdir(parents=True)
|
||||
linked_worktree.parent.mkdir(parents=True)
|
||||
|
||||
self.git(repo_root, "init")
|
||||
(repo_root / "internal").mkdir()
|
||||
(repo_root / "internal" / "existing.go").write_text("package internal\n", encoding="utf-8")
|
||||
self.git(repo_root, "add", "internal/existing.go")
|
||||
self.git(
|
||||
repo_root,
|
||||
"-c",
|
||||
"user.name=Pulse Test",
|
||||
"-c",
|
||||
"user.email=pulse-test@example.invalid",
|
||||
"commit",
|
||||
"-m",
|
||||
"initial",
|
||||
)
|
||||
self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD")
|
||||
|
||||
(linked_worktree / "internal" / "staged.go").write_text("package internal\n", encoding="utf-8")
|
||||
self.git(linked_worktree, "add", "internal/staged.go")
|
||||
contracts_dir = linked_worktree / "docs" / "release-control" / "v6" / "internal" / "subsystems"
|
||||
contracts_dir.mkdir(parents=True)
|
||||
|
||||
with patch("registry_audit.REPO_ROOT", linked_worktree), patch(
|
||||
"registry_audit.DEFAULT_CONTROL_PLANE",
|
||||
{
|
||||
**registry_audit.DEFAULT_CONTROL_PLANE,
|
||||
"subsystems_dir_path": str(contracts_dir),
|
||||
},
|
||||
), patch.dict(os.environ, hook_env, clear=False):
|
||||
files = tracked_workspace_files(
|
||||
active_repos=["pulse"],
|
||||
local_repo=canonical_repo_id(linked_worktree),
|
||||
)
|
||||
|
||||
self.assertIn("internal/staged.go", files)
|
||||
self.assertNotIn("README.md", files)
|
||||
|
||||
def test_tracked_workspace_files_scrubs_hook_env_for_sibling_repos(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
workspace = Path(tmpdir) / "workspace"
|
||||
|
||||
@@ -29,6 +29,20 @@ LOCAL_GIT_ENV_VARS = (
|
||||
"GIT_SHALLOW_FILE",
|
||||
"GIT_COMMON_DIR",
|
||||
)
|
||||
REPO_TARGET_GIT_ENV_VARS = (
|
||||
"GIT_ALTERNATE_OBJECT_DIRECTORIES",
|
||||
"GIT_OBJECT_DIRECTORY",
|
||||
"GIT_DIR",
|
||||
"GIT_WORK_TREE",
|
||||
"GIT_IMPLICIT_WORK_TREE",
|
||||
"GIT_GRAFT_FILE",
|
||||
"GIT_INDEX_FILE",
|
||||
"GIT_NO_REPLACE_OBJECTS",
|
||||
"GIT_REPLACE_REF_BASE",
|
||||
"GIT_PREFIX",
|
||||
"GIT_SHALLOW_FILE",
|
||||
"GIT_COMMON_DIR",
|
||||
)
|
||||
|
||||
|
||||
def strip_local_git_env(env: dict[str, str]) -> dict[str, str]:
|
||||
@@ -37,12 +51,83 @@ def strip_local_git_env(env: dict[str, str]) -> dict[str, str]:
|
||||
return env
|
||||
|
||||
|
||||
def resolve_env_path(value: str, base: Path) -> Path:
|
||||
path = Path(value)
|
||||
if not path.is_absolute():
|
||||
path = base / path
|
||||
return path.resolve()
|
||||
|
||||
|
||||
def env_path_matches(value: str, expected: Path, bases: Iterable[Path]) -> bool:
|
||||
expected = expected.resolve()
|
||||
return any(resolve_env_path(value, base) == expected for base in bases)
|
||||
|
||||
|
||||
def repo_git_dir(repo_root: Path) -> Path | None:
|
||||
dot_git = repo_root / ".git"
|
||||
try:
|
||||
if dot_git.is_dir():
|
||||
return dot_git.resolve()
|
||||
if not dot_git.is_file():
|
||||
return None
|
||||
gitdir_prefix = "gitdir:"
|
||||
for line in dot_git.read_text(encoding="utf-8").splitlines():
|
||||
if line.startswith(gitdir_prefix):
|
||||
git_dir = Path(line[len(gitdir_prefix) :].strip())
|
||||
if not git_dir.is_absolute():
|
||||
git_dir = dot_git.parent / git_dir
|
||||
return git_dir.resolve()
|
||||
except OSError:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def repo_common_dir(git_dir: Path) -> Path | None:
|
||||
common_dir_file = git_dir / "commondir"
|
||||
try:
|
||||
if not common_dir_file.is_file():
|
||||
return git_dir.resolve()
|
||||
common_dir = Path(common_dir_file.read_text(encoding="utf-8").strip())
|
||||
if not common_dir.is_absolute():
|
||||
common_dir = git_dir / common_dir
|
||||
return common_dir.resolve()
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def git_env_targets_repo(env: dict[str, str], repo_root: Path) -> bool:
|
||||
if not any(name in env for name in REPO_TARGET_GIT_ENV_VARS):
|
||||
return True
|
||||
|
||||
repo_root = repo_root.resolve()
|
||||
work_tree = env.get("GIT_WORK_TREE")
|
||||
if not work_tree:
|
||||
return True
|
||||
return False
|
||||
|
||||
try:
|
||||
return Path(work_tree).resolve() == repo_root.resolve()
|
||||
if resolve_env_path(work_tree, repo_root) != repo_root:
|
||||
return False
|
||||
|
||||
git_dir = repo_git_dir(repo_root)
|
||||
if git_dir is None:
|
||||
return not any(name in env for name in ("GIT_DIR", "GIT_COMMON_DIR", "GIT_INDEX_FILE"))
|
||||
|
||||
env_git_dir = env.get("GIT_DIR")
|
||||
if env_git_dir and not env_path_matches(env_git_dir, git_dir, (repo_root,)):
|
||||
return False
|
||||
|
||||
common_dir = repo_common_dir(git_dir)
|
||||
env_common_dir = env.get("GIT_COMMON_DIR")
|
||||
if env_common_dir and (
|
||||
common_dir is None or not env_path_matches(env_common_dir, common_dir, (repo_root, git_dir))
|
||||
):
|
||||
return False
|
||||
|
||||
env_index_file = env.get("GIT_INDEX_FILE")
|
||||
if env_index_file and not env_path_matches(env_index_file, git_dir / "index", (repo_root, git_dir)):
|
||||
return False
|
||||
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
@@ -31,6 +31,20 @@ class RepoFileIoTest(unittest.TestCase):
|
||||
env=env,
|
||||
)
|
||||
|
||||
def git_stdout(self, repo_root: Path, *args: str) -> str:
|
||||
return self.git(repo_root, *args).stdout.strip()
|
||||
|
||||
def hook_env_for_worktree(self, worktree_root: Path) -> dict[str, str]:
|
||||
git_dir = self.git_stdout(worktree_root, "rev-parse", "--path-format=absolute", "--git-dir")
|
||||
common_dir = self.git_stdout(worktree_root, "rev-parse", "--path-format=absolute", "--git-common-dir")
|
||||
work_tree = self.git_stdout(worktree_root, "rev-parse", "--show-toplevel")
|
||||
return {
|
||||
"GIT_DIR": git_dir,
|
||||
"GIT_WORK_TREE": work_tree,
|
||||
"GIT_INDEX_FILE": str(Path(git_dir) / "index"),
|
||||
"GIT_COMMON_DIR": common_dir,
|
||||
}
|
||||
|
||||
def test_read_repo_text_and_load_repo_json_can_read_staged_content(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
repo_root = Path(tmpdir)
|
||||
@@ -91,6 +105,7 @@ class RepoFileIoTest(unittest.TestCase):
|
||||
other_local_repo.mkdir(parents=True)
|
||||
sibling_repo.mkdir(parents=True)
|
||||
|
||||
self.git(local_repo, "init")
|
||||
leaked_env = {
|
||||
"PATH": os.environ.get("PATH", ""),
|
||||
"GIT_DIR": str(local_repo / ".git"),
|
||||
@@ -121,6 +136,40 @@ class RepoFileIoTest(unittest.TestCase):
|
||||
self.assertNotIn("GIT_COMMON_DIR", mismatched_local_env)
|
||||
self.assertNotIn("GIT_PREFIX", mismatched_local_env)
|
||||
|
||||
def test_git_env_scrubs_unproven_linked_worktree_index_env(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
workspace = Path(tmpdir) / "workspace"
|
||||
repo_root = workspace / "repos" / "pulse"
|
||||
linked_worktree = workspace / ".worktrees" / "pulse-first-session-onboarding-parity"
|
||||
repo_root.mkdir(parents=True)
|
||||
linked_worktree.parent.mkdir(parents=True)
|
||||
|
||||
self.git(repo_root, "init")
|
||||
(repo_root / "README.md").write_text("pulse\n", encoding="utf-8")
|
||||
self.git(repo_root, "add", "README.md")
|
||||
self.git(
|
||||
repo_root,
|
||||
"-c",
|
||||
"user.name=Pulse Test",
|
||||
"-c",
|
||||
"user.email=pulse-test@example.invalid",
|
||||
"commit",
|
||||
"-m",
|
||||
"initial",
|
||||
)
|
||||
self.git(repo_root, "worktree", "add", "--detach", str(linked_worktree), "HEAD")
|
||||
|
||||
hook_env = self.hook_env_for_worktree(linked_worktree)
|
||||
hook_env.pop("GIT_WORK_TREE")
|
||||
|
||||
with patch.dict(os.environ, hook_env, clear=True):
|
||||
env = git_env(linked_worktree, local_repo_root=linked_worktree)
|
||||
|
||||
self.assertNotIn("GIT_DIR", env)
|
||||
self.assertNotIn("GIT_WORK_TREE", env)
|
||||
self.assertNotIn("GIT_INDEX_FILE", env)
|
||||
self.assertNotIn("GIT_COMMON_DIR", env)
|
||||
|
||||
def test_canonical_repo_identity_uses_git_common_dir_for_linked_worktree(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
workspace = Path(tmpdir) / "workspace"
|
||||
|
||||
Reference in New Issue
Block a user