Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 101 Pull Requests 11 Actions Packages Projects Releases Wiki Activity

fix(runtime): support validated cross-repository canonical roots (#973)

Browse Source
This commit is contained in:
sysadmin
2026-07-29 13:19:29 -04:00
parent 626be8b178
commit 6a53308473
4 changed files with 271 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_config.py
+4
View File
@@ -1180,6 +1180,10 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({
"GITEA_SERVER_PROVENANCE",
"GITEA_AUTHOR_WORKTREE",
"GITEA_ACTIVE_WORKTREE",
"GITEA_REVIEWER_WORKTREE",
"GITEA_MERGER_WORKTREE",
"GITEA_CANONICAL_REPOSITORY_ROOT",
"GITEA_MCP_SESSION_STATE_TTL_HOURS",
"GITEA_DISABLE_KEYCHAIN",
"GITEA_CONTROL_PLANE_DB",
"GITEA_DB_PATH",
namespace_workspace_binding.py
+106 -5
View File
@@ -8,8 +8,10 @@ poison workspace purity checks in another namespace.
from __future__ import annotations
import os
import subprocess
import author_mutation_worktree as amw
import canonical_repository_root as crr
ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV
@@ -152,6 +154,60 @@ def resolve_namespace_workspace(
return os.path.realpath(process_project_root), "MCP server process root (default)"
def verify_git_common_directory_membership(
workspace_path: str,
canonical_repo_root: str,
) -> tuple[bool, str | None]:
"""Verify that workspace_path belongs to canonical_repo_root via git common-dir or branches containment."""
ws = (workspace_path or "").strip()
root = (canonical_repo_root or "").strip()
if not ws or not root:
return False, "empty workspace or canonical root path"
try:
real_ws = os.path.realpath(os.path.abspath(ws))
real_root = os.path.realpath(os.path.abspath(root))
except Exception as exc:
return False, f"invalid workspace or root path: {exc}"
if real_ws == real_root:
return True, None
if not os.path.isdir(real_ws):
return False, f"workspace directory '{real_ws}' does not exist"
try:
res = subprocess.run(
["git", "-C", real_ws, "rev-parse", "--git-common-dir"],
capture_output=True,
text=True,
check=False,
)
if res.returncode == 0:
common_raw = (res.stdout or "").strip()
common_dir = amw._realpath_git_common_dir(real_ws, common_raw)
real_common = os.path.realpath(common_dir)
canonical_git = os.path.realpath(os.path.join(real_root, ".git"))
if real_common in (canonical_git, real_root) or os.path.dirname(real_common) == real_root:
return True, None
return (
False,
f"workspace '{real_ws}' git common directory '{real_common}' does not match "
f"canonical repository root '{real_root}' (.git at '{canonical_git}')"
)
except Exception:
pass
if amw.is_path_under_branches(real_ws, real_root):
return True, None
return (
False,
f"workspace '{real_ws}' does not belong to canonical repository root '{real_root}'"
)
def resolve_namespace_mutation_context(
*,
role_kind: str,
@@ -163,6 +219,8 @@ def resolve_namespace_mutation_context(
worktree: str | None = None,
profile_name: str | None = None,
configured_canonical_root: str | None = None,
expected_slug: str | None = None,
remote: str | None = None,
) -> dict:
"""Shared workspace resolution for runtime_context and mutation guards.
@@ -180,11 +238,30 @@ def resolve_namespace_mutation_context(
env_map = env if env is not None else os.environ
process_root = os.path.realpath(process_project_root)
role = normalize_role_kind(role_kind, profile_name=profile_name)
configured = (configured_canonical_root or "").strip()
if configured:
canonical_root = os.path.realpath(configured)
configured_val = (configured_canonical_root or "").strip()
if configured_val:
crr_assessment = crr.assess_canonical_repository_root(
configured_value=configured_val,
source="configured_canonical_root",
expected_slug=expected_slug,
process_project_root=process_root,
remote=remote,
)
canonical_root = crr_assessment["canonical_repo_root"]
roots_aligned = crr_assessment["proven"]
else:
canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
crr_assessment = {
"proven": True,
"block": False,
"reasons": [],
"configured": False,
"canonical_repo_root": amw.resolve_canonical_repo_root(process_root, process_root),
"resolved_slug": None,
"source": None,
}
canonical_root = crr_assessment["canonical_repo_root"]
roots_aligned = (canonical_root == process_root)
durable: dict | None = None
if role == "author":
@@ -234,7 +311,8 @@ def resolve_namespace_mutation_context(
"ignored_bindings": demotions + (pollution.get("ignored_bindings") or []),
"process_project_root": process_root,
"canonical_repo_root": canonical_root,
"roots_aligned": canonical_root == process_root,
"roots_aligned": roots_aligned,
"canonical_root_assessment": crr_assessment,
}
if durable is not None:
result["author_worktree_resolution"] = durable
@@ -387,6 +465,8 @@ def assess_namespace_mutation_workspace(
profile_name: str | None = None,
current_branch: str | None = None,
configured_canonical_root: str | None = None,
expected_slug: str | None = None,
remote: str | None = None,
) -> dict:
"""Evaluate namespace workspace binding before preflight/mutation."""
ctx = resolve_namespace_mutation_context(
@@ -399,6 +479,8 @@ def assess_namespace_mutation_workspace(
session_lock_worktree=session_lock_worktree,
profile_name=profile_name,
configured_canonical_root=configured_canonical_root,
expected_slug=expected_slug,
remote=remote,
)
mutation_workspace = ctx["workspace_path"]
binding_source = ctx["workspace_binding_source"]
@@ -422,6 +504,25 @@ def assess_namespace_mutation_workspace(
reasons = list(metadata.get("reasons") or [])
operator_recovery = ctx.get("operator_recovery")
crr_reasons = list(ctx.get("canonical_root_assessment", {}).get("reasons") or [])
if crr_reasons:
reasons.extend(crr_reasons)
if not ctx.get("roots_aligned"):
if not crr_reasons:
reasons.append(
f"unsafe_process_root_workspace_alignment: process root '{process_root}' "
f"and canonical root '{ctx['canonical_repo_root']}' disagree"
)
if os.path.exists(mutation_workspace):
valid_common, common_err = verify_git_common_directory_membership(
mutation_workspace, ctx["canonical_repo_root"]
)
if not valid_common and common_err:
reasons.append(common_err)
if role == "author":
# #618 durable resolution already validated existence, membership,
# branches/, lock ownership, and traversal safety when present.
tests/test_issue_706_canonical_repository_root.py
+1 -1
View File
@@ -243,7 +243,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
configured_canonical_root=self.target,
)
self.assertEqual(ctx["canonical_repo_root"], self.target)
self.assertFalse(ctx["roots_aligned"])
self.assertTrue(ctx["roots_aligned"])
def test_target_worktree_is_member_of_target_root(self):
got = nwb.amw.assess_workspace_repo_membership(
tests/test_issue_973_cross_repo_canonical_roots.py
+160
View File
@@ -0,0 +1,160 @@
"""Regression tests for Issue #973: validated cross-repository canonical roots."""
from __future__ import annotations
import os
import shutil
import tempfile
import unittest
from pathlib import Path
import subprocess
import gitea_config
import namespace_workspace_binding as nwb
import canonical_repository_root as crr
class TestIssue973RecognizedEnvKeys(unittest.TestCase):
"""Test recognized environment variable keys under #973."""
def test_recognized_gitea_env_keys(self):
for key in (
"GITEA_CANONICAL_REPOSITORY_ROOT",
"GITEA_REVIEWER_WORKTREE",
"GITEA_MERGER_WORKTREE",
"GITEA_MCP_SESSION_STATE_TTL_HOURS",
):
self.assertIn(key, gitea_config.RECOGNIZED_GITEA_ENV_KEYS)
def test_get_unconsumed_gitea_env_overrides_ignores_recognized(self):
env = {
"GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path",
"GITEA_REVIEWER_WORKTREE": "/some/reviewer/path",
"GITEA_MERGER_WORKTREE": "/some/merger/path",
"GITEA_MCP_SESSION_STATE_TTL_HOURS": "24",
"GITEA_UNRECOGNIZED_FOO_VAR": "bar",
}
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
self.assertNotIn("GITEA_CANONICAL_REPOSITORY_ROOT", unconsumed)
self.assertNotIn("GITEA_REVIEWER_WORKTREE", unconsumed)
self.assertNotIn("GITEA_MERGER_WORKTREE", unconsumed)
self.assertNotIn("GITEA_MCP_SESSION_STATE_TTL_HOURS", unconsumed)
self.assertIn("GITEA_UNRECOGNIZED_FOO_VAR", unconsumed)
class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase):
"""Test workspace binding and canonical root validation for cross-repo namespaces."""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.tmp_dir = os.path.realpath(self._tmp.name)
# Create simulated installation root
self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools")
os.makedirs(self.install_root)
subprocess.run(["git", "init", "-b", "master"], cwd=self.install_root, check=True)
# Create simulated target repository root
self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane")
os.makedirs(self.target_root)
subprocess.run(["git", "init", "-b", "master"], cwd=self.target_root, check=True)
# Create branches/ directory and a valid worktree in target repository
self.target_branches = os.path.join(self.target_root, "branches")
os.makedirs(self.target_branches)
self.target_worktree = os.path.join(self.target_branches, "rev-pr-99")
os.makedirs(self.target_worktree)
# Create git common dir linking target_worktree to target_root
dot_git_file = os.path.join(self.target_worktree, ".git")
git_dir_target = os.path.join(self.target_root, ".git", "worktrees", "rev-pr-99")
os.makedirs(git_dir_target, exist_ok=True)
with open(dot_git_file, "w") as f:
f.write(f"gitdir: {git_dir_target}\n")
with open(os.path.join(git_dir_target, "commondir"), "w") as f:
f.write("../../..\n")
def tearDown(self):
self._tmp.cleanup()
def test_valid_cross_repo_canonical_root(self):
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
worktree_path=self.target_worktree,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertEqual(ctx["canonical_repo_root"], self.target_root)
self.assertTrue(ctx["roots_aligned"])
self.assertTrue(ctx["canonical_root_assessment"]["proven"])
def test_nonexistent_configured_canonical_root(self):
nonexistent = os.path.join(self.tmp_dir, "nonexistent-repo")
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
worktree_path=self.target_worktree,
process_project_root=self.install_root,
env={},
configured_canonical_root=nonexistent,
)
self.assertFalse(ctx["roots_aligned"])
self.assertFalse(ctx["canonical_root_assessment"]["proven"])
self.assertTrue(any("does not exist" in r for r in ctx["canonical_root_assessment"]["reasons"]))
def test_non_git_configured_canonical_root(self):
non_git = os.path.join(self.tmp_dir, "non-git-dir")
os.makedirs(non_git)
ctx = nwb.resolve_namespace_mutation_context(
role_kind="reviewer",
worktree_path=self.target_worktree,
process_project_root=self.install_root,
env={},
configured_canonical_root=non_git,
)
self.assertFalse(ctx["roots_aligned"])
self.assertFalse(ctx["canonical_root_assessment"]["proven"])
self.assertTrue(any("not a git repository" in r for r in ctx["canonical_root_assessment"]["reasons"]))
def test_git_common_directory_membership_matching(self):
valid, err = nwb.verify_git_common_directory_membership(
self.target_worktree, self.target_root
)
self.assertTrue(valid, err)
self.assertIsNone(err)
def test_git_common_directory_membership_foreign_repo_blocks(self):
# Foreign worktree created under install_root
foreign_wt = os.path.join(self.install_root, "branches", "foreign-wt")
os.makedirs(foreign_wt)
dot_git_file = os.path.join(foreign_wt, ".git")
git_dir_install = os.path.join(self.install_root, ".git", "worktrees", "foreign-wt")
os.makedirs(git_dir_install, exist_ok=True)
with open(dot_git_file, "w") as f:
f.write(f"gitdir: {git_dir_install}\n")
with open(os.path.join(git_dir_install, "commondir"), "w") as f:
f.write("../../..\n")
valid, err = nwb.verify_git_common_directory_membership(
foreign_wt, self.target_root
)
self.assertFalse(valid)
self.assertIn("does not belong", err)
def test_reviewer_worktree_outside_branches_blocks(self):
outside_wt = os.path.join(self.target_root, "outside_branches_wt")
os.makedirs(outside_wt)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind="reviewer",
worktree_path=outside_wt,
worktree=None,
process_project_root=self.install_root,
env={},
configured_canonical_root=self.target_root,
)
self.assertTrue(assessment["block"])
self.assertTrue(any("is not under" in r for r in assessment["reasons"]))
if __name__ == "__main__":
unittest.main()