Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
caaec9ac7a |
+4
-62
@@ -22,62 +22,8 @@ import gitea_config
|
|||||||
|
|
||||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||||
|
|
||||||
# Reserved runtime-control / workspace-binding environment variables (#704).
|
# Load standard .env if present
|
||||||
# Repository .env files MUST NOT populate or override any of these keys.
|
load_dotenv(os.path.join(PROJECT_ROOT, ".env"))
|
||||||
RESERVED_WORKTREE_ENV_KEYS: frozenset[str] = frozenset({
|
|
||||||
"GITEA_ACTIVE_WORKTREE",
|
|
||||||
"GITEA_AUTHOR_WORKTREE",
|
|
||||||
"GITEA_REVIEWER_WORKTREE",
|
|
||||||
"GITEA_MERGER_WORKTREE",
|
|
||||||
"GITEA_RECONCILER_WORKTREE",
|
|
||||||
})
|
|
||||||
|
|
||||||
|
|
||||||
def is_reserved_worktree_env_key(key: str | None) -> bool:
|
|
||||||
"""Return True if *key* is a reserved runtime workspace binding variable (#704)."""
|
|
||||||
if not key:
|
|
||||||
return False
|
|
||||||
k = str(key).upper().strip()
|
|
||||||
return k in RESERVED_WORKTREE_ENV_KEYS or (k.startswith("GITEA_") and k.endswith("_WORKTREE"))
|
|
||||||
|
|
||||||
|
|
||||||
def load_env_file_sanitized(
|
|
||||||
env_path: str,
|
|
||||||
*,
|
|
||||||
target_env: dict | os._Environ | None = None,
|
|
||||||
) -> list[str]:
|
|
||||||
"""Load a .env file without populating or overriding reserved workspace keys (#704).
|
|
||||||
|
|
||||||
Pre-existing process environment values retain their precedence. Reserved
|
|
||||||
runtime-control keys found in repository files are ignored (without logging
|
|
||||||
their values). Returns a list of sanitized rejection reasons.
|
|
||||||
"""
|
|
||||||
if target_env is None:
|
|
||||||
target_env = os.environ
|
|
||||||
if not os.path.exists(env_path) or os.path.isdir(env_path):
|
|
||||||
return []
|
|
||||||
|
|
||||||
rejection_reasons: list[str] = []
|
|
||||||
try:
|
|
||||||
file_vals = dotenv_values(env_path)
|
|
||||||
for key, val in file_vals.items():
|
|
||||||
if not key or val is None:
|
|
||||||
continue
|
|
||||||
if is_reserved_worktree_env_key(key):
|
|
||||||
filename = os.path.basename(env_path)
|
|
||||||
rejection_reasons.append(
|
|
||||||
f"Ignored reserved runtime workspace key '{key}' from repository {filename}"
|
|
||||||
)
|
|
||||||
continue
|
|
||||||
if key not in target_env:
|
|
||||||
target_env[key] = val
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
return rejection_reasons
|
|
||||||
|
|
||||||
|
|
||||||
# Load standard .env if present (sanitized to prevent repo workspace binding contamination #704)
|
|
||||||
load_env_file_sanitized(os.path.join(PROJECT_ROOT, ".env"))
|
|
||||||
|
|
||||||
# Dictionary to store configurations parsed dynamically from .env.* files
|
# Dictionary to store configurations parsed dynamically from .env.* files
|
||||||
DYNAMIC_CONFIGS = {}
|
DYNAMIC_CONFIGS = {}
|
||||||
@@ -91,13 +37,9 @@ for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
|
|||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
config_vals = dotenv_values(env_path)
|
config_vals = dotenv_values(env_path)
|
||||||
# Filter out reserved workspace keys from dynamic configs (#704)
|
site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
|
||||||
sanitized_config = {
|
|
||||||
k: v for k, v in config_vals.items() if not is_reserved_worktree_env_key(k)
|
|
||||||
}
|
|
||||||
site = sanitized_config.get("GITEA_SITE") or sanitized_config.get("GITEA_HOST")
|
|
||||||
if site:
|
if site:
|
||||||
DYNAMIC_CONFIGS[site.lower().strip()] = sanitized_config
|
DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
+28
-11
@@ -3325,10 +3325,16 @@ def _effective_remote(remote: str) -> str:
|
|||||||
return remote
|
return remote
|
||||||
|
|
||||||
|
|
||||||
def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
|
def _resolve(
|
||||||
|
remote: str,
|
||||||
|
host: str | None,
|
||||||
|
org: str | None,
|
||||||
|
repo: str | None,
|
||||||
|
for_mutation: bool = False,
|
||||||
|
):
|
||||||
"""Resolve remote + overrides to (host, org, repo).
|
"""Resolve remote + overrides to (host, org, repo).
|
||||||
|
|
||||||
#714 / #530: when the caller omits org and/or repo, prefer the
|
#714 / #530 / #707: when the caller omits org and/or repo, prefer the
|
||||||
workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare
|
workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare
|
||||||
``remote=prgs`` must not force ``Timesheet`` when the checkout is
|
``remote=prgs`` must not force ``Timesheet`` when the checkout is
|
||||||
``Gitea-Tools``). Explicit caller org/repo always win and are validated
|
``Gitea-Tools``). Explicit caller org/repo always win and are validated
|
||||||
@@ -3414,6 +3420,7 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
|
|||||||
# Workspace-filled sides are intentional alignment for #530.
|
# Workspace-filled sides are intentional alignment for #530.
|
||||||
org_explicit=org_explicit or filled_org,
|
org_explicit=org_explicit or filled_org,
|
||||||
repo_explicit=repo_explicit or filled_repo,
|
repo_explicit=repo_explicit or filled_repo,
|
||||||
|
for_mutation=for_mutation,
|
||||||
)
|
)
|
||||||
return (resolved_host, resolved_org, resolved_repo)
|
return (resolved_host, resolved_org, resolved_repo)
|
||||||
|
|
||||||
@@ -3475,8 +3482,9 @@ def _enforce_remote_repo_guard(
|
|||||||
*,
|
*,
|
||||||
org_explicit: bool,
|
org_explicit: bool,
|
||||||
repo_explicit: bool,
|
repo_explicit: bool,
|
||||||
|
for_mutation: bool = False,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530).
|
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530/#707).
|
||||||
|
|
||||||
Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
|
Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
|
||||||
set, so the unit suite (which calls tools with bare remotes against mocked APIs)
|
set, so the unit suite (which calls tools with bare remotes against mocked APIs)
|
||||||
@@ -3488,6 +3496,12 @@ def _enforce_remote_repo_guard(
|
|||||||
):
|
):
|
||||||
return
|
return
|
||||||
local_remote_url = _local_git_remote_url(remote)
|
local_remote_url = _local_git_remote_url(remote)
|
||||||
|
primary_org = None
|
||||||
|
primary_repo = None
|
||||||
|
ctx = session_ctx.get_session_context()
|
||||||
|
if ctx:
|
||||||
|
primary_org = ctx.get("org")
|
||||||
|
primary_repo = ctx.get("repository")
|
||||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||||
remote=remote,
|
remote=remote,
|
||||||
resolved_org=resolved_org,
|
resolved_org=resolved_org,
|
||||||
@@ -3495,6 +3509,9 @@ def _enforce_remote_repo_guard(
|
|||||||
local_remote_url=local_remote_url,
|
local_remote_url=local_remote_url,
|
||||||
org_explicit=org_explicit,
|
org_explicit=org_explicit,
|
||||||
repo_explicit=repo_explicit,
|
repo_explicit=repo_explicit,
|
||||||
|
for_mutation=for_mutation,
|
||||||
|
primary_org=primary_org,
|
||||||
|
primary_repo=primary_repo,
|
||||||
)
|
)
|
||||||
if assessment["block"]:
|
if assessment["block"]:
|
||||||
raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
|
raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
|
||||||
@@ -4063,7 +4080,7 @@ def gitea_lock_issue(
|
|||||||
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
|
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
|
||||||
worktree_path, _canonical_local_git_root()
|
worktree_path, _canonical_local_git_root()
|
||||||
)
|
)
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
existing_issue_lock = _load_existing_issue_lock(
|
existing_issue_lock = _load_existing_issue_lock(
|
||||||
remote=remote, org=o, repo=r, issue_number=issue_number
|
remote=remote, org=o, repo=r, issue_number=issue_number
|
||||||
)
|
)
|
||||||
@@ -5239,7 +5256,7 @@ def gitea_create_pr(
|
|||||||
org=org,
|
org=org,
|
||||||
repo=repo,
|
repo=repo,
|
||||||
)
|
)
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
|
|
||||||
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
|
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
|
||||||
lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
|
lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
|
||||||
@@ -9560,7 +9577,7 @@ def gitea_edit_pr(
|
|||||||
required_permission="gitea.pr.close",
|
required_permission="gitea.pr.close",
|
||||||
)
|
)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
|
url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
|
||||||
|
|
||||||
@@ -9844,7 +9861,7 @@ def gitea_commit_files(
|
|||||||
verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
|
verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
|
||||||
processed_files, source_proofs = _prepare_commit_payload_files(files)
|
processed_files, source_proofs = _prepare_commit_payload_files(files)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/contents"
|
url = f"{repo_api_url(h, o, r)}/contents"
|
||||||
|
|
||||||
@@ -9980,7 +9997,7 @@ def gitea_publish_unpublished_issue_branch(
|
|||||||
repo=repo,
|
repo=repo,
|
||||||
)
|
)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
git_remote = (git_remote_name or remote or "").strip()
|
git_remote = (git_remote_name or remote or "").strip()
|
||||||
|
|
||||||
existing_lock = issue_lock_store.load_issue_lock(
|
existing_lock = issue_lock_store.load_issue_lock(
|
||||||
@@ -10183,7 +10200,7 @@ def gitea_bootstrap_author_issue_worktree(
|
|||||||
repo=repo,
|
repo=repo,
|
||||||
)
|
)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
canonical_root = _canonical_local_git_root()
|
canonical_root = _canonical_local_git_root()
|
||||||
|
|
||||||
import author_issue_bootstrap
|
import author_issue_bootstrap
|
||||||
@@ -12131,7 +12148,7 @@ def gitea_reconcile_merged_cleanups(
|
|||||||
"blocker_kind": "invalid_pr_number",
|
"blocker_kind": "invalid_pr_number",
|
||||||
}
|
}
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
base = repo_api_url(h, o, r)
|
base = repo_api_url(h, o, r)
|
||||||
|
|
||||||
@@ -12521,7 +12538,7 @@ def gitea_assess_already_landed_reconciliation(
|
|||||||
"permission_report": _permission_block_report("gitea.read"),
|
"permission_report": _permission_block_report("gitea.read"),
|
||||||
}
|
}
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
|
pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
|
||||||
|
|
||||||
|
|||||||
+60
-7
@@ -54,20 +54,62 @@ def assess_remote_repo_match(
|
|||||||
local_remote_url: str | None,
|
local_remote_url: str | None,
|
||||||
org_explicit: bool,
|
org_explicit: bool,
|
||||||
repo_explicit: bool,
|
repo_explicit: bool,
|
||||||
|
for_mutation: bool = False,
|
||||||
|
primary_org: str | None = None,
|
||||||
|
primary_repo: str | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Fail closed when the resolved org/repo disagrees with the local git remote.
|
"""Fail closed when the resolved org/repo disagrees with the local git remote.
|
||||||
|
|
||||||
The guard is intentionally conservative:
|
The guard enforces two protection levels:
|
||||||
|
|
||||||
* When the caller passed both ``org`` and ``repo`` explicitly, their intent is
|
1. Cross-Project Mutation Boundary (#707):
|
||||||
authoritative and the guard never blocks.
|
When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR,
|
||||||
* When the local git remote URL is unavailable (``None``/empty), corroboration
|
merge, branch deletion), the target repository (``resolved_org/resolved_repo``)
|
||||||
is impossible, so the guard does not block (best-effort only).
|
MUST match the primary authorized project context (``primary_org/primary_repo``
|
||||||
* Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL
|
or parsed from ``local_remote_url``). Any attempt to mutate a different project
|
||||||
(case-insensitive); if it does not, the guard blocks.
|
fails closed, even if explicit org/repo were passed. Metadata operations (such
|
||||||
|
as creating issues or commenting on issues) across projects remain allowed.
|
||||||
|
|
||||||
|
2. Workspace Mismatch Guard (#530):
|
||||||
|
When ``for_mutation`` is False (or targets match), bare remotes must resolve
|
||||||
|
to an org/repo slug present in the local git remote URL. When explicit org/repo
|
||||||
|
are supplied for non-mutation operations, the caller's intent is authoritative.
|
||||||
"""
|
"""
|
||||||
reasons: list[str] = []
|
reasons: list[str] = []
|
||||||
|
|
||||||
|
eff_primary_org = primary_org
|
||||||
|
eff_primary_repo = primary_repo
|
||||||
|
if not eff_primary_org or not eff_primary_repo:
|
||||||
|
parsed_primary = parse_org_repo_from_remote_url(local_remote_url)
|
||||||
|
if parsed_primary:
|
||||||
|
eff_primary_org = eff_primary_org or parsed_primary[0]
|
||||||
|
eff_primary_repo = eff_primary_repo or parsed_primary[1]
|
||||||
|
|
||||||
|
# #707: Enforce cross-project codebase mutation boundary if for_mutation is True
|
||||||
|
if for_mutation and eff_primary_org and eff_primary_repo:
|
||||||
|
if (
|
||||||
|
resolved_org.lower() != eff_primary_org.lower()
|
||||||
|
or resolved_repo.lower() != eff_primary_repo.lower()
|
||||||
|
):
|
||||||
|
reasons.append(
|
||||||
|
f"Cross-project mutation guard (#707): Attempted codebase mutation targeting "
|
||||||
|
f"'{resolved_org}/{resolved_repo}' outside of primary authorized project "
|
||||||
|
f"context '{eff_primary_org}/{eff_primary_repo}'"
|
||||||
|
)
|
||||||
|
return {
|
||||||
|
"proven": False,
|
||||||
|
"block": True,
|
||||||
|
"cross_project_mutation_block": True,
|
||||||
|
"reasons": reasons,
|
||||||
|
"remote": remote,
|
||||||
|
"resolved_org": resolved_org,
|
||||||
|
"resolved_repo": resolved_repo,
|
||||||
|
"primary_org": eff_primary_org,
|
||||||
|
"primary_repo": eff_primary_repo,
|
||||||
|
"local_remote_url": local_remote_url,
|
||||||
|
"remediation": f"Cross-project codebase work is forbidden. Create an issue in the target repository ('{resolved_org}/{resolved_repo}') instead.",
|
||||||
|
}
|
||||||
|
|
||||||
if org_explicit and repo_explicit:
|
if org_explicit and repo_explicit:
|
||||||
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
|
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
|
||||||
|
|
||||||
@@ -88,6 +130,16 @@ def assess_remote_repo_match(
|
|||||||
|
|
||||||
def format_remote_repo_guard_error(assessment: dict) -> str:
|
def format_remote_repo_guard_error(assessment: dict) -> str:
|
||||||
"""Single RuntimeError message for the MCP resolver gate."""
|
"""Single RuntimeError message for the MCP resolver gate."""
|
||||||
|
if assessment.get("cross_project_mutation_block"):
|
||||||
|
resolved = f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
|
||||||
|
primary = f"{assessment.get('primary_org')}/{assessment.get('primary_repo')}"
|
||||||
|
return (
|
||||||
|
f"Cross-project mutation guard (#707): Attempted codebase mutation targeting '{resolved}' "
|
||||||
|
f"outside of primary authorized project context '{primary}'. "
|
||||||
|
f"Cross-project codebase work (creating branches, committing files, creating PRs) is forbidden; "
|
||||||
|
f"create an issue in the target repository ('{resolved}') instead."
|
||||||
|
)
|
||||||
|
|
||||||
reasons = "; ".join(
|
reasons = "; ".join(
|
||||||
assessment.get("reasons") or ["remote/repo resolution mismatch"]
|
assessment.get("reasons") or ["remote/repo resolution mismatch"]
|
||||||
)
|
)
|
||||||
@@ -120,3 +172,4 @@ def _assessment(
|
|||||||
"local_remote_url": local_remote_url,
|
"local_remote_url": local_remote_url,
|
||||||
"remediation": REMEDIATION,
|
"remediation": REMEDIATION,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,94 @@
|
|||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import unittest
|
||||||
|
from unittest import mock
|
||||||
|
|
||||||
|
import remote_repo_guard
|
||||||
|
import gitea_mcp_server as server
|
||||||
|
from session_context_binding import _reset_session_context_for_testing, bind_session_context
|
||||||
|
|
||||||
|
|
||||||
|
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||||
|
|
||||||
|
|
||||||
|
class TestCrossProjectMutationBoundary(unittest.TestCase):
|
||||||
|
def setUp(self):
|
||||||
|
os.environ["PYTEST_CURRENT_TEST"] = "test"
|
||||||
|
_reset_session_context_for_testing()
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
_reset_session_context_for_testing()
|
||||||
|
os.environ.pop("PYTEST_CURRENT_TEST", None)
|
||||||
|
|
||||||
|
def test_cross_project_codebase_mutation_blocked(self):
|
||||||
|
"""Codebase mutations targeting another project must fail closed (#707)."""
|
||||||
|
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||||
|
remote="prgs",
|
||||||
|
resolved_org="Other-Org",
|
||||||
|
resolved_repo="Other-Repo",
|
||||||
|
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||||
|
org_explicit=True,
|
||||||
|
repo_explicit=True,
|
||||||
|
for_mutation=True,
|
||||||
|
)
|
||||||
|
self.assertTrue(assessment["block"])
|
||||||
|
self.assertTrue(assessment.get("cross_project_mutation_block"))
|
||||||
|
self.assertEqual(assessment["primary_org"], "Scaled-Tech-Consulting")
|
||||||
|
self.assertEqual(assessment["primary_repo"], "Gitea-Tools")
|
||||||
|
|
||||||
|
err_msg = remote_repo_guard.format_remote_repo_guard_error(assessment)
|
||||||
|
self.assertIn("Cross-project mutation guard (#707)", err_msg)
|
||||||
|
self.assertIn("Attempted codebase mutation targeting 'Other-Org/Other-Repo'", err_msg)
|
||||||
|
self.assertIn("outside of primary authorized project context 'Scaled-Tech-Consulting/Gitea-Tools'", err_msg)
|
||||||
|
self.assertIn("create an issue in the target repository ('Other-Org/Other-Repo') instead", err_msg)
|
||||||
|
|
||||||
|
def test_same_project_codebase_mutation_allowed(self):
|
||||||
|
"""Codebase mutations targeting the primary project context are allowed."""
|
||||||
|
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||||
|
remote="prgs",
|
||||||
|
resolved_org="Scaled-Tech-Consulting",
|
||||||
|
resolved_repo="Gitea-Tools",
|
||||||
|
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||||
|
org_explicit=True,
|
||||||
|
repo_explicit=True,
|
||||||
|
for_mutation=True,
|
||||||
|
)
|
||||||
|
self.assertFalse(assessment["block"])
|
||||||
|
self.assertFalse(assessment.get("cross_project_mutation_block", False))
|
||||||
|
|
||||||
|
def test_cross_project_metadata_operation_allowed(self):
|
||||||
|
"""Metadata operations (issue creation, comments) across project boundaries are allowed."""
|
||||||
|
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||||
|
remote="prgs",
|
||||||
|
resolved_org="Other-Org",
|
||||||
|
resolved_repo="Other-Repo",
|
||||||
|
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||||
|
org_explicit=True,
|
||||||
|
repo_explicit=True,
|
||||||
|
for_mutation=False,
|
||||||
|
)
|
||||||
|
self.assertTrue(assessment["proven"])
|
||||||
|
self.assertFalse(assessment["block"])
|
||||||
|
|
||||||
|
@mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
|
||||||
|
@mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
|
||||||
|
def test_mcp_server_resolve_cross_project_mutation_fails_closed(self, mock_remote):
|
||||||
|
"""_resolve with for_mutation=True blocks cross-project targets."""
|
||||||
|
with self.assertRaises(RuntimeError) as ctx:
|
||||||
|
server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=True)
|
||||||
|
|
||||||
|
err = str(ctx.exception)
|
||||||
|
self.assertIn("Cross-project mutation guard (#707)", err)
|
||||||
|
self.assertIn("create an issue in the target repository", err)
|
||||||
|
|
||||||
|
@mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
|
||||||
|
@mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
|
||||||
|
def test_mcp_server_resolve_cross_project_metadata_succeeds(self, mock_remote):
|
||||||
|
"""_resolve with for_mutation=False allows explicit cross-project target."""
|
||||||
|
host, org, repo = server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=False)
|
||||||
|
self.assertEqual(org, "Other-Org")
|
||||||
|
self.assertEqual(repo, "Other-Repo")
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -1,166 +0,0 @@
|
|||||||
"""Tests for Issue #704: Preventing repository .env files from injecting workspace bindings.
|
|
||||||
|
|
||||||
Acceptance Criteria (#704):
|
|
||||||
1. Repository .env loading cannot populate or override GITEA_ACTIVE_WORKTREE or any role-specific GITEA_*_WORKTREE runtime-binding variable.
|
|
||||||
2. Runtime workspace bindings are accepted only from sanctioned managed-launch/session mechanisms.
|
|
||||||
3. Pre-existing sanctioned process environment values retain their intended precedence.
|
|
||||||
4. Importing gitea_auth or related modules does not mutate workspace-binding state from repository files.
|
|
||||||
5. Reserved runtime-control keys found in .env are ignored or rejected with a sanitized actionable reason; their values are never logged.
|
|
||||||
6. The protection applies consistently to author, reviewer, merger, and reconciler namespaces.
|
|
||||||
7. Comprehensive test coverage for stale worktree, missing worktree, task-specific, role-specific, launcher binding, repeated imports, namespace isolation, precedence, and absence of secret leakage.
|
|
||||||
8. Dirty-state and workspace-preflight gates cannot be bypassed by an injected missing-path binding.
|
|
||||||
9. No environment, dotenv, offline-import, or caller-controlled path can forge native transport or mutation provenance.
|
|
||||||
10. Cross-linked with #702, PR #703, #510.
|
|
||||||
11. Required immediate follow-up to Issue #702 / PR #703.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import tempfile
|
|
||||||
import importlib
|
|
||||||
import unittest
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
||||||
|
|
||||||
import gitea_auth
|
|
||||||
from gitea_auth import is_reserved_worktree_env_key, load_env_file_sanitized
|
|
||||||
|
|
||||||
|
|
||||||
class TestIssue704PreventEnvWorkspaceBindings(unittest.TestCase):
|
|
||||||
"""Test suite verifying .env workspace-binding injection prevention (#704)."""
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.tmpdir = tempfile.TemporaryDirectory()
|
|
||||||
self.addCleanup(self.tmpdir.cleanup)
|
|
||||||
self.env_dir = Path(self.tmpdir.name)
|
|
||||||
|
|
||||||
def test_is_reserved_worktree_env_key(self):
|
|
||||||
"""Verify key classification for all role namespaces (#704 AC6)."""
|
|
||||||
reserved_keys = [
|
|
||||||
"GITEA_ACTIVE_WORKTREE",
|
|
||||||
"GITEA_AUTHOR_WORKTREE",
|
|
||||||
"GITEA_REVIEWER_WORKTREE",
|
|
||||||
"GITEA_MERGER_WORKTREE",
|
|
||||||
"GITEA_RECONCILER_WORKTREE",
|
|
||||||
"gitea_active_worktree",
|
|
||||||
"GITEA_CUSTOM_ROLE_WORKTREE",
|
|
||||||
]
|
|
||||||
for key in reserved_keys:
|
|
||||||
self.assertTrue(
|
|
||||||
is_reserved_worktree_env_key(key),
|
|
||||||
f"Expected {key} to be recognized as a reserved worktree key",
|
|
||||||
)
|
|
||||||
|
|
||||||
unreserved_keys = [
|
|
||||||
"GITEA_USER",
|
|
||||||
"GITEA_PASS",
|
|
||||||
"GITEA_TOKEN",
|
|
||||||
"GITEA_HOST",
|
|
||||||
"PATH",
|
|
||||||
]
|
|
||||||
for key in unreserved_keys:
|
|
||||||
self.assertFalse(
|
|
||||||
is_reserved_worktree_env_key(key),
|
|
||||||
f"Expected {key} to NOT be recognized as a reserved worktree key",
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_load_env_file_sanitized_ignores_reserved_keys(self):
|
|
||||||
"""Verify .env loading ignores GITEA_ACTIVE_WORKTREE and role-specific keys (#704 AC1)."""
|
|
||||||
env_file = self.env_dir / ".env"
|
|
||||||
stale_path = "/tmp/stale-worktree-path-1234"
|
|
||||||
env_file.write_text(
|
|
||||||
f"GITEA_USER=testuser\n"
|
|
||||||
f"GITEA_ACTIVE_WORKTREE={stale_path}\n"
|
|
||||||
f"GITEA_AUTHOR_WORKTREE={stale_path}\n"
|
|
||||||
f"GITEA_REVIEWER_WORKTREE={stale_path}\n"
|
|
||||||
f"GITEA_MERGER_WORKTREE={stale_path}\n"
|
|
||||||
f"GITEA_RECONCILER_WORKTREE={stale_path}\n"
|
|
||||||
)
|
|
||||||
|
|
||||||
test_env = {}
|
|
||||||
reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
|
|
||||||
|
|
||||||
# Unreserved key loaded
|
|
||||||
self.assertEqual(test_env.get("GITEA_USER"), "testuser")
|
|
||||||
|
|
||||||
# Reserved keys ignored
|
|
||||||
self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
|
|
||||||
self.assertNotIn("GITEA_AUTHOR_WORKTREE", test_env)
|
|
||||||
self.assertNotIn("GITEA_REVIEWER_WORKTREE", test_env)
|
|
||||||
self.assertNotIn("GITEA_MERGER_WORKTREE", test_env)
|
|
||||||
self.assertNotIn("GITEA_RECONCILER_WORKTREE", test_env)
|
|
||||||
|
|
||||||
# Rejection reasons populated without leaking the secret value (#704 AC5)
|
|
||||||
self.assertTrue(len(reasons) >= 5)
|
|
||||||
for r in reasons:
|
|
||||||
self.assertNotIn(stale_path, r, "Secret/path value must not leak into rejection reason")
|
|
||||||
|
|
||||||
def test_preexisting_sanctioned_launcher_env_retained(self):
|
|
||||||
"""Sanctioned launcher values in process env are retained (#704 AC2, AC3)."""
|
|
||||||
sanctioned_path = "/tmp/sanctioned-launcher-worktree"
|
|
||||||
test_env = {"GITEA_ACTIVE_WORKTREE": sanctioned_path}
|
|
||||||
|
|
||||||
env_file = self.env_dir / ".env"
|
|
||||||
env_file.write_text("GITEA_ACTIVE_WORKTREE=/tmp/injected-repo-worktree\n")
|
|
||||||
|
|
||||||
load_env_file_sanitized(str(env_file), target_env=test_env)
|
|
||||||
|
|
||||||
# Pre-existing value retained, not overwritten by .env
|
|
||||||
self.assertEqual(test_env.get("GITEA_ACTIVE_WORKTREE"), sanctioned_path)
|
|
||||||
|
|
||||||
def test_stale_or_missing_worktree_in_env_ignored(self):
|
|
||||||
"""Stale or non-existent worktree path in .env file is ignored (#704 AC7)."""
|
|
||||||
nonexistent_path = "/nonexistent/branches/stale-issue-999"
|
|
||||||
env_file = self.env_dir / ".env"
|
|
||||||
env_file.write_text(f"GITEA_ACTIVE_WORKTREE={nonexistent_path}\n")
|
|
||||||
|
|
||||||
test_env = {}
|
|
||||||
load_env_file_sanitized(str(env_file), target_env=test_env)
|
|
||||||
|
|
||||||
self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
|
|
||||||
|
|
||||||
def test_repeated_module_import_does_not_mutate_workspace_env(self):
|
|
||||||
"""Repeated imports of gitea_auth leave os.environ un-contaminated (#704 AC4, AC7)."""
|
|
||||||
# Ensure no active worktree env exists initially
|
|
||||||
original_val = os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
|
|
||||||
try:
|
|
||||||
importlib.reload(gitea_auth)
|
|
||||||
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
|
|
||||||
|
|
||||||
importlib.reload(gitea_auth)
|
|
||||||
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
|
|
||||||
finally:
|
|
||||||
if original_val is not None:
|
|
||||||
os.environ["GITEA_ACTIVE_WORKTREE"] = original_val
|
|
||||||
|
|
||||||
def test_namespace_isolation_all_roles_protected(self):
|
|
||||||
"""Verify protection across author, reviewer, merger, reconciler (#704 AC6)."""
|
|
||||||
env_file = self.env_dir / ".env"
|
|
||||||
env_file.write_text(
|
|
||||||
"GITEA_AUTHOR_WORKTREE=/bad/author\n"
|
|
||||||
"GITEA_REVIEWER_WORKTREE=/bad/reviewer\n"
|
|
||||||
"GITEA_MERGER_WORKTREE=/bad/merger\n"
|
|
||||||
"GITEA_RECONCILER_WORKTREE=/bad/reconciler\n"
|
|
||||||
)
|
|
||||||
test_env = {}
|
|
||||||
load_env_file_sanitized(str(env_file), target_env=test_env)
|
|
||||||
|
|
||||||
self.assertEqual(test_env, {})
|
|
||||||
|
|
||||||
def test_absence_of_secret_leakage(self):
|
|
||||||
"""Rejection reasons contain key names but never secret path values (#704 AC5)."""
|
|
||||||
sensitive_path = "/Users/secret/path/private_repo"
|
|
||||||
env_file = self.env_dir / ".env"
|
|
||||||
env_file.write_text(f"GITEA_ACTIVE_WORKTREE={sensitive_path}\n")
|
|
||||||
|
|
||||||
test_env = {}
|
|
||||||
reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
|
|
||||||
for reason in reasons:
|
|
||||||
self.assertNotIn(sensitive_path, reason)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user