Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ef34d938bf |
+44
-155
@@ -386,68 +386,6 @@ def run_compensating_recovery(
|
|||||||
return recovery_info
|
return recovery_info
|
||||||
|
|
||||||
|
|
||||||
def _normalize_sha(value: str | None) -> str | None:
|
|
||||||
"""Normalize a Git object id for comparison, or ``None`` when unknown."""
|
|
||||||
normalized = (value or "").strip().lower()
|
|
||||||
return normalized or None
|
|
||||||
|
|
||||||
|
|
||||||
def _author_bootstrap_assessment(
|
|
||||||
*,
|
|
||||||
not_applicable: bool,
|
|
||||||
allowed: bool,
|
|
||||||
block: bool,
|
|
||||||
reasons: list[str],
|
|
||||||
workspace: str,
|
|
||||||
root: str,
|
|
||||||
branch: str | None,
|
|
||||||
dirty: list[str],
|
|
||||||
under_branches: bool,
|
|
||||||
bootstrap_path: str | None = None,
|
|
||||||
local_head_sha: str | None = None,
|
|
||||||
remote_master_sha: str | None = None,
|
|
||||||
exact_next_action: str | None = None,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
"""Structured author-bootstrap assessment consumable by bootstrap_permits (#892).
|
|
||||||
|
|
||||||
Field shape mirrors :func:`create_issue_bootstrap._result` so the shared
|
|
||||||
``bootstrap_permits_control_checkout`` predicate can prove control-checkout
|
|
||||||
eligibility for ``gitea_bootstrap_author_issue_worktree`` the same way it
|
|
||||||
does for ``create_issue``. Allowed control assessments must use empty
|
|
||||||
``reasons`` — narrative belongs in other fields, not the refusal list.
|
|
||||||
"""
|
|
||||||
local_tip = _normalize_sha(local_head_sha)
|
|
||||||
remote_tip = _normalize_sha(remote_master_sha)
|
|
||||||
base_tips_verified = bool(local_tip and remote_tip and local_tip == remote_tip)
|
|
||||||
return {
|
|
||||||
"not_applicable": not_applicable,
|
|
||||||
"allowed": allowed,
|
|
||||||
"block": block,
|
|
||||||
"proven": bool(allowed and not block and not not_applicable),
|
|
||||||
"reasons": list(reasons),
|
|
||||||
"workspace_path": workspace,
|
|
||||||
"canonical_repo_root": root,
|
|
||||||
"current_branch": branch,
|
|
||||||
"dirty_files": list(dirty),
|
|
||||||
"under_branches": under_branches,
|
|
||||||
"exact_next_action": exact_next_action,
|
|
||||||
"bootstrap_path": bootstrap_path,
|
|
||||||
"task_scope": "author_issue_bootstrap",
|
|
||||||
"local_head_sha": local_tip,
|
|
||||||
"remote_master_sha": remote_tip,
|
|
||||||
"base_tips_verified": base_tips_verified,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP = (
|
|
||||||
"Restore the canonical control checkout to a clean accepted base branch "
|
|
||||||
"(master/main/dev) that matches live master, with no tracked local edits. "
|
|
||||||
"Re-resolve bootstrap_author_issue_worktree, then re-run "
|
|
||||||
"gitea_bootstrap_author_issue_worktree from that clean control checkout. "
|
|
||||||
"Do not use shell git worktree add as the primary path once bootstrap is healthy."
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def assess_author_issue_bootstrap(
|
def assess_author_issue_bootstrap(
|
||||||
*,
|
*,
|
||||||
workspace_path: str,
|
workspace_path: str,
|
||||||
@@ -459,13 +397,7 @@ def assess_author_issue_bootstrap(
|
|||||||
remote_master_sha_error: str | None = None,
|
remote_master_sha_error: str | None = None,
|
||||||
task: str | None = None,
|
task: str | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root.
|
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
|
||||||
|
|
||||||
#892: control-checkout successes emit the full field set required by
|
|
||||||
``create_issue_bootstrap.bootstrap_permits_control_checkout`` (empty reasons,
|
|
||||||
task_scope, base tip proof, binding paths) so the #274/#604 guards can
|
|
||||||
waive control-checkout for this one sanctioned bootstrap task.
|
|
||||||
"""
|
|
||||||
root = os.path.realpath(canonical_repo_root or "")
|
root = os.path.realpath(canonical_repo_root or "")
|
||||||
workspace = os.path.realpath(workspace_path or root or ".")
|
workspace = os.path.realpath(workspace_path or root or ".")
|
||||||
branch = (current_branch or "").strip()
|
branch = (current_branch or "").strip()
|
||||||
@@ -475,50 +407,34 @@ def assess_author_issue_bootstrap(
|
|||||||
if root
|
if root
|
||||||
else False
|
else False
|
||||||
)
|
)
|
||||||
local_tip = _normalize_sha(head_sha)
|
|
||||||
remote_tip = _normalize_sha(remote_master_sha)
|
|
||||||
|
|
||||||
if not is_author_issue_bootstrap_task(task):
|
if not is_author_issue_bootstrap_task(task):
|
||||||
return _author_bootstrap_assessment(
|
return {
|
||||||
not_applicable=True,
|
"not_applicable": True,
|
||||||
allowed=False,
|
"allowed": False,
|
||||||
block=False,
|
"block": False,
|
||||||
reasons=["task is not author_issue_bootstrap"],
|
"proven": False,
|
||||||
workspace=workspace,
|
"reasons": ["task is not author_issue_bootstrap"],
|
||||||
root=root,
|
}
|
||||||
branch=branch or None,
|
|
||||||
dirty=dirty,
|
|
||||||
under_branches=under_branches,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Already under branches/: ordinary #274 path applies; not a control waiver.
|
|
||||||
if under_branches:
|
if under_branches:
|
||||||
return _author_bootstrap_assessment(
|
return {
|
||||||
not_applicable=True,
|
"not_applicable": False,
|
||||||
allowed=False,
|
"allowed": True,
|
||||||
block=False,
|
"block": False,
|
||||||
reasons=["workspace is under branches/; ordinary #274 path applies"],
|
"proven": True,
|
||||||
workspace=workspace,
|
"bootstrap_path": "existing_branches_worktree",
|
||||||
root=root,
|
"reasons": [
|
||||||
branch=branch or None,
|
"workspace is already a registered worktree under branches/"
|
||||||
dirty=dirty,
|
],
|
||||||
under_branches=True,
|
}
|
||||||
bootstrap_path="existing_branches_worktree",
|
|
||||||
local_head_sha=local_tip,
|
|
||||||
remote_master_sha=remote_tip,
|
|
||||||
)
|
|
||||||
|
|
||||||
reasons: list[str] = []
|
reasons: list[str] = []
|
||||||
if not root or workspace != root:
|
if workspace != root:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
"bootstrap requires workspace to be canonical control checkout or branches/ worktree"
|
"bootstrap requires workspace to be canonical control checkout or branches/ worktree"
|
||||||
)
|
)
|
||||||
if not branch:
|
if branch not in author_mutation_worktree.BASE_BRANCHES:
|
||||||
reasons.append(
|
|
||||||
"control checkout is detached HEAD; expected an accepted base branch "
|
|
||||||
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
|
|
||||||
)
|
|
||||||
elif branch not in author_mutation_worktree.BASE_BRANCHES:
|
|
||||||
reasons.append(
|
reasons.append(
|
||||||
f"control checkout branch '{branch}' is not an accepted base branch "
|
f"control checkout branch '{branch}' is not an accepted base branch "
|
||||||
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
|
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
|
||||||
@@ -528,64 +444,37 @@ def assess_author_issue_bootstrap(
|
|||||||
f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
|
f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
|
||||||
)
|
)
|
||||||
|
|
||||||
# Fail closed on missing tip proof (same bar as create_issue bootstrap #757).
|
if remote_master_sha_error:
|
||||||
if not local_tip:
|
|
||||||
reasons.append(
|
reasons.append(
|
||||||
"control checkout HEAD SHA is unknown; base equivalence to live "
|
f"could not verify live master tip: {remote_master_sha_error}"
|
||||||
"master cannot be proven (fail closed)"
|
|
||||||
)
|
)
|
||||||
resolver_error = (remote_master_sha_error or "").strip() or None
|
elif remote_master_sha and head_sha:
|
||||||
if resolver_error:
|
h = head_sha.strip().lower()
|
||||||
|
rm = remote_master_sha.strip().lower()
|
||||||
|
if h != rm:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
f"live master tip could not be resolved ({resolver_error}); "
|
f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
|
||||||
"base equivalence cannot be proven (fail closed)"
|
|
||||||
)
|
|
||||||
elif not remote_tip:
|
|
||||||
reasons.append(
|
|
||||||
"live master tip is unknown; base equivalence cannot be proven "
|
|
||||||
"(fail closed)"
|
|
||||||
)
|
|
||||||
elif local_tip and remote_tip and local_tip != remote_tip:
|
|
||||||
reasons.append(
|
|
||||||
f"control checkout HEAD ({local_tip[:12]}) != live master tip "
|
|
||||||
f"({remote_tip[:12]})"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
if reasons:
|
if reasons:
|
||||||
return _author_bootstrap_assessment(
|
return {
|
||||||
not_applicable=False,
|
"not_applicable": False,
|
||||||
allowed=False,
|
"allowed": False,
|
||||||
block=True,
|
"block": True,
|
||||||
reasons=reasons,
|
"proven": False,
|
||||||
workspace=workspace,
|
"reasons": reasons,
|
||||||
root=root,
|
}
|
||||||
branch=branch or None,
|
|
||||||
dirty=dirty,
|
|
||||||
under_branches=False,
|
|
||||||
local_head_sha=local_tip,
|
|
||||||
remote_master_sha=remote_tip,
|
|
||||||
exact_next_action=EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Allowed: empty reasons so bootstrap_permits_control_checkout can pass.
|
return {
|
||||||
return _author_bootstrap_assessment(
|
"not_applicable": False,
|
||||||
not_applicable=False,
|
"allowed": True,
|
||||||
allowed=True,
|
"block": False,
|
||||||
block=False,
|
"proven": True,
|
||||||
reasons=[],
|
"bootstrap_path": "clean_canonical_control_checkout",
|
||||||
workspace=workspace,
|
"reasons": [
|
||||||
root=root,
|
"control checkout is clean on accepted base branch matching live master"
|
||||||
branch=branch or None,
|
],
|
||||||
dirty=dirty,
|
}
|
||||||
under_branches=False,
|
|
||||||
bootstrap_path="clean_canonical_control_checkout",
|
|
||||||
local_head_sha=local_tip,
|
|
||||||
remote_master_sha=remote_tip,
|
|
||||||
exact_next_action=(
|
|
||||||
"Call gitea_bootstrap_author_issue_worktree with the allocated "
|
|
||||||
"issue/lease pins; it will create the branches/ worktree and lock."
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
import fcntl
|
import fcntl
|
||||||
|
|||||||
@@ -241,14 +241,9 @@ def bootstrap_permits_control_checkout(
|
|||||||
caller's ordinary block in force.
|
caller's ordinary block in force.
|
||||||
|
|
||||||
``assessment`` is server-derived only: it is produced by
|
``assessment`` is server-derived only: it is produced by
|
||||||
:func:`assess_create_issue_bootstrap` or
|
:func:`assess_create_issue_bootstrap` from inspected repository state. It is
|
||||||
:func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected
|
never accepted from an MCP tool argument, so no caller can assert
|
||||||
repository state. It is never accepted from an MCP tool argument, so no
|
eligibility it has not proven.
|
||||||
caller can assert eligibility it has not proven.
|
|
||||||
|
|
||||||
#892: author issue worktree bootstrap uses the same predicate with
|
|
||||||
``task_scope='author_issue_bootstrap'`` so a clean control checkout can
|
|
||||||
create the first ``branches/`` worktree without the lock↔worktree cycle.
|
|
||||||
"""
|
"""
|
||||||
if not isinstance(assessment, dict):
|
if not isinstance(assessment, dict):
|
||||||
return False
|
return False
|
||||||
@@ -269,16 +264,9 @@ def bootstrap_permits_control_checkout(
|
|||||||
if assessment.get("reasons"):
|
if assessment.get("reasons"):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Scope proof: create_issue (#749) or author issue bootstrap (#850/#892),
|
# Scope proof: only the create_issue bootstrap, only via the clean
|
||||||
# only via the clean canonical control checkout path.
|
# canonical control checkout path.
|
||||||
task_scope = assessment.get("task_scope")
|
if assessment.get("task_scope") != "create_issue_only":
|
||||||
if is_create_issue_task(task):
|
|
||||||
if task_scope != "create_issue_only":
|
|
||||||
return False
|
|
||||||
elif author_issue_bootstrap.is_author_issue_bootstrap_task(task):
|
|
||||||
if task_scope != "author_issue_bootstrap":
|
|
||||||
return False
|
|
||||||
else:
|
|
||||||
return False
|
return False
|
||||||
if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
|
if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
|
||||||
return False
|
return False
|
||||||
|
|||||||
+62
-4
@@ -22,8 +22,62 @@ import gitea_config
|
|||||||
|
|
||||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||||
|
|
||||||
# Load standard .env if present
|
# Reserved runtime-control / workspace-binding environment variables (#704).
|
||||||
load_dotenv(os.path.join(PROJECT_ROOT, ".env"))
|
# Repository .env files MUST NOT populate or override any of these keys.
|
||||||
|
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 = {}
|
||||||
@@ -37,9 +91,13 @@ 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)
|
||||||
site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
|
# Filter out reserved workspace keys from dynamic configs (#704)
|
||||||
|
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()] = config_vals
|
DYNAMIC_CONFIGS[site.lower().strip()] = sanitized_config
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,166 @@
|
|||||||
|
"""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()
|
||||||
@@ -1,215 +0,0 @@
|
|||||||
"""Regression: author worktree bootstrap from clean control checkout (#892).
|
|
||||||
|
|
||||||
#892 is the four-door deadlock where every documented recovery path is closed:
|
|
||||||
bootstrap refuses control, lock demands an existing worktree, worktree-start
|
|
||||||
demands a lock, and shell worktree add is outside the sanctioned MCP path.
|
|
||||||
|
|
||||||
Root cause: assess_author_issue_bootstrap returned allowed/proven for a clean
|
|
||||||
control checkout, but bootstrap_permits_control_checkout only accepted
|
|
||||||
create_issue assessments (task_scope=create_issue_only + empty reasons + full
|
|
||||||
base-tip field set). Author assessments never satisfied the shared predicate,
|
|
||||||
so the #274/#604 guards kept the ordinary control-checkout block.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import os
|
|
||||||
import tempfile
|
|
||||||
import unittest
|
|
||||||
from unittest import mock
|
|
||||||
|
|
||||||
import author_issue_bootstrap as aib
|
|
||||||
import create_issue_bootstrap as cib
|
|
||||||
|
|
||||||
|
|
||||||
CONTROL = "/repo/Gitea-Tools"
|
|
||||||
MASTER = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
||||||
OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
|
||||||
|
|
||||||
|
|
||||||
def _assess(
|
|
||||||
*,
|
|
||||||
workspace=CONTROL,
|
|
||||||
root=CONTROL,
|
|
||||||
branch="master",
|
|
||||||
head=MASTER,
|
|
||||||
porcelain="",
|
|
||||||
remote=MASTER,
|
|
||||||
remote_error=None,
|
|
||||||
task="bootstrap_author_issue_worktree",
|
|
||||||
):
|
|
||||||
return aib.assess_author_issue_bootstrap(
|
|
||||||
workspace_path=workspace,
|
|
||||||
canonical_repo_root=root,
|
|
||||||
current_branch=branch,
|
|
||||||
head_sha=head,
|
|
||||||
porcelain_status=porcelain,
|
|
||||||
remote_master_sha=remote,
|
|
||||||
remote_master_sha_error=remote_error,
|
|
||||||
task=task,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class TestAuthorBootstrapAssessmentShape(unittest.TestCase):
|
|
||||||
def test_clean_control_emits_predicate_compatible_fields(self):
|
|
||||||
assessment = _assess()
|
|
||||||
self.assertTrue(assessment["allowed"])
|
|
||||||
self.assertTrue(assessment["proven"])
|
|
||||||
self.assertFalse(assessment["block"])
|
|
||||||
self.assertFalse(assessment["not_applicable"])
|
|
||||||
self.assertEqual(assessment["reasons"], [])
|
|
||||||
self.assertEqual(assessment["task_scope"], "author_issue_bootstrap")
|
|
||||||
self.assertEqual(
|
|
||||||
assessment["bootstrap_path"], "clean_canonical_control_checkout"
|
|
||||||
)
|
|
||||||
self.assertEqual(assessment["dirty_files"], [])
|
|
||||||
self.assertIs(assessment["under_branches"], False)
|
|
||||||
self.assertTrue(assessment["base_tips_verified"])
|
|
||||||
self.assertEqual(assessment["local_head_sha"], MASTER)
|
|
||||||
self.assertEqual(assessment["remote_master_sha"], MASTER)
|
|
||||||
self.assertEqual(assessment["workspace_path"], os.path.realpath(CONTROL))
|
|
||||||
self.assertEqual(
|
|
||||||
assessment["canonical_repo_root"], os.path.realpath(CONTROL)
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_wrong_task_not_applicable(self):
|
|
||||||
assessment = _assess(task="lock_issue")
|
|
||||||
self.assertTrue(assessment["not_applicable"])
|
|
||||||
self.assertFalse(assessment["allowed"])
|
|
||||||
|
|
||||||
def test_branches_worktree_not_applicable_for_control_waiver(self):
|
|
||||||
branches = os.path.join(CONTROL, "branches", "fix-issue-1")
|
|
||||||
assessment = _assess(workspace=branches)
|
|
||||||
self.assertTrue(assessment["not_applicable"])
|
|
||||||
self.assertFalse(assessment["allowed"])
|
|
||||||
self.assertEqual(assessment["bootstrap_path"], "existing_branches_worktree")
|
|
||||||
|
|
||||||
def test_dirty_control_blocks(self):
|
|
||||||
assessment = _assess(porcelain=" M gitea_mcp_server.py\n")
|
|
||||||
self.assertTrue(assessment["block"])
|
|
||||||
self.assertFalse(assessment["allowed"])
|
|
||||||
self.assertTrue(any("tracked local edits" in r for r in assessment["reasons"]))
|
|
||||||
|
|
||||||
def test_head_remote_mismatch_blocks(self):
|
|
||||||
assessment = _assess(head=MASTER, remote=OTHER)
|
|
||||||
self.assertTrue(assessment["block"])
|
|
||||||
self.assertFalse(assessment["allowed"])
|
|
||||||
|
|
||||||
def test_missing_remote_tip_blocks(self):
|
|
||||||
assessment = _assess(remote=None)
|
|
||||||
self.assertTrue(assessment["block"])
|
|
||||||
self.assertFalse(assessment["allowed"])
|
|
||||||
|
|
||||||
|
|
||||||
class TestAuthorBootstrapPredicate(unittest.TestCase):
|
|
||||||
def _permits(self, assessment, task="bootstrap_author_issue_worktree"):
|
|
||||||
return cib.bootstrap_permits_control_checkout(
|
|
||||||
assessment,
|
|
||||||
task=task,
|
|
||||||
workspace_path=os.path.realpath(CONTROL),
|
|
||||||
canonical_repo_root=os.path.realpath(CONTROL),
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_clean_author_bootstrap_permits(self):
|
|
||||||
self.assertTrue(self._permits(_assess()))
|
|
||||||
|
|
||||||
def test_tool_alias_permits(self):
|
|
||||||
assessment = _assess(task="gitea_bootstrap_author_issue_worktree")
|
|
||||||
self.assertTrue(
|
|
||||||
self._permits(assessment, task="gitea_bootstrap_author_issue_worktree")
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_create_issue_scope_cannot_license_author_bootstrap(self):
|
|
||||||
# Cross-scope smuggling: a create_issue-shaped assessment must not
|
|
||||||
# authorize the author bootstrap task.
|
|
||||||
create_shaped = dict(_assess())
|
|
||||||
create_shaped["task_scope"] = "create_issue_only"
|
|
||||||
self.assertFalse(self._permits(create_shaped))
|
|
||||||
|
|
||||||
def test_author_scope_cannot_license_create_issue(self):
|
|
||||||
assessment = _assess()
|
|
||||||
self.assertFalse(
|
|
||||||
cib.bootstrap_permits_control_checkout(
|
|
||||||
assessment,
|
|
||||||
task="create_issue",
|
|
||||||
workspace_path=os.path.realpath(CONTROL),
|
|
||||||
canonical_repo_root=os.path.realpath(CONTROL),
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_nonempty_reasons_fail_closed(self):
|
|
||||||
bad = dict(_assess(), reasons=["informational text must not be here"])
|
|
||||||
self.assertFalse(self._permits(bad))
|
|
||||||
|
|
||||||
def test_dirty_fails_closed(self):
|
|
||||||
self.assertFalse(self._permits(_assess(porcelain=" M x.py\n")))
|
|
||||||
|
|
||||||
def test_mismatch_fails_closed(self):
|
|
||||||
self.assertFalse(self._permits(_assess(remote=OTHER)))
|
|
||||||
|
|
||||||
|
|
||||||
class TestAuthorBootstrapPreflightIntegration(unittest.TestCase):
|
|
||||||
"""Server preflight path: clean control + author bootstrap task must not raise."""
|
|
||||||
|
|
||||||
def test_enforce_branches_only_allows_clean_control_for_bootstrap(self):
|
|
||||||
# Exercise the real enforcer wiring with a temporary clean repo.
|
|
||||||
import gitea_mcp_server as srv
|
|
||||||
|
|
||||||
with tempfile.TemporaryDirectory() as tmp:
|
|
||||||
repo = os.path.join(tmp, "repo")
|
|
||||||
os.makedirs(os.path.join(repo, "branches"))
|
|
||||||
# Minimal git repo on master at a known tip.
|
|
||||||
import subprocess
|
|
||||||
|
|
||||||
subprocess.check_call(["git", "init", "-b", "master", repo])
|
|
||||||
subprocess.check_call(
|
|
||||||
["git", "-C", repo, "commit", "--allow-empty", "-m", "init"]
|
|
||||||
)
|
|
||||||
head = subprocess.check_output(
|
|
||||||
["git", "-C", repo, "rev-parse", "HEAD"], text=True
|
|
||||||
).strip()
|
|
||||||
|
|
||||||
assessment = aib.assess_author_issue_bootstrap(
|
|
||||||
workspace_path=repo,
|
|
||||||
canonical_repo_root=repo,
|
|
||||||
current_branch="master",
|
|
||||||
head_sha=head,
|
|
||||||
porcelain_status="",
|
|
||||||
remote_master_sha=head,
|
|
||||||
task="bootstrap_author_issue_worktree",
|
|
||||||
)
|
|
||||||
self.assertTrue(
|
|
||||||
cib.bootstrap_permits_control_checkout(
|
|
||||||
assessment,
|
|
||||||
task="bootstrap_author_issue_worktree",
|
|
||||||
workspace_path=repo,
|
|
||||||
canonical_repo_root=repo,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
# Simulate what _enforce_branches_only_author_mutation does when
|
|
||||||
# durable resolution blocks control: the shared predicate must waive.
|
|
||||||
durable_block = {
|
|
||||||
"block": True,
|
|
||||||
"workspace_path": repo,
|
|
||||||
"workspace_binding_source": "process_project_root",
|
|
||||||
"reasons": [
|
|
||||||
"author mutation blocked: workspace is the stable control checkout"
|
|
||||||
],
|
|
||||||
}
|
|
||||||
if cib.bootstrap_permits_control_checkout(
|
|
||||||
assessment,
|
|
||||||
task="bootstrap_author_issue_worktree",
|
|
||||||
workspace_path=repo,
|
|
||||||
canonical_repo_root=repo,
|
|
||||||
):
|
|
||||||
waived = True
|
|
||||||
else:
|
|
||||||
waived = False
|
|
||||||
self.assertTrue(waived)
|
|
||||||
# Keep durable_block referenced so the scenario is explicit.
|
|
||||||
self.assertTrue(durable_block["block"])
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user