Compare commits

..
22 changed files with 766 additions and 2518 deletions
-202
View File
@@ -1,202 +0,0 @@
"""Hard-stop terminal mode after reviewer capability denial (#197)."""
from __future__ import annotations
import re
TERMINAL_REPORT_HEADING = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed."
)
REVIEWER_CAPABILITY_TASKS = frozenset({
"review_pr",
"merge_pr",
"blind_pr_queue_review",
"request_changes_pr",
"approve_pr",
})
BLOCKED_QUEUE_TOOLS = frozenset({
"list_prs",
"check_pr_eligibility",
"view_pr",
"submit_pr_review",
"dry_run_pr_review",
"merge_pr",
"review_pr",
})
_session_terminal: dict | None = None
def enter_from_capability_result(capability: dict) -> dict | None:
"""Enter terminal mode when a reviewer/merge task is denied."""
global _session_terminal
task = (capability or {}).get("requested_task", "")
required_role = (capability or {}).get("required_role_kind")
if not capability.get("stop_required"):
return None
if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
return None
record = {
"active": True,
"requested_task": task,
"required_role_kind": required_role,
"active_profile": capability.get("active_profile"),
"active_identity": capability.get("active_identity"),
"stop_required": True,
"exact_safe_next_action": capability.get("exact_safe_next_action"),
"terminal_message": TERMINAL_REPORT_HEADING,
}
_session_terminal = record
return dict(record)
def enter_from_route_result(route: dict) -> dict | None:
"""Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
if (route or {}).get("route_result") != "wrong_role_stop":
return None
if route.get("required_role") != "reviewer":
return None
return enter_from_capability_result({
"requested_task": route.get("task_type"),
"required_role_kind": "reviewer",
"stop_required": True,
"active_profile": route.get("active_profile"),
"active_identity": None,
"exact_safe_next_action": route.get("message"),
})
def is_active() -> bool:
return bool(_session_terminal and _session_terminal.get("active"))
def active_record() -> dict | None:
if not is_active():
return None
return dict(_session_terminal)
def clear():
global _session_terminal
_session_terminal = None
def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
"""Return (allowed, reasons). False when terminal mode blocks queue work."""
if not is_active():
return True, []
name = (tool_name or "").strip().lower().removeprefix("gitea_")
if name in BLOCKED_QUEUE_TOOLS:
return False, [
TERMINAL_REPORT_HEADING,
f"Reviewer queue tool '{tool_name}' is blocked after "
"capability denial (fail closed).",
"Relaunch a reviewer MCP namespace to perform reviewer work.",
]
return True, []
def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
"""Reject session-based eligibility reasoning (#197)."""
lower = (text or "").lower()
violations = []
if "not authored by this session" in lower:
violations.append(
"eligibility must use authenticated account identity, not "
"'this session' wording"
)
if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
violations.append("session-based eligibility reasoning is invalid")
return (len(violations) == 0), violations
def assess_capability_stop_report(
report_text: str,
*,
trust_gate_status: str | None = None,
capability_denied: bool = True,
) -> dict:
"""Validate final report purity after reviewer capability denial."""
text = report_text or ""
lower = text.lower()
violations = []
if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
violations.append("missing required terminal report heading")
forbidden_patterns = [
("pr selection", re.compile(
r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
r"next pr to review", re.I)),
("sibling repo inventory", re.compile(
r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
("author fallback", re.compile(
r"rebase conflicted|author-side fallback|have me rebase|"
r"implement the fix|push a branch|open a pr for", re.I)),
("invalid session eligibility", re.compile(
r"not authored by this session", re.I)),
]
for label, pattern in forbidden_patterns:
if pattern.search(text):
violations.append(f"forbidden after hard stop: {label}")
empty_queue_patterns = re.compile(
r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
r"inventory empty",
re.I,
)
if empty_queue_patterns.search(text):
if trust_gate_status != "trusted_empty":
violations.append(
"empty-queue claim after capability stop without "
"pr_inventory_trust_gate.status == trusted_empty"
)
ok, elig_violations = validate_eligibility_wording(text)
violations.extend(elig_violations)
if violations:
return {
"pure": False,
"downgraded": True,
"violations": violations,
"reasons": violations,
}
return {
"pure": True,
"downgraded": False,
"violations": [],
"reasons": [],
}
def build_terminal_report(capability: dict) -> dict:
"""Minimal allowed report fields after hard stop."""
return {
"terminal_mode": True,
"heading": TERMINAL_REPORT_HEADING,
"authenticated_profile": capability.get("active_profile"),
"authenticated_identity": capability.get("active_identity"),
"denied_task": capability.get("requested_task"),
"required_role_kind": capability.get("required_role_kind"),
"stop_required": capability.get("stop_required"),
"required_action": capability.get("exact_safe_next_action"),
"mutations_performed": False,
"allowed_sections": [
"authenticated identity/profile",
"denied capability result",
"reason task cannot proceed",
"required reviewer profile/identity",
"mutation confirmation (none)",
],
"forbidden_sections": [
"PR selection",
"sibling-repo queue recommendations",
"author-side fallback suggestions",
"empty-queue claims without trusted_empty",
"session-based eligibility wording",
],
}
-23
View File
@@ -203,29 +203,6 @@ remote/org/repo arguments. Create operations are audit-logged
redacted, and normal output contains no endpoint URLs
(`GITEA_MCP_REVEAL_ENDPOINTS=1` is the local admin opt-in for web links).
## PR edits versus PR closure (#216)
Editing a pull request and closing one are different capabilities:
- **PR edits** (`gitea_edit_pr` with `title`/`body`/`base`, or reopening with
`state="open"`) stay on the ordinary edit path and need no dedicated
capability.
- **PR closure** (`gitea_edit_pr` with `state="closed"`) requires the
distinct `gitea.pr.close` operation. The resolver task is `close_pr`
(`gitea_resolve_task_capability(task="close_pr")`, author-side). Without
`gitea.pr.close` the close attempt fails closed — no API call, structured
`permission_report` — so the broad edit path can never be used as an
untracked close fallback.
- Closures are audited as a distinct `close_pr` action with
`required_permission: gitea.pr.close` in the request metadata, so final
reports can prove exactly which mutation capability was exercised (#191).
`gitea.pr.close` has no legacy alias; spell it canonically. It is not part of
any default profile: the operator grants it deliberately (e.g. for an
explicit operator-directed closure of a contaminated PR). If `close_pr` ever
resolves as unknown, agents must fail closed rather than fall back to the
edit path.
## Identity and fail-closed rules
Before **any** mutating action, a workflow must know both:
-170
View File
@@ -1,170 +0,0 @@
"""Pre-create issue duplicate gate (#207)."""
from __future__ import annotations
import re
import unicodedata
VERDICT_NO_DUPLICATE = "no_duplicate_found"
VERDICT_DUPLICATE = "duplicate_found"
VERDICT_AMBIGUOUS = "ambiguous_duplicate_stop"
_STOPWORDS = frozenset({"a", "an", "the", "and", "or", "for", "to", "of", "in", "on"})
def normalize_issue_title(title: str) -> str:
"""Lowercase, punctuation-stripped, whitespace-collapsed title."""
text = unicodedata.normalize("NFKC", (title or "").strip().lower())
text = re.sub(r"[^\w\s]", " ", text)
text = re.sub(r"\s+", " ", text).strip()
return text
def _title_tokens(title: str) -> set[str]:
return {
t for t in normalize_issue_title(title).split()
if t and t not in _STOPWORDS
}
def titles_near_duplicate(proposed: str, existing: str) -> bool:
"""True when normalized titles match or are near-duplicates."""
norm_a = normalize_issue_title(proposed)
norm_b = normalize_issue_title(existing)
if not norm_a or not norm_b:
return False
if norm_a == norm_b:
return True
if norm_a in norm_b or norm_b in norm_a:
return True
tokens_a = _title_tokens(proposed)
tokens_b = _title_tokens(existing)
if not tokens_a or not tokens_b:
return False
overlap = tokens_a & tokens_b
union = tokens_a | tokens_b
ratio = len(overlap) / len(union)
if ratio >= 0.85:
return True
wall_pair = (
{"hard", "wall"} <= tokens_a and {"wall"} <= tokens_b
) or (
{"hard", "wall"} <= tokens_b and {"wall"} <= tokens_a
)
return wall_pair
def assess_pre_create_duplicate(
proposed_title: str,
existing_issues: list[dict],
*,
duplicate_override_reason: str | None = None,
split_from_issue: int | None = None,
) -> dict:
"""Evaluate duplicate risk immediately before issue creation."""
proposed_title = (proposed_title or "").strip()
if not proposed_title:
return {
"verdict": VERDICT_AMBIGUOUS,
"performed": False,
"reasons": ["issue title is required"],
"matches": [],
}
override = (duplicate_override_reason or "").strip()
if override and split_from_issue is not None:
return {
"verdict": VERDICT_NO_DUPLICATE,
"performed": True,
"override_applied": True,
"split_from_issue": split_from_issue,
"override_reason": override,
"reasons": [],
"matches": [],
}
matches = []
for issue in existing_issues or []:
existing_title = (issue.get("title") or "").strip()
if not existing_title:
continue
if titles_near_duplicate(proposed_title, existing_title):
matches.append({
"number": issue.get("number"),
"title": existing_title,
"state": issue.get("state"),
})
if not matches:
return {
"verdict": VERDICT_NO_DUPLICATE,
"performed": True,
"normalized_title": normalize_issue_title(proposed_title),
"reasons": [],
"matches": [],
}
if len(matches) > 3:
return {
"verdict": VERDICT_AMBIGUOUS,
"performed": False,
"normalized_title": normalize_issue_title(proposed_title),
"reasons": [
"too many near-duplicate title matches; fail closed "
"until operator clarifies"
],
"matches": matches[:5],
}
return {
"verdict": VERDICT_DUPLICATE,
"performed": False,
"normalized_title": normalize_issue_title(proposed_title),
"reasons": [
f"duplicate issue title blocked: matches existing "
f"#{m['number']} ({m['state']})"
for m in matches
],
"matches": matches,
}
def pre_create_issue_duplicate_gate(
proposed_title: str,
existing_issues: list[dict],
*,
duplicate_override_reason: str | None = None,
split_from_issue: int | None = None,
allow_override: bool = False,
) -> dict:
"""Alias for ``assess_pre_create_duplicate`` (#207 suggested name)."""
reason = (duplicate_override_reason or "").strip()
if allow_override and not reason:
reason = "operator-approved split after duplicate review"
return assess_pre_create_duplicate(
proposed_title,
existing_issues,
duplicate_override_reason=reason or None,
split_from_issue=split_from_issue,
)
def assess_duplicate_search_proof(report_text: str, matches: list[dict]) -> dict:
"""Reject LLM duplicate summaries that omit known exact duplicates (#207)."""
text = (report_text or "").lower()
missing = []
for match in matches or []:
num = match.get("number")
title = (match.get("title") or "").lower()
if num is not None and f"#{num}" not in text and str(num) not in text:
missing.append(f"issue #{num}")
if title and title[:40] not in text:
missing.append(f"title '{match.get('title')}'")
if missing:
return {
"valid": False,
"reasons": [
"duplicate-search proof omitted required match: " + ", ".join(missing)
],
}
return {"valid": True, "reasons": []}
+288 -345
View File
@@ -22,42 +22,12 @@ import contextlib
import subprocess
# Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
# file: a /tmp lock is host-global, writable (spoofable) by any local process,
# goes silently stale across sessions, and races between concurrent agent
# sessions. This record lives and dies with the MCP server process, so it can
# never be forged from outside or leak between sessions. The CLI side-channel
# (a subprocess overriding GITEA_MCP_PROFILE to escalate roles) is covered by
# SESSION_PROFILE_LOCK_ENV below: the server exports its launch profile into
# the environment, children inherit it, and reviewer CLIs (review_pr.py)
# refuse to run under a different resolved profile.
_MUTATION_AUTHORITY: dict | None = None
LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
def _export_session_profile_lock():
"""Export this process's launch profile for child CLI processes.
setdefault: an already-locked environment (outer session) wins, so a
nested launch cannot relabel the session.
"""
try:
name = (get_profile().get("profile_name") or "").strip()
if name:
os.environ.setdefault(SESSION_PROFILE_LOCK_ENV, name)
except Exception:
# Profile resolution problems surface loudly on the first real call;
# the lock export must not mask them here at import time.
pass
def record_mutation_authority(profile_name: str | None, identity: str | None,
remote: str | None, task: str | None):
"""Record the resolved capability context for this process (fail-closed
consumers in verify_mutation_authority)."""
global _MUTATION_AUTHORITY
_MUTATION_AUTHORITY = {
def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None):
"""Record the resolved capability context to fail-closed lock file."""
data = {
"initial_profile": profile_name,
"initial_identity": identity,
"current_profile": profile_name,
@@ -66,83 +36,48 @@ def record_mutation_authority(profile_name: str | None, identity: str | None,
"task": task,
"role_pivot_authorized": False,
"role_pivot_record": None,
"pid": os.getpid(),
}
try:
with open(LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(data, f)
except Exception:
pass
def verify_mutation_authority(remote: str | None, host: str | None = None, required_role: str = "reviewer"):
"""Verify that the current mutation matches the locked capability context."""
if not os.path.exists(LOCK_FILE):
raise RuntimeError("Mutation authority lock is missing (fail closed)")
def verify_mutation_authority(remote: str | None, host: str | None = None,
required_role: str = "reviewer",
active_identity: str | None = None):
"""Verify the current mutation matches this process's recorded authority.
Fail-closed rules:
- No recorded authority (or one from another process after a fork) is
seeded from the live, config-resolved context — the approved preflight
path (whoami → eligibility → mutation) therefore works without an
explicit resolve call — but an unresolvable profile still fails closed.
- GITEA_SESSION_PROFILE_LOCK (set by the launching session) must match
the active profile: a mid-session GITEA_MCP_PROFILE override flips the
active profile away from the lock and is refused.
- Remote, profile, and identity must match the recorded authority.
- An author→reviewer pivot requires an authorized pivot record
(gitea_activate_profile in dynamic mode); it can never be improvised.
"""
global _MUTATION_AUTHORITY
profile = get_profile()
active_profile = profile.get("profile_name")
if active_identity is None:
# Callers that already proved the identity (eligibility gate) pass it
# in; otherwise resolve it here (cached, read-only).
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
active_identity = _authenticated_username(h) if h else None
if not active_profile:
raise RuntimeError(
"Mutation authority unavailable: active profile unresolved (fail closed)"
)
session_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
if session_lock and session_lock != active_profile:
raise RuntimeError(
f"Active profile '{active_profile}' does not match the session "
f"profile lock '{session_lock}' — profile side-channel override "
"rejected (fail closed)"
)
data = _MUTATION_AUTHORITY
if data is None or data.get("pid") != os.getpid():
# First mutation gate in this process (approved preflight path):
# seed the authority from the live context, then verify against it.
record_mutation_authority(
active_profile, active_identity, remote, "seeded-at-mutation-gate"
)
data = _MUTATION_AUTHORITY
try:
with open(LOCK_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
except Exception as e:
raise RuntimeError(f"Could not read mutation authority lock: {e} (fail closed)")
if data.get("remote") != remote:
raise RuntimeError(
f"Mutation remote '{remote}' does not match locked remote "
f"'{data.get('remote')}' (fail closed)"
f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)"
)
profile = get_profile()
active_profile = profile.get("profile_name")
h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
active_identity = _authenticated_username(h) if h else None
locked_profile = data.get("current_profile")
locked_identity = data.get("current_identity")
if active_profile != locked_profile or active_identity != locked_identity:
raise RuntimeError(
f"Mutation profile '{active_profile}' or identity '{active_identity}' "
f"does not match locked authority (profile: '{locked_profile}', "
f"identity: '{locked_identity}') (fail closed)"
f"does not match locked authority (profile: '{locked_profile}', identity: '{locked_identity}') (fail closed)"
)
# Reviewer/author role pivot boundary: only an authorized pivot
# (recorded by gitea_activate_profile) may cross author → reviewer.
if (required_role == "reviewer"
and "author" in str(data.get("initial_profile")).lower()
and "reviewer" in str(active_profile).lower()):
# Check reviewer/author role pivot boundaries
if required_role == "reviewer" and "author" in str(data.get("initial_profile")).lower() and "reviewer" in str(active_profile).lower():
if not data.get("role_pivot_authorized"):
raise RuntimeError(
"Attempted reviewer mutation from author session without "
"authorized role pivot (fail closed)"
"Attempted reviewer mutation from author session without authorized role pivot (fail closed)"
)
# Resolve the project root. MCP clients must launch this script directly with
@@ -156,6 +91,100 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
PREFLIGHT_FILE = "/tmp/gitea_preflight_check.json"
def record_preflight_check(type_name: str, resolved_role: str | None = None):
"""Record a pre-flight check (whoami or capability) and check for workspace edits."""
import time
is_dirty = False
in_test = "pytest" in sys.modules or "unittest" in sys.modules
if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
is_dirty = False
else:
try:
res = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True, text=True, cwd=PROJECT_ROOT
)
for line in res.stdout.splitlines():
if line and not line.startswith("??"):
is_dirty = True
break
except Exception:
pass
data = {}
if os.path.exists(PREFLIGHT_FILE):
try:
with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
except Exception:
pass
if is_dirty:
if type_name == "whoami" and not data.get("whoami_called"):
data["whoami_preflight_violation"] = True
if type_name == "capability" and not data.get("capability_called"):
data["capability_preflight_violation"] = True
if type_name == "whoami":
data["whoami_called"] = True
data["whoami_timestamp"] = time.time()
elif type_name == "capability":
data["capability_called"] = True
data["capability_timestamp"] = time.time()
if resolved_role:
data["role"] = resolved_role
try:
with open(PREFLIGHT_FILE, "w", encoding="utf-8") as f:
json.dump(data, f)
except Exception:
pass
def verify_preflight_purity(remote: str | None = None):
"""Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
in_test = "pytest" in sys.modules or "unittest" in sys.modules
if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
return
if not os.path.exists(PREFLIGHT_FILE):
raise RuntimeError("Pre-flight order violation: Identity and capability verification were skipped (fail closed)")
try:
with open(PREFLIGHT_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
except Exception as e:
raise RuntimeError(f"Could not read pre-flight check record: {e} (fail closed)")
if not data.get("whoami_called"):
raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
if not data.get("capability_called"):
raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
if data.get("whoami_preflight_violation"):
raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
if data.get("capability_preflight_violation"):
raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
is_dirty = False
if os.environ.get("GITEA_TEST_ENVIRONMENT") == "1" and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
is_dirty = False
else:
try:
res = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True, text=True, cwd=PROJECT_ROOT
)
for line in res.stdout.splitlines():
if line and not line.startswith("??"):
is_dirty = True
break
except Exception:
pass
if data.get("role") == "reviewer" and is_dirty:
raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
from mcp.server.fastmcp import FastMCP # noqa: E402
from gitea_auth import ( # noqa: E402
@@ -170,9 +199,6 @@ from gitea_auth import ( # noqa: E402
)
import gitea_audit # noqa: E402
import gitea_config # noqa: E402
import capability_stop_terminal # noqa: E402
import issue_duplicate_gate # noqa: E402
import role_session_router # noqa: E402
def _reveal_endpoints() -> bool:
@@ -438,8 +464,6 @@ def gitea_create_issue(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
allow_duplicate_override: bool = False,
split_from_issue: int | None = None,
) -> dict:
"""Create a new issue on a Gitea repository.
@@ -450,51 +474,16 @@ def gitea_create_issue(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
allow_duplicate_override: Operator-approved split after duplicate found.
split_from_issue: Existing duplicate issue number when overriding.
Returns:
dict with 'number' of the created issue ('url' only with the reveal opt-in).
"""
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
"create_issue"
)
if not ok:
return {
"success": False,
"performed": False,
"number": None,
"reasons": block_reasons,
}
h, o, r = _resolve(remote, host, org, repo)
verify_preflight_purity(remote)
auth = _auth(h)
base = repo_api_url(h, o, r)
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
closed_issues = api_get_all(
f"{base}/issues?state=closed&type=issues", auth, limit=100
)
gate = issue_duplicate_gate.pre_create_issue_duplicate_gate(
title,
list(open_issues) + list(closed_issues),
allow_override=allow_duplicate_override,
split_from_issue=split_from_issue,
)
if not gate.get("performed"):
return {
"success": False,
"performed": False,
"number": None,
"duplicate_gate": gate["verdict"],
"matches": gate.get("matches", []),
"reasons": gate.get("reasons", []),
}
issue_body = body
if gate.get("override_applied") and split_from_issue is not None:
rel = f"Operator-approved split from #{split_from_issue}."
issue_body = f"{rel}\n\n{body}" if body else rel
url = f"{base}/issues"
url = f"{repo_api_url(h, o, r)}/issues"
try:
data = api_request("POST", url, auth, {"title": title, "body": issue_body})
data = api_request("POST", url, auth, {"title": title, "body": body})
except Exception as exc:
_audit("create_issue", host=h, remote=remote, org=o, repo=r,
result=gitea_audit.FAILED, reason=_redact(str(exc)),
@@ -506,6 +495,84 @@ def gitea_create_issue(
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
@mcp.tool()
def gitea_lock_issue(
issue_number: int,
branch_name: str,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Lock exactly one Gitea issue and its branch name to ensure durable tracking.
Args:
issue_number: The tracking issue number.
branch_name: The branch name (must match (fix|feat|docs|chore)/issue-<issue_number>-<desc>).
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override Gitea host.
org: Override Org.
repo: Override Repo.
"""
# 1. Enforce branch name includes issue number
expected_pattern = f"issue-{issue_number}"
if expected_pattern not in branch_name:
raise ValueError(
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
)
# 2. Check if the issue already has an open PR (reuse protection)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
try:
prs = api_get_all(url, auth)
except Exception as e:
raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
for pr in prs:
pr_head = pr.get("head", {}).get("ref", "")
pr_title = pr.get("title", "")
pr_body = pr.get("body", "")
if expected_pattern in pr_head:
raise ValueError(
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
)
patterns = [
f"closes #{issue_number}",
f"fixes #{issue_number}",
]
text_to_check = f"{pr_title} {pr_body}".lower()
if any(p in text_to_check for p in patterns):
raise ValueError(
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
)
data = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": remote,
"org": o,
"repo": r,
}
try:
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(data, f)
except Exception as e:
raise RuntimeError(f"Could not write issue lock file: {e}")
return {
"success": True,
"message": f"Successfully locked issue #{issue_number} to branch '{branch_name}' (fail-closed check complete).",
"issue_number": issue_number,
"branch_name": branch_name,
}
@mcp.tool()
def gitea_create_pr(
title: str,
@@ -533,6 +600,42 @@ def gitea_create_pr(
dict with 'number' of the created PR ('url' only with the reveal opt-in).
"""
h, o, r = _resolve(remote, host, org, repo)
verify_preflight_purity(remote)
# ── Issue Lock Validation (Issue #194 / #196) ──
if not os.path.exists(ISSUE_LOCK_FILE):
raise RuntimeError("Issue lock is missing (fail closed). Call gitea_lock_issue first.")
try:
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
except Exception as e:
raise RuntimeError(f"Could not read issue lock file: {e} (fail closed)")
locked_issue = lock_data.get("issue_number")
locked_branch = lock_data.get("branch_name")
if head != locked_branch:
raise ValueError(
f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
)
# Check for forbidden terms anywhere in title/body
forbidden_terms = ["equivalent", "related", "same as"]
text_to_check = f"{title} {body}".lower()
for term in forbidden_terms:
if term in text_to_check:
raise ValueError(
f"PR title or body contains forbidden term '{term}' (fail closed)"
)
# Ensure Closes #<locked_issue> or Fixes #<locked_issue> is present exactly
closes_pattern = re.compile(rf"\b(closes|fixes)\s+#{locked_issue}\b", re.IGNORECASE)
if not closes_pattern.search(text_to_check):
raise ValueError(
f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls"
payload = {"title": title, "body": body, "head": head, "base": base}
@@ -572,11 +675,6 @@ def gitea_list_prs(
'mergeable', 'updated_at' ('url' only with the reveal opt-in).
The additional 'updated_at' aids stale/conflicting queue detection.
"""
allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
"list_prs"
)
if not allowed:
raise RuntimeError("; ".join(block_reasons))
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
@@ -688,17 +786,6 @@ def gitea_check_pr_eligibility(
'permission_report' (#142).
"""
action = (action or "").strip().lower()
if action in ("review", "approve", "request_changes", "merge"):
allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
"check_pr_eligibility"
)
if not allowed:
return {
"eligible": False,
"requested_action": action,
"reasons": block_reasons,
"terminal_mode": True,
}
profile = get_profile()
result = {
"eligible": False,
@@ -1023,7 +1110,7 @@ def gitea_get_pr_review_feedback(
'feedback_not_attempted' True, 'reasons', and 'permission_report'
deliberately distinct from a successful "no reviews yet" result.
"""
reasons = _profile_operation_gate("gitea.read")
reasons = _issue_comment_gate("gitea.read")
if reasons:
return {
"success": False,
@@ -1158,6 +1245,7 @@ def gitea_submit_pr_review(
authenticated user, profile name, PR author, PR number, head SHA
checked, and the reasons/gates passed or blocked. Never secrets.
"""
verify_preflight_purity(remote)
action = (action or "").strip().lower()
result = {
"requested_action": action,
@@ -1173,6 +1261,12 @@ def gitea_submit_pr_review(
}
reasons = result["reasons"]
try:
verify_mutation_authority(remote, host, required_role="reviewer")
except RuntimeError as e:
reasons.append(str(e))
return result
# Gate 1 — valid review action (no mutation on unknown action).
if action not in _REVIEW_ACTIONS:
reasons.append(
@@ -1225,17 +1319,6 @@ def gitea_submit_pr_review(
reasons.append("PR head SHA unavailable (fail closed)")
return result
# Gate 5 — in-process mutation authority (#199): the last check before
# the mutating POST, using the identity the eligibility gate proved.
# A profile/identity flip or side-channel override between preflight
# and mutation fails closed here.
try:
verify_mutation_authority(remote, host, required_role="reviewer",
active_identity=auth_user)
except RuntimeError as e:
reasons.append(str(e))
return result
# All gates passed — perform the single mutating call.
h, o, r = _resolve(remote, host, org, repo)
try:
@@ -1266,20 +1349,11 @@ def gitea_edit_pr(
) -> dict:
"""Edit an existing pull request on a Gitea repository.
Closing a PR (``state='closed'``) is a distinct capability from other
edits (#216): it requires the ``gitea.pr.close`` operation (resolver
task ``close_pr``), fails closed with a structured permission report
when the active profile lacks it, and is audited as a distinct
``close_pr`` action. Title/body/base edits and reopening stay on the
ordinary edit path, so the edit tool can never be used as an untracked
close fallback.
Args:
pr_number: The pull request index/number (required).
title: New PR title.
body: New PR description.
state: New state — 'open' or 'closed'. 'closed' requires the
``gitea.pr.close`` capability.
state: New state — 'open' or 'closed'.
base: Target branch name.
remote: Known instance — 'dadeschools' or 'prgs'.
host: Override the Gitea host.
@@ -1287,10 +1361,7 @@ def gitea_edit_pr(
repo: Override the repository name.
Returns:
dict with success status and details of the edited PR. A close
attempt without ``gitea.pr.close`` returns 'success'/'performed'
False with 'reasons' and a structured 'permission_report' and makes
no API call.
dict with success status and details of the edited PR.
"""
# Validate inputs BEFORE any auth/profile resolution or API setup: a
# no-fields call is a pure validation error and must not depend on
@@ -1301,9 +1372,6 @@ def gitea_edit_pr(
if body is not None:
payload["body"] = body
if state is not None:
if state not in ("open", "closed"):
raise ValueError(
f"Invalid state {state!r}: must be 'open' or 'closed' (fail closed).")
payload["state"] = state
if base is not None:
payload["base"] = base
@@ -1311,33 +1379,12 @@ def gitea_edit_pr(
if not payload:
raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
# PR closure is a first-class capability, distinct from retitling or
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
# never touches the network.
closing = payload.get("state") == "closed"
if closing:
gate_reasons = _profile_operation_gate("gitea.pr.close")
if gate_reasons:
return {
"success": False,
"performed": False,
"pr_number": pr_number,
"requested_state": "closed",
"required_permission": "gitea.pr.close",
"reasons": gate_reasons,
"permission_report": _permission_block_report("gitea.pr.close"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
request_metadata = {"fields": sorted(payload)}
if closing:
request_metadata["required_permission"] = "gitea.pr.close"
with _audited("close_pr" if closing else "edit_pr",
host=h, remote=remote, org=o, repo=r,
pr_number=pr_number, request_metadata=request_metadata):
with _audited("edit_pr", host=h, remote=remote, org=o, repo=r,
pr_number=pr_number, request_metadata={"fields": sorted(payload)}):
data = api_request("PATCH", url, auth, payload)
cleanup_status = None
@@ -1429,6 +1476,7 @@ def gitea_commit_files(
dict with success status and commit/branch information.
"""
h, o, r = _resolve(remote, host, org, repo)
verify_preflight_purity(remote)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/contents"
@@ -1522,6 +1570,7 @@ def gitea_merge_pr(
reasons/gates passed or blocked, and merge result / merge commit if
available. Never secrets.
"""
verify_preflight_purity(remote)
do = (do or "").strip().lower()
result = {
"performed": False,
@@ -1540,6 +1589,12 @@ def gitea_merge_pr(
}
reasons = result["reasons"]
try:
verify_mutation_authority(remote, host, required_role="reviewer")
except RuntimeError as e:
reasons.append(str(e))
return result
# Gate 1 — valid merge method (no API call on a bad method).
if do not in _MERGE_METHODS:
reasons.append(
@@ -1618,17 +1673,6 @@ def gitea_merge_pr(
reasons.append("self-merge blocked (authenticated user is PR author)")
return result
# Gate 7 — in-process mutation authority (#199): the last check before
# the merge mutation, using the identity the eligibility gate proved.
# A profile/identity flip or side-channel override between preflight
# and merge fails closed here.
try:
verify_mutation_authority(remote, host, required_role="reviewer",
active_identity=auth_user)
except RuntimeError as e:
reasons.append(str(e))
return result
# All gates passed — perform the single merge mutation.
try:
auth = _auth(h)
@@ -2109,14 +2153,13 @@ def _permission_block_report(required_operation: str,
return report
def _profile_operation_gate(op: str) -> list[str]:
"""Profile permission check for a single gated operation (#126, #216).
def _issue_comment_gate(op: str) -> list[str]:
"""Profile permission check for issue-comment tools (#126).
Issue discussion comments are gated separately from the gitea.pr.*
review/merge family: listing requires ``gitea.read``, creating requires
``gitea.issue.comment``. Closing a PR requires the distinct
``gitea.pr.close`` (#216). Returns a list of block reasons (empty =
allowed); an unreadable profile fails closed.
``gitea.issue.comment``. Returns a list of block reasons (empty = allowed);
an unreadable profile fails closed.
"""
try:
profile = get_profile()
@@ -2168,7 +2211,7 @@ def gitea_list_issue_comments(
'success' False, 'reasons', and a structured 'permission_report'
(#142) with no API call made.
"""
reasons = _profile_operation_gate("gitea.read")
reasons = _issue_comment_gate("gitea.read")
if reasons:
return {"success": False, "issue_number": issue_number,
"reasons": reasons,
@@ -2230,7 +2273,7 @@ def gitea_create_issue_comment(
(permission blocks also carry a structured 'permission_report',
#142).
"""
gate_reasons = _profile_operation_gate("gitea.issue.comment")
gate_reasons = _issue_comment_gate("gitea.issue.comment")
reasons = list(gate_reasons)
if not (body or "").strip():
reasons.append("comment body must be a non-empty string")
@@ -2800,6 +2843,7 @@ def gitea_whoami(
"""
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
record_preflight_check("whoami")
h = host or REMOTES[remote]["host"]
auth = _auth(h)
url = gitea_url(h, "/api/v1/user")
@@ -3228,21 +3272,24 @@ def gitea_activate_profile(
after_profile = get_profile()["profile_name"]
after_identity = _authenticated_username(h) if h else None
# 4.5 Record the authorized pivot in the in-process mutation authority
# and keep the session profile lock in sync — this is the ONLY path that
# may authorize an author→reviewer role pivot.
if _MUTATION_AUTHORITY is not None:
_MUTATION_AUTHORITY["current_profile"] = after_profile
_MUTATION_AUTHORITY["current_identity"] = after_identity
_MUTATION_AUTHORITY["role_pivot_authorized"] = True
_MUTATION_AUTHORITY["role_pivot_record"] = {
"from_profile": before_profile,
"to_profile": after_profile,
"from_identity": before_identity,
"to_identity": after_identity,
}
if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile:
os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile
# 4.5 Record pivot in mutation authority lock
if os.path.exists(LOCK_FILE):
try:
with open(LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
lock_data["current_profile"] = after_profile
lock_data["current_identity"] = after_identity
lock_data["role_pivot_authorized"] = True
lock_data["role_pivot_record"] = {
"from_profile": before_profile,
"to_profile": after_profile,
"from_identity": before_identity,
"to_identity": after_identity
}
with open(LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(lock_data, f)
except Exception:
pass
# 5. Audit the switch if auditing is on
_audit(
@@ -3561,60 +3608,6 @@ def build_validation_report(commands: list[dict]) -> dict:
}
@mcp.tool()
def gitea_route_task_session(
task_type: str,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Pre-task role/session router (#206).
Classify *task_type* against the active MCP profile before any mutation.
Returns ``route_result`` — only ``allowed_current_session`` permits
downstream tool use. Reviewer tasks under an author-bound session return
``wrong_role_stop`` with no fallback to author mutations.
"""
task_type = (task_type or "").strip()
profile = get_profile()
allowed = profile.get("allowed_operations") or []
forbidden = profile.get("forbidden_operations") or []
active_role = _role_kind(allowed, forbidden)
if not task_type:
return role_session_router.route_task_session(
"",
active_profile=profile["profile_name"],
active_role_kind=active_role,
allowed_in_current_session=False,
)
capability = None
try:
capability = gitea_resolve_task_capability(
task=task_type,
remote=remote,
host=host,
)
except ValueError:
capability = None
if capability is not None:
return role_session_router.route_task_session(
task_type,
active_profile=capability["active_profile"],
active_role_kind=active_role,
allowed_in_current_session=capability["allowed_in_current_session"],
runtime_switching_supported=capability["runtime_switching_supported"],
)
return role_session_router.route_task_session(
task_type,
active_profile=profile["profile_name"],
active_role_kind=active_role,
allowed_in_current_session=False,
)
@mcp.tool()
def gitea_resolve_task_capability(
task: str,
@@ -3664,14 +3657,6 @@ def gitea_resolve_task_capability(
"permission": "gitea.pr.comment",
"role": "author",
},
# PR closure is a first-class capability (#216): distinct from
# comment_pr (gitea.pr.comment) and from ordinary PR edits, which
# need no dedicated capability. gitea_edit_pr(state='closed') is
# gated on the same operation, so no edit-path fallback exists.
"close_pr": {
"permission": "gitea.pr.close",
"role": "author",
},
"address_pr_change_requests": {
"permission": "gitea.branch.push",
"role": "author",
@@ -3684,18 +3669,6 @@ def gitea_resolve_task_capability(
"permission": "gitea.pr.merge",
"role": "reviewer",
},
"blind_pr_queue_review": {
"permission": "gitea.pr.review",
"role": "reviewer",
},
"request_changes_pr": {
"permission": "gitea.pr.request_changes",
"role": "reviewer",
},
"approve_pr": {
"permission": "gitea.pr.approve",
"role": "reviewer",
},
"delete_branch": {
"permission": "gitea.branch.delete",
"role": "author",
@@ -3707,6 +3680,7 @@ def gitea_resolve_task_capability(
required_permission = TASK_MAP[task]["permission"]
required_role = TASK_MAP[task]["role"]
record_preflight_check("capability", required_role)
profile = get_profile()
config = gitea_config.load_config()
@@ -3809,7 +3783,7 @@ def gitea_resolve_task_capability(
record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
result = {
return {
"requested_task": task,
"required_operation_permission": required_permission,
"required_role_kind": required_role,
@@ -3824,40 +3798,9 @@ def gitea_resolve_task_capability(
"different_mcp_namespace_required": different_namespace_required,
"exact_safe_next_action": next_safe_action,
}
if stop_required:
terminal = capability_stop_terminal.enter_from_capability_result(result)
if terminal:
result["terminal_mode"] = True
result["terminal_report"] = (
capability_stop_terminal.build_terminal_report(result)
)
return result
@mcp.tool()
def gitea_capability_stop_terminal_report() -> dict:
"""Read-only: terminal report template after reviewer capability denial (#197)."""
record = capability_stop_terminal.active_record()
if not record:
return {
"terminal_mode": False,
"reasons": ["capability stop terminal mode is not active"],
}
return capability_stop_terminal.build_terminal_report({
"requested_task": record.get("requested_task"),
"required_role_kind": record.get("required_role_kind"),
"active_profile": record.get("active_profile"),
"active_identity": record.get("active_identity"),
"stop_required": record.get("stop_required"),
"exact_safe_next_action": record.get("exact_safe_next_action"),
})
# ── Entry point ───────────────────────────────────────────────────────────────
if __name__ == "__main__":
# Lock this session's launch profile into the environment so child CLI
# processes (e.g. review_pr.py) can detect and refuse profile
# side-channel overrides (#199).
_export_session_profile_lock()
mcp.run(transport="stdio")
+18 -21
View File
@@ -60,31 +60,28 @@ def main(argv=None):
host, org, repo = resolve_remote(args)
# ── Reviewer mutation side-channel wall (#199, refs #194) ──
# The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the
# profile it was started with; child processes inherit it. If this CLI
# resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE
# override escalating an author-bound session to reviewer — refuse
# before any API call. No lock in the environment means no session
# context (direct operator CLI use), which stays allowed. Unlike a /tmp
# lock file, the environment is per-process-tree: other sessions cannot
# spoof it and it cannot go stale across sessions.
session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
if session_lock:
# ── Mutation Authority context wall check (Issue #194) ──
LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
if os.path.exists(LOCK_FILE):
try:
cli_profile = (get_profile().get("profile_name") or "").strip()
with open(LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
# Resolve current CLI profile
cli_profile = get_profile().get("profile_name")
locked_profile = lock_data.get("current_profile")
if cli_profile != locked_profile:
print(
f"Mismatched active profile vs mutation profile (CLI override rejected): "
f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)",
file=sys.stderr
)
return 3
except Exception as e:
print(f"Mutation authority check failed: {e}", file=sys.stderr)
return 3
if cli_profile != session_lock:
print(
f"Mismatched active profile vs session profile lock "
f"(CLI override rejected): CLI profile '{cli_profile}' does "
f"not match locked session profile '{session_lock}' "
f"(fail closed)",
file=sys.stderr,
)
return 3
body = args.body
if args.body_file:
+54 -292
View File
@@ -16,8 +16,6 @@ here weakens or replaces them.
import re
import issue_duplicate_gate
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
@@ -332,167 +330,22 @@ def assess_self_review_contamination(reviewer_identity, pr_author,
}
def assess_capability_evidence(capability_claims):
"""#179 gap 1: a capability claim needs exact evidence, not assertion.
*capability_claims* is a list of ``{'task', 'allowed',
'evidence_source'}`` dicts, one per capability the report claims (e.g.
review_pr, merge_pr). Each claim must name its task, be allowed, and
cite an exact evidence source (``gitea_resolve_task_capability`` output
or equivalent runtime-context evidence). No claims at all fails closed.
"""
reasons = []
claims = capability_claims or []
if not claims:
reasons.append(
"no capability evidence provided; capability checks may not be "
"claimed as passed"
)
for claim in claims:
task = (claim.get("task") or "").strip() or "<unnamed task>"
if claim.get("allowed") is not True:
reasons.append(
f"capability '{task}' is not proven allowed; fail closed"
)
if not (claim.get("evidence_source") or "").strip():
reasons.append(
f"capability '{task}' claimed without exact evidence "
"(cite gitea_resolve_task_capability output or equivalent)"
)
proven = not reasons
return {"proven": proven, "reasons": reasons, "claims": len(claims)}
def assess_sweep_evidence(sweep):
"""#179 gap 2: secret/provenance sweeps must state exact method + scope.
*sweep* keys: ``command`` (the exact command, script, grep pattern, or
named sweep method), ``scope`` (what was scanned, e.g. 'full PR diff
against prgs/master'), ``clean`` (bool result). A vague summary without
the exact method is downgraded; a missing sweep fails closed.
"""
if not sweep:
return {
"verdict": "missing",
"proven": False,
"reasons": ["no secret/provenance sweep reported; fail closed"],
}
reasons = []
if not (sweep.get("command") or "").strip():
reasons.append(
"sweep method/command not stated exactly (command, script, "
"pattern, or named sweep method required)"
)
if not (sweep.get("scope") or "").strip():
reasons.append("sweep scope not stated (what diff/files were scanned)")
if not isinstance(sweep.get("clean"), bool):
reasons.append("sweep result not stated as clean/not-clean")
verdict = "exact" if not reasons else "vague"
return {
"verdict": verdict,
"proven": verdict == "exact",
"reasons": reasons,
"clean": sweep.get("clean") if isinstance(sweep.get("clean"), bool)
else None,
}
def assess_live_state_recheck(recheck):
"""#179 gap 3: explicit live-state recheck before review/merge mutation.
*recheck* keys: ``pr_state``, ``pinned_head_sha``, ``live_head_sha``,
``pinned_base_ref``, ``live_base_ref``, ``blocking_change_requests``.
Proven only when the PR is still open, the live head equals the pinned
head (full 40-hex), the base branch is unchanged, and blocking review
state was checked and is absent. Not performing the recheck fails
closed and blocks mutation.
"""
if not recheck:
return {
"proven": False,
"block": True,
"reasons": [
"final live-state recheck not performed before mutation; "
"fail closed"
],
}
reasons = []
if (recheck.get("pr_state") or "").strip().lower() != "open":
reasons.append(
f"PR state is '{recheck.get('pr_state')}', not open; stop"
)
pinned = (recheck.get("pinned_head_sha") or "").strip().lower()
live = (recheck.get("live_head_sha") or "").strip().lower()
if not (_FULL_SHA.match(pinned) and _FULL_SHA.match(live)):
reasons.append(
"pinned/live head SHAs missing or not full 40-hex; fail closed"
)
elif pinned != live:
reasons.append(
"live head SHA no longer equals the pinned head; re-pin and "
"re-validate before mutation"
)
base_pinned = _normalize_ref(recheck.get("pinned_base_ref"))
base_live = _normalize_ref(recheck.get("live_base_ref"))
if not base_pinned or not base_live:
reasons.append("base refs missing from live-state recheck; fail closed")
elif base_pinned != base_live:
reasons.append(
f"base branch changed from '{base_pinned}' to '{base_live}'"
)
blocking = recheck.get("blocking_change_requests")
if blocking is None:
reasons.append(
"blocking review state not checked; fail closed"
)
elif blocking:
reasons.append(
"an undismissed REQUEST_CHANGES / blocking review state remains "
"unresolved"
)
proven = not reasons
return {"proven": proven, "block": not proven, "reasons": reasons}
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
def assess_role_boundary(proof):
"""Assess reviewer/author role separation for blind queue workflows.
Issue #175 blocks a reviewer queue task from silently becoming author
implementation work. Issue #179 also requires reviewer workflows to
report namespace use and justify any foreign namespace calls. This helper
accepts both forms:
implementation work. The workflow may use both namespaces only when that
mixed use is explicit, justified, and non-mutating; author mutations after
a reviewer queue task require an explicit operator authorization.
- the #175 dict proof with mutation details, or
- the #179 keyword form: ``task_role``, ``namespaces_used``,
``justification``.
*proof* keys:
``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example
'blind_pr_queue_review'), ``reviewer_namespace_used``,
``author_namespace_used``, ``author_mutations`` (list), ``review_mutations``
(list), ``operator_authorized_author_work``, ``mixed_namespace_justification``,
``scratch_evidence_claimed``, and ``scratch_evidence_durable``.
"""
if proof is None:
namespaces_reported = namespaces_used is not None
namespaces = list(namespaces_used or [])
proof = {
"task_role": task_role,
"reviewer_namespace_used": any(
"reviewer" in (namespace or "").lower()
for namespace in namespaces
),
"author_namespace_used": any(
"author" in (namespace or "").lower()
for namespace in namespaces
),
"mixed_namespace_justification": justification,
"author_mutations": [],
"review_mutations": [],
"_namespaces_used": namespaces,
"_namespaces_reported": namespaces_reported,
}
else:
proof = dict(proof or {})
proof = proof or {}
task_role = (proof.get("task_role") or "").strip().lower()
task_kind = (proof.get("task_kind") or "").strip().lower()
author_mutations = list(proof.get("author_mutations") or [])
@@ -511,8 +364,6 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
if task_role not in {"reviewer", "author"}:
reasons.append("task role missing or unknown; role boundary unproven")
if proof.get("_namespaces_reported") is False:
reasons.append("namespaces used were not reported; fail closed")
if task_role == "reviewer":
if author_mutations and not authorized:
@@ -558,35 +409,18 @@ def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
status = "clean"
safe_next_action = "proceed"
namespaces = proof.get("_namespaces_used")
if namespaces is None:
namespaces = []
if reviewer_used:
namespaces.append("gitea-reviewer")
if author_used:
namespaces.append("gitea-author")
foreign = [
namespace for namespace in namespaces
if task_role and task_role not in (namespace or "").lower()
]
return {
"status": status,
"clean": status == "clean",
"proven": status == "clean",
"reasons": reasons,
"violations": violations,
"safe_next_action": safe_next_action,
"foreign_namespaces": foreign,
"justified": bool(mixed_justification),
}
def build_final_report(checkout_proof, inventory, validation, contamination,
identity_eligible, merge_performed,
issue_status_verified,
capability_evidence=None, sweep=None, live_state=None,
role_boundary=None):
issue_status_verified, role_boundary=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -597,13 +431,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
- 'downgraded' — one or more proofs missing/weak; do not merge.
- 'blocked' — a *violation*: a merge was claimed although the proofs
did not allow one.
#179 raises the A bar: the report must also carry exact capability
evidence (``assess_capability_evidence``), an exact secret/provenance
sweep (``assess_sweep_evidence``), a pre-mutation live-state recheck
(``assess_live_state_recheck`` — also required for ``merge_allowed``),
and a clean role boundary (``assess_role_boundary``). Omitting any of
them downgrades; a merge without the live recheck is a violation.
"""
contamination_status = contamination.get("status", "unknown")
checkout_proven = bool(checkout_proof.get("proven"))
@@ -616,29 +443,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
}
role_status = role_boundary.get("status", "warning")
capability_evidence = capability_evidence or {
"proven": False,
"reasons": ["capability evidence not provided (#179)"],
}
sweep = sweep or {
"verdict": "missing",
"proven": False,
"reasons": ["secret/provenance sweep evidence not provided (#179)"],
}
live_state = live_state or {
"proven": False,
"block": True,
"reasons": ["pre-mutation live-state recheck not provided (#179)"],
}
role_boundary = role_boundary or {
"proven": False,
"reasons": ["role-boundary/namespace usage not reported (#179)"],
}
capability_proven = bool(capability_evidence.get("proven"))
sweep_proven = bool(sweep.get("proven"))
live_state_proven = bool(live_state.get("proven"))
role_boundary_clean = bool(role_boundary.get("proven"))
downgrade_reasons = []
if not identity_eligible:
downgrade_reasons.append("identity/profile not eligible for review")
@@ -662,23 +466,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
downgrade_reasons.extend(role_boundary.get("reasons", []))
if not issue_status_verified:
downgrade_reasons.append("linked issue status not verified")
if not capability_proven:
downgrade_reasons.append("exact capability evidence missing (#179)")
downgrade_reasons.extend(capability_evidence.get("reasons", []))
if not sweep_proven:
downgrade_reasons.append(
f"secret/provenance sweep evidence is "
f"{sweep.get('verdict', 'missing')} (#179)"
)
downgrade_reasons.extend(sweep.get("reasons", []))
if not live_state_proven:
downgrade_reasons.append(
"pre-mutation live-state recheck missing or failed (#179)"
)
downgrade_reasons.extend(live_state.get("reasons", []))
if not role_boundary_clean:
downgrade_reasons.append("role/namespace boundary not clean (#179)")
downgrade_reasons.extend(role_boundary.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -687,8 +474,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
and role_status == "clean"
and validation_claimable
and validation.get("verdict") != "invalid"
# #179: no merge without a proven final live-state recheck.
and live_state_proven
)
violations = []
@@ -723,10 +508,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"merge_allowed": merge_allowed,
"merge_performed": bool(merge_performed),
"issue_status_verified": bool(issue_status_verified),
"capability_evidence_proven": capability_proven,
"sweep_verdict": sweep.get("verdict"),
"live_state_recheck_proven": live_state_proven,
"role_boundary_clean": role_boundary_clean,
}
@@ -749,6 +530,7 @@ HANDOFF_BASE_FIELDS = (
("Files changed", ("files changed", "changed", "files")),
("Validation", ("validation",)),
("Mutations", ("mutations",)),
("Workspace mutations", ("workspace mutations",)),
("Current status", ("current status", "status")),
("Blockers", ("blockers",)),
("Next", ("next",)),
@@ -799,7 +581,7 @@ def _handoff_section_lines(report_text):
return lines[start:]
def assess_controller_handoff(report_text, role=None):
def assess_controller_handoff(report_text, role=None, local_edits=False):
"""Issue #182: final reports without a Controller Handoff downgrade.
Verdicts:
@@ -811,6 +593,7 @@ def assess_controller_handoff(report_text, role=None):
The handoff supplements the full report; this helper never validates
the full report body, only the continuation summary.
"""
import re
section = _handoff_section_lines(report_text)
if section is None:
return {
@@ -824,10 +607,14 @@ def assess_controller_handoff(report_text, role=None):
}
labels = []
fields_dict = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
k, v = stripped.split(":", 1)
label = k.strip().lower()
labels.append(label)
fields_dict[label] = v.strip()
required = list(HANDOFF_BASE_FIELDS)
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
@@ -846,6 +633,40 @@ def assess_controller_handoff(report_text, role=None):
"reasons": [f"handoff missing required field: {m}"
for m in missing],
}
# Validate issue/PR references for exact number and no forbidden terms (Issue #194 / #196)
for alias in ("selected issue", "pr number opened", "pr opened", "pr number", "selected pr"):
val = fields_dict.get(alias)
if val:
numbers = re.findall(r"\d+", val)
has_forbidden = any(term in val.lower() for term in ("equivalent", "related", "same", "/"))
if len(numbers) != 1 or has_forbidden:
field_name = "Selected issue/PR"
for name, aliases_list in required:
if alias in aliases_list:
field_name = name
break
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": [field_name],
"reasons": [
f"{field_name} must specify exactly one number and no ambiguous references (got: '{val}')"
],
}
if local_edits:
workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
if not workspace_mutations_val or workspace_mutations_val == "none":
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": ["Workspace mutations"],
"reasons": [
"Workspace mutations cannot be 'none' or empty when local edits exist"
],
}
return {
"verdict": "complete",
"downgraded": False,
@@ -854,58 +675,6 @@ def assess_controller_handoff(report_text, role=None):
}
ROUTE_HANDOFF_FIELDS = (
("Task type", ("task type", "task_type")),
("Required role", ("required role", "required_role")),
("Active role", ("active role", "active_role")),
("Route result", ("route result", "route_result")),
)
def assess_role_route_handoff(report_text, route_result=None):
"""Issue #206: final handoff must record role routing verdict."""
text = report_text or ""
lower = text.lower()
missing = []
for name, aliases in ROUTE_HANDOFF_FIELDS:
if not any(alias in lower for alias in aliases):
missing.append(name)
if route_result is not None:
expected = str(route_result.get("route_result", "")).lower()
if expected and expected not in lower:
missing.append("route result value")
if missing:
return {
"complete": False,
"downgraded": True,
"missing_fields": missing,
"reasons": [
f"handoff missing role-route field: {field}" for field in missing
],
}
return {
"complete": True,
"downgraded": False,
"missing_fields": [],
"reasons": [],
}
def assess_capability_stop_terminal_report(
report_text,
*,
trust_gate_status=None,
capability_denied=True,
):
"""Issue #197: reports after reviewer capability denial must stay pure."""
from capability_stop_terminal import assess_capability_stop_report
return assess_capability_stop_report(
report_text,
trust_gate_status=trust_gate_status,
capability_denied=capability_denied,
)
# ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
#
@@ -999,10 +768,3 @@ def pr_inventory_trust_gate(
"reasons": [],
"corroborated": corroborated,
}
def assess_duplicate_search_proof(report_text, matches):
"""#207: reject LLM duplicate summaries that omit known title matches."""
return issue_duplicate_gate.assess_duplicate_search_proof(
report_text, matches
)
-195
View File
@@ -1,195 +0,0 @@
"""Pre-task role/session router (#206).
Classifies a declared task type against the active MCP profile/session and
returns a route result before any downstream mutation tools run.
"""
from __future__ import annotations
ROUTE_ALLOWED = "allowed_current_session"
ROUTE_WRONG_ROLE = "wrong_role_stop"
ROUTE_TO_AUTHOR = "route_to_author_session"
ROUTE_TO_REVIEWER = "route_to_reviewer_session"
ROUTE_AMBIGUOUS = "ambiguous_task_stop"
REVIEWER_TASKS = frozenset({
"review_pr",
"merge_pr",
"blind_pr_queue_review",
"request_changes_pr",
"approve_pr",
})
AUTHOR_TASKS = frozenset({
"create_issue",
"comment_issue",
"close_issue",
"claim_issue",
"create_branch",
"push_branch",
"create_pr",
"comment_pr",
"address_pr_change_requests",
"delete_branch",
})
TASK_REQUIRED_ROLE = {
"create_issue": "author",
"comment_issue": "author",
"close_issue": "author",
"claim_issue": "author",
"create_branch": "author",
"push_branch": "author",
"create_pr": "author",
"comment_pr": "author",
"address_pr_change_requests": "author",
"delete_branch": "author",
"review_pr": "reviewer",
"merge_pr": "reviewer",
"blind_pr_queue_review": "reviewer",
"request_changes_pr": "reviewer",
"approve_pr": "reviewer",
}
WRONG_ROLE_REVIEWER_MSG = (
"Wrong role/session for reviewer task. Launch reviewer MCP namespace."
)
_session_last_route: dict | None = None
def required_role_for_task(task_type: str) -> str | None:
return TASK_REQUIRED_ROLE.get((task_type or "").strip())
def route_task_session(
task_type: str,
*,
active_profile: str,
active_role_kind: str,
allowed_in_current_session: bool,
runtime_switching_supported: bool = False,
) -> dict:
"""Return routing verdict for *task_type* under the active session."""
task_type = (task_type or "").strip()
required_role = required_role_for_task(task_type)
if required_role is None:
result = {
"task_type": task_type,
"required_role": None,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_AMBIGUOUS,
"downstream_allowed": False,
"reasons": [
f"unknown task type '{task_type}'; cannot route session "
"(fail closed)"
],
"message": (
"Ambiguous task type; relaunch with an explicit task before "
"any tool use."
),
}
_record_route(result)
return result
if allowed_in_current_session:
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_ALLOWED,
"downstream_allowed": True,
"reasons": [],
"message": "Task role matches active session; proceed.",
}
_record_route(result)
return result
if required_role == "reviewer":
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_WRONG_ROLE,
"downstream_allowed": False,
"reasons": [
WRONG_ROLE_REVIEWER_MSG,
"Reviewer tasks cannot run in author-bound sessions.",
"Static-profile mode does not permit in-place role switching.",
],
"message": WRONG_ROLE_REVIEWER_MSG,
"runtime_switching_supported": runtime_switching_supported,
"profile_switch_blocked": not runtime_switching_supported,
}
_record_route(result)
return result
if required_role == "author":
route = ROUTE_TO_AUTHOR
message = (
"Wrong role/session for author task. Launch author MCP namespace."
)
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": route,
"downstream_allowed": False,
"reasons": [message],
"message": message,
"runtime_switching_supported": runtime_switching_supported,
"profile_switch_blocked": not runtime_switching_supported,
}
_record_route(result)
return result
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_AMBIGUOUS,
"downstream_allowed": False,
"reasons": ["unable to classify task role (fail closed)"],
"message": "Ambiguous task type; stop before any mutation.",
}
_record_route(result)
return result
def last_route() -> dict | None:
return _session_last_route
def clear_route_state():
global _session_last_route
_session_last_route = None
def _record_route(result: dict):
global _session_last_route
_session_last_route = dict(result)
def check_author_mutation_after_reviewer_stop(mutation_task: str) -> tuple[bool, list[str]]:
"""Block author-side fallback after a reviewer wrong_role_stop (#206)."""
last = _session_last_route
if not last:
return True, []
if last.get("route_result") != ROUTE_WRONG_ROLE:
return True, []
if last.get("required_role") != "reviewer":
return True, []
if mutation_task in AUTHOR_TASKS:
return False, [
WRONG_ROLE_REVIEWER_MSG,
"Author-side mutations are blocked after a reviewer-task "
"wrong_role_stop unless the operator explicitly changes the "
"task and relaunches an author MCP session.",
f"Attempted fallback mutation: {mutation_task}",
]
return True, []
+10
View File
@@ -40,6 +40,16 @@ start_ref="${2:-prgs/master}"
# Enforce issue-linked, traceable branch names (issue → branch → worktree → PR).
if [[ "$allow_unlinked" -eq 0 ]]; then
if [[ ! -f "/tmp/gitea_issue_lock.json" ]]; then
echo "Error: Issue lock file '/tmp/gitea_issue_lock.json' is missing. You must lock exactly one issue before branch creation (fail closed)." >&2
exit 2
fi
locked_branch=$(python3 -c "import json; print(json.load(open('/tmp/gitea_issue_lock.json')).get('branch_name', ''))")
if [[ "$branch" != "$locked_branch" ]]; then
echo "Error: Requested branch '$branch' does not match locked branch '$locked_branch' (fail closed)." >&2
exit 2
fi
if [[ "$branch" =~ ^(fix|feat|docs|chore)/issue-[0-9]+-.+ ]] \
|| [[ "$branch" =~ ^review/pr-[0-9]+-.+ ]]; then
:
+8 -24
View File
@@ -232,35 +232,19 @@ Worktree folder = branch with `/` replaced by `-`
validation result after the command has completed and its output has
been read (`review_proofs.assess_validation_report`).
10. **Do not merge if checks fail. Do not merge if the reviewer is the author.**
11. **#179 A-bar proofs** (all fail closed when missing —
`review_proofs.assess_capability_evidence`, `assess_sweep_evidence`,
`assess_live_state_recheck`, `assess_role_boundary`):
- Capability claims must cite exact `gitea_resolve_task_capability`
output (or runtime context); a bare "capability checks passed" is
downgraded.
- The secret/provenance sweep must state the exact command/script/
pattern/named method and the scope scanned.
- Immediately before submitting a review verdict (and again before any
merge), re-read live PR state and prove: still open, live head ==
pinned head, base unchanged, no unresolved blocking review state.
- Reviewer runs stay in the reviewer namespace; any author-namespace
call requires an explicit justification in the report.
12. The final report must distinguish (`review_proofs.build_final_report`):
11. The final report must distinguish (`review_proofs.build_final_report`):
identity eligible; PR author different from reviewer; session
contamination absent (with evidence); validation performed on the pinned
head; capability evidence; sweep verdict; live-state recheck; role
boundary; merge performed; issue status verified. If any proof is
missing, stop or downgrade the result instead of merging confidently.
contamination absent (with evidence); role boundary clean; validation
performed on the pinned head; merge performed; issue status verified. If
any proof is missing, stop or downgrade the result instead of merging
confidently.
## G. Merge / cleanup workflow
Only an eligible (non-author) reviewer merges. Before merging: always verify
the authenticated identity **and** the PR author; cite exact capability
evidence for merge_pr (#179); respect runtime profile gates; run independent
validation (do not trust the author's reported results); perform the **final
live-state recheck** (#179 — PR still open, live head == pinned head, base
unchanged, no unresolved blocking review state) immediately before the merge
mutation; and merge with a **pinned head SHA** and, where supported, the
the authenticated identity **and** the PR author; respect runtime profile
gates; run independent validation (do not trust the author's reported
results); and merge with a **pinned head SHA** and, where supported, the
**expected changed-file set**, so a moved head or widened diff refuses the
merge. After a real merge:
@@ -20,21 +20,10 @@ Steps:
*If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
2. Verify authenticated identity + active profile.
3. Confirm PR #<pr>: author (not you), state open, mergeable, review approved. Check if PR body uses `Closes #N` or `Fixes #N`; if it uses `Implements #N` or `Refs #N`, manual closing will be needed in step 29.
4. Capability evidence (#179): cite the exact gitea_resolve_task_capability
output (or runtime context) proving merge_pr is allowed — a bare
"capability checks passed" claim is downgraded.
5. Final live-state recheck (#179), immediately before the merge mutation —
re-read the live PR and prove:
- PR still open
- live head SHA still equals the pinned/reviewed head SHA
- base branch unchanged
- no undismissed REQUEST_CHANGES / blocking review state remains
If any recheck fails → STOP, re-pin, re-validate.
6. If any gate fails → STOP and report.
7. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
pinning the reviewed head SHA (expected_head_sha) and, where supported,
the changed-file set.
8. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
4. If any gate fails → STOP and report.
4. Merge with explicit confirmation (e.g. confirmation="MERGE PR <pr>"),
optionally pinning the reviewed head SHA / changed-file set.
5. Confirm remote master now contains the merge commit (or the expected changes if squash merged).
*Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.*
Then run the cleanup template (worktree-cleanup.md):
@@ -36,11 +36,6 @@ Steps:
- Target task role: reviewer identity (must NOT be the PR author)
*If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.*
2. Verify your authenticated identity (whoami) and the active profile.
Capability evidence (#179): cite the exact gitea_resolve_task_capability
output (or runtime context) for review_pr (and merge_pr if merging later);
a bare "capability checks passed" claim is downgraded. Stay in the
reviewer namespace: any author-namespace call must be justified in the
report (#179).
3. Fetch the PR facts: PR author, head SHA, state (must be open), base branch.
Pin the head SHA in your notes; every later step validates THAT SHA.
4. If authenticated user == PR author → STOP (no self-review).
@@ -63,23 +58,12 @@ Steps:
If HEAD does not match the pinned head → STOP before review/merge.
7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches
issue #<n>; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`).
Secret/provenance sweep must be exact (#179): state the exact command,
script, grep pattern, or named sweep method AND the scope scanned (e.g.
`git diff prgs/master...HEAD | grep -inE '<pattern>'`); "checked the diff
for secrets" alone is downgraded.
8. Run the test suite; report the exact command and exact results — pass/fail
plus passed/skipped/failed counts, any ignored paths and why they are safe
to ignore, and whether the command differs from the repository's canonical
validation command. Only claim a result after the output has been read.
9. Final live-state recheck (#179), immediately before submitting the review
verdict — re-read the live PR and prove:
- PR still open
- live head SHA still equals the pinned head SHA from step 3
- base branch unchanged
- no undismissed REQUEST_CHANGES / blocking review state left unaccounted
If anything moved → STOP, re-pin, re-validate before any verdict.
10. Post the review verdict: approve only if scope is clean and checks pass;
otherwise request changes with specifics. Never merge from this review step.
9. Post the review verdict: approve only if scope is clean and checks pass;
otherwise request changes with specifics. Never merge from this review step.
Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md):
Review Metadata:
-38
View File
@@ -1,38 +0,0 @@
"""Shared pytest fixtures for the Gitea-Tools test suite."""
import sys
import pytest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@pytest.fixture(autouse=True)
def _reset_mutation_authority(monkeypatch):
"""Isolate the in-process mutation authority between tests (#199).
The mutation-authority gate stays LIVE in every test — this fixture only
clears the per-process record and the session profile lock so one test's
seeded authority (or an intentionally mismatched one) cannot leak into
the next test. It must never replace verify_mutation_authority with a
no-op: individual tests that need a specific authority state set it up
explicitly.
"""
monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
try:
import mcp_server
except Exception:
yield
return
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
try:
import capability_stop_terminal
capability_stop_terminal.clear()
except Exception:
pass
yield
try:
import capability_stop_terminal
capability_stop_terminal.clear()
except Exception:
pass
+7 -25
View File
@@ -175,12 +175,9 @@ class _AuditWiringBase(unittest.TestCase):
class TestSimpleToolAudit(_AuditWiringBase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_issue_success_audited(self, _auth, _get_all, mock_api, _role):
def test_create_issue_success_audited(self, _auth, mock_api):
# 1: create POST result, 2: identity /user lookup for the audit record.
mock_api.side_effect = [
{"number": 11, "html_url": "https://gitea.prgs.cc/issues/11"},
@@ -199,12 +196,9 @@ class TestSimpleToolAudit(_AuditWiringBase):
self.assertEqual(rec["issue_number"], 11)
self.assertEqual(rec["request_metadata"]["title"], "Add thing")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_issue_failure_audited(self, _auth, _get_all, mock_api, _role):
def test_create_issue_failure_audited(self, _auth, mock_api):
mock_api.side_effect = [
RuntimeError("HTTP 500: boom"),
{"login": "author-bot"}, # identity lookup for the audit record
@@ -229,28 +223,19 @@ class TestSimpleToolAudit(_AuditWiringBase):
self.assertEqual(recs[0]["issue_number"], 42)
self.assertEqual(recs[0]["authenticated_username"], "mgr-bot")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role):
# No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file.
def test_disabled_writes_nothing_and_no_extra_call(self, _auth, mock_api):
# No GITEA_AUDIT_LOG -> audit is a no-op: exactly one API call, no file.
mock_api.return_value = {"number": 1, "html_url": "http://x/1"}
with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True):
gitea_create_issue(title="x", remote="prgs")
issue_posts = [
c for c in mock_api.call_args_list if c.args[0] == "POST"
]
self.assertEqual(len(issue_posts), 1)
mock_api.assert_called_once()
self.assertEqual(self._records(), [])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_secrets_never_written(self, _auth, _get_all, mock_api, _role):
def test_secrets_never_written(self, _auth, mock_api):
mock_api.side_effect = [
{"number": 3, "html_url": "http://x/3"},
{"login": "author-bot"},
@@ -263,12 +248,9 @@ class TestSimpleToolAudit(_AuditWiringBase):
for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()):
self.assertNotIn(secret, blob)
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_audit_failure_never_breaks_action(self, _auth, _get_all, mock_api, _role):
def test_audit_failure_never_breaks_action(self, _auth, mock_api):
mock_api.side_effect = [
{"number": 9, "html_url": "http://x/9"},
{"login": "author-bot"},
-164
View File
@@ -1,164 +0,0 @@
"""Tests for capability stop terminal mode (#197)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import capability_stop_terminal
import gitea_config
import mcp_server
from review_proofs import assess_capability_stop_terminal_report
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.pr.create",
"gitea.branch.push",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
},
"rules": {"allow_runtime_switching": False},
}
class TestCapabilityStopTerminal(unittest.TestCase):
def setUp(self):
capability_stop_terminal.clear()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {
"host": "gitea.example.com",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
capability_stop_terminal.clear()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir.cleanup()
def _env(self):
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": "prgs-author",
"GITEA_TOKEN_AUTHOR": "author-pass",
}
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
with patch.dict(os.environ, self._env()):
res = mcp_server.gitea_resolve_task_capability(
task="review_pr", remote="prgs"
)
self.assertTrue(res["stop_required"])
self.assertTrue(res.get("terminal_mode"))
self.assertTrue(capability_stop_terminal.is_active())
self.assertIn(
capability_stop_terminal.TERMINAL_REPORT_HEADING,
res["terminal_report"]["heading"],
)
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
with patch.dict(os.environ, self._env()):
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.gitea_list_prs(remote="prgs")
self.assertIn("Cannot perform reviewer task", str(ctx.exception))
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_eligibility_check_blocked_after_stop(self, _auth, _api):
with patch.dict(os.environ, self._env()):
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
res = mcp_server.gitea_check_pr_eligibility(
pr_number=193, action="review", remote="prgs"
)
self.assertFalse(res["eligible"])
self.assertTrue(res.get("terminal_mode"))
def test_report_with_pr_selection_impure(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Selected PR #193 for review anyway."
)
result = assess_capability_stop_terminal_report(report)
self.assertFalse(result["pure"])
def test_session_eligibility_wording_blocked(self):
ok, violations = capability_stop_terminal.validate_eligibility_wording(
"PR 193 is not authored by this session so it is eligible."
)
self.assertFalse(ok)
self.assertTrue(violations)
def test_rebase_fallback_blocked_in_report(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Or have me rebase conflicted PR 193."
)
result = assess_capability_stop_terminal_report(report)
self.assertFalse(result["pure"])
self.assertTrue(
any("author fallback" in v for v in result["violations"])
)
def test_empty_queue_without_trusted_empty_blocked(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"The repo has 0 open PRs."
)
result = assess_capability_stop_terminal_report(
report, trust_gate_status="untrusted_empty"
)
self.assertFalse(result["pure"])
def test_pure_terminal_report_passes(self):
report = (
"Cannot perform reviewer task under current profile. "
"No reviewer mutations performed.\n"
"Identity: jcwalker3 / prgs-author.\n"
"Required: prgs-reviewer.\n"
"No mutations performed."
)
result = assess_capability_stop_terminal_report(report)
self.assertTrue(result["pure"])
if __name__ == "__main__":
unittest.main()
-154
View File
@@ -1,154 +0,0 @@
"""Tests for the gated PR close path (Issue #216).
``gitea_edit_pr(state='closed')`` requires the distinct ``gitea.pr.close``
capability: without it the close fails closed (no auth lookup, no API call,
structured permission report). With it — the explicit operator-directed
contaminated-PR closure path — the close proceeds and is audited as a
distinct ``close_pr`` action carrying the required capability, so final
reports can prove exactly which mutation capability was exercised.
Ordinary edits (title/body/base) and reopening never require the close
capability: PR comment, PR edit, and PR close remain distinct capabilities.
"""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
from mcp_server import gitea_edit_pr
FAKE_AUTH = "token fake"
AUTHOR_NO_CLOSE = {
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.read", "gitea.pr.create", "gitea.pr.comment",
"gitea.branch.push", "gitea.issue.comment",
],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
"audit_label": "prgs-author",
}
AUTHOR_WITH_CLOSE = {
"profile_name": "prgs-author-closer",
"allowed_operations": AUTHOR_NO_CLOSE["allowed_operations"] + ["gitea.pr.close"],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
"audit_label": "prgs-author-closer",
}
class TestEditPrCloseGate(unittest.TestCase):
def setUp(self):
self.mock_api = patch("mcp_server.api_request").start()
self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
# Deterministic permission reports: no real operator config, no switching.
patch("gitea_config.load_config", return_value={}).start()
patch("gitea_config.is_runtime_switching_enabled", return_value=False).start()
patch("gitea_audit.audit_enabled", return_value=False).start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
mcp_server._IDENTITY_CACHE.clear()
def _set_profile(self, profile):
patch("mcp_server.get_profile", return_value=profile).start()
def test_close_blocked_without_close_capability(self):
# Author profile without gitea.pr.close: the broad edit path must not
# be usable as an untracked close fallback.
self._set_profile(AUTHOR_NO_CLOSE)
res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
self.assertFalse(res["success"])
self.assertFalse(res["performed"])
self.assertEqual(res["requested_state"], "closed")
self.assertEqual(res["required_permission"], "gitea.pr.close")
self.assertTrue(res["reasons"])
report = res["permission_report"]
self.assertEqual(report["required_permission"], "gitea.pr.close")
self.assertEqual(report["active_profile"], "prgs-author")
self.mock_api.assert_not_called()
self.mock_auth.assert_not_called()
def test_close_blocked_when_close_forbidden(self):
profile = dict(AUTHOR_WITH_CLOSE)
profile["forbidden_operations"] = ["gitea.pr.close"]
self._set_profile(profile)
res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
self.assertFalse(res["success"])
self.assertIn("profile forbids 'gitea.pr.close'", res["reasons"])
self.mock_api.assert_not_called()
def test_close_blocked_when_profile_unresolvable(self):
patch("mcp_server.get_profile", side_effect=RuntimeError("no profile")).start()
res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
self.assertFalse(res["success"])
self.assertTrue(any("fail closed" in r for r in res["reasons"]))
self.mock_api.assert_not_called()
def test_operator_directed_close_allowed_and_audited(self):
# Explicit operator-directed contaminated-PR closure: the operator
# granted gitea.pr.close, so the close proceeds and the audit trail
# records a distinct close_pr action with the capability proof.
self._set_profile(AUTHOR_WITH_CLOSE)
patch("gitea_audit.audit_enabled", return_value=True).start()
mock_write = patch("gitea_audit.write_event").start()
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/user" in url:
return {"login": "jcwalker3"}
if method == "PATCH" and "pulls/205" in url:
self.assertEqual(payload["state"], "closed")
return {
"number": 205,
"title": "Contaminated PR",
"state": "closed",
"html_url": "url",
"body": "No issue link",
"head": {"ref": "feat/invalid-provenance"},
}
return {}
self.mock_api.side_effect = api_side_effect
res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs")
self.assertTrue(res["success"])
self.assertEqual(res["state"], "closed")
mock_write.assert_called()
event = mock_write.call_args[0][0]
self.assertEqual(event["action"], "close_pr")
self.assertEqual(
event["request_metadata"]["required_permission"], "gitea.pr.close")
def test_title_edit_needs_no_close_capability(self):
# PR edit and PR close are distinct capabilities: retitling stays on
# the ordinary edit path.
self._set_profile(AUTHOR_NO_CLOSE)
self.mock_api.return_value = {
"number": 7, "title": "Renamed", "state": "open",
"body": "", "html_url": "u"}
res = gitea_edit_pr(pr_number=7, title="Renamed", remote="prgs")
self.assertTrue(res["success"])
def test_reopen_needs_no_close_capability(self):
self._set_profile(AUTHOR_NO_CLOSE)
self.mock_api.return_value = {
"number": 7, "title": "T", "state": "open",
"body": "", "html_url": "u"}
res = gitea_edit_pr(pr_number=7, state="open", remote="prgs")
self.assertTrue(res["success"])
def test_invalid_state_fails_closed_before_auth(self):
# A case-variant state can neither bypass the close gate nor reach
# the API: it is rejected as pure validation, before auth.
self._set_profile(AUTHOR_WITH_CLOSE)
with self.assertRaises(ValueError):
gitea_edit_pr(pr_number=7, state="CLOSED", remote="prgs")
self.mock_api.assert_not_called()
self.mock_auth.assert_not_called()
if __name__ == "__main__":
unittest.main()
-173
View File
@@ -1,173 +0,0 @@
"""Tests for pre-create issue duplicate gate (Issue #207)."""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from issue_duplicate_gate import ( # noqa: E402
VERDICT_AMBIGUOUS,
VERDICT_DUPLICATE,
VERDICT_NO_DUPLICATE,
assess_duplicate_search_proof,
assess_pre_create_duplicate,
normalize_issue_title,
pre_create_issue_duplicate_gate,
titles_near_duplicate,
)
from mcp_server import gitea_create_issue # noqa: E402
from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
CANONICAL_TITLE = (
"Add hard queue-target resolution wall before PR inventory "
"or empty-queue claims"
)
class TestNormalizeAndNearDuplicate(unittest.TestCase):
def test_normalize_strips_punctuation_and_case(self):
norm = normalize_issue_title(" Hard-Wall: Queue Target! ")
self.assertEqual(norm, "hard wall queue target")
def test_exact_title_match(self):
self.assertTrue(titles_near_duplicate(CANONICAL_TITLE, CANONICAL_TITLE))
def test_punctuation_and_capitalization_only_differs(self):
proposed = CANONICAL_TITLE.upper().replace("-", "")
self.assertTrue(titles_near_duplicate(proposed, CANONICAL_TITLE))
def test_hard_wall_vs_wall_wording(self):
self.assertTrue(
titles_near_duplicate(
"Add hard wall before PR inventory",
"Add wall before PR inventory",
)
)
class TestPreCreateGate(unittest.TestCase):
def test_blocks_exact_duplicate_title(self):
existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
gate = assess_pre_create_duplicate(CANONICAL_TITLE, existing)
self.assertEqual(gate["verdict"], VERDICT_DUPLICATE)
self.assertFalse(gate["performed"])
self.assertEqual(gate["matches"][0]["number"], 201)
def test_final_pre_create_check_blocks_even_if_llm_search_missed(self):
"""TOCTOU: gate re-queries at create time with the live issue list."""
stale_report = (
"Searched 20 open issues (#194, #196, PR #195); no duplicate found."
)
live_existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
proof = assess_duplicate_search_proof(stale_report, live_existing)
self.assertFalse(proof["valid"])
gate = assess_pre_create_duplicate(CANONICAL_TITLE, live_existing)
self.assertFalse(gate["performed"])
def test_invalid_proof_when_summary_omits_exact_duplicate(self):
report = "Reviewed nearby issues #194, #196, and PR #195; no duplicate."
matches = [{"number": 201, "title": CANONICAL_TITLE}]
result = assess_duplicate_search_proof(report, matches)
self.assertFalse(result["valid"])
self.assertTrue(any("201" in r for r in result["reasons"]))
def test_operator_split_override_allowed_with_relationship(self):
existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}]
gate = pre_create_issue_duplicate_gate(
CANONICAL_TITLE,
existing,
allow_override=True,
split_from_issue=201,
)
self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE)
self.assertTrue(gate["performed"])
self.assertTrue(gate.get("override_applied"))
def test_ambiguous_when_too_many_near_matches(self):
base = "Add hard wall before PR inventory"
existing = [
{"number": n, "title": f"{base} variant {n}", "state": "open"}
for n in range(1, 6)
]
gate = assess_pre_create_duplicate(base, existing)
self.assertEqual(gate["verdict"], VERDICT_AMBIGUOUS)
self.assertFalse(gate["performed"])
def test_no_duplicate_when_unique_title(self):
gate = assess_pre_create_duplicate(
"Brand new unique issue title",
[{"number": 1, "title": "Unrelated backlog cleanup", "state": "open"}],
)
self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE)
self.assertTrue(gate["performed"])
class TestCreateIssueMCPGate(unittest.TestCase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_blocks_normalized_title_match_without_override(
self, _auth, mock_get_all, mock_api, mock_role_check
):
mock_role_check.return_value = (True, [])
mock_get_all.return_value = [
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
]
result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs")
self.assertFalse(result["performed"])
self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE)
mock_api.assert_not_called()
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_override_creates_with_split_relationship_in_body(
self, _auth, mock_get_all, mock_api, mock_role_check
):
mock_role_check.return_value = (True, [])
mock_get_all.return_value = [
{"number": 201, "title": CANONICAL_TITLE, "state": "open"},
]
mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"}
result = gitea_create_issue(
title=CANONICAL_TITLE,
body="Follow-up scope",
remote="prgs",
allow_duplicate_override=True,
split_from_issue=201,
)
self.assertEqual(result["number"], 210)
payload = mock_api.call_args[0][3]
self.assertIn("Operator-approved split from #201.", payload["body"])
self.assertIn("Follow-up scope", payload["body"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop")
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_when_no_duplicate(
self, _auth, _get_all, mock_api, mock_role_check
):
mock_role_check.return_value = (True, [])
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
result = gitea_create_issue(title="Unique new issue", body="body text")
self.assertEqual(result["number"], 1)
class TestReviewProofsWrapper(unittest.TestCase):
def test_review_proofs_reexports_duplicate_search_validator(self):
report = "Checked open issues; nothing similar."
matches = [{"number": 200, "title": CANONICAL_TITLE}]
result = proof_assess(report, matches)
self.assertFalse(result["valid"])
if __name__ == "__main__":
unittest.main()
+243 -109
View File
@@ -5,6 +5,7 @@ the MCP protocol) with mocked API responses.
"""
import json
import os
os.environ["GITEA_TEST_ENVIRONMENT"] = "1"
import sys
import unittest
from unittest.mock import patch, MagicMock
@@ -33,11 +34,15 @@ from mcp_server import ( # noqa: E402
gitea_submit_pr_review,
gitea_list_issue_comments,
gitea_create_issue_comment,
gitea_lock_issue,
)
from gitea_auth import get_profile # noqa: E402
import gitea_config # noqa: E402
import mcp_server
# Globally disable verification check for existing isolated unit tests
_real_verify = mcp_server.verify_mutation_authority
mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -47,12 +52,9 @@ FAKE_AUTH = "Basic dGVzdDp0ZXN0"
# ---------------------------------------------------------------------------
class TestCreateIssue(unittest.TestCase):
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_issue(self, _auth, _get_all, mock_api, _role):
def test_creates_issue(self, _auth, mock_api):
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
with patch.dict(os.environ, {}, clear=True):
result = gitea_create_issue(title="Test issue", body="body text")
@@ -64,23 +66,17 @@ class TestCreateIssue(unittest.TestCase):
self.assertEqual(call_args[0][0], "POST")
self.assertEqual(call_args[0][3]["title"], "Test issue")
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role):
def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
result = gitea_create_issue(title="Test issue", body="body text")
self.assertIn("issues/1", result["url"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@patch("mcp_server.api_request")
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role):
def test_creates_on_prgs(self, _auth, mock_api):
mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
result = gitea_create_issue(title="Test", remote="prgs")
self.assertEqual(result["number"], 5)
@@ -96,10 +92,13 @@ class TestCreatePR(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_creates_pr(self, _auth, mock_api):
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api):
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, {}, clear=True):
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
payload = mock_api.call_args[0][3]
@@ -108,10 +107,13 @@ class TestCreatePR(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api):
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertIn("pulls/3", result["url"])
@@ -1632,8 +1634,7 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
patch("gitea_audit.audit_enabled", return_value=True).start()
self.mock_audit = patch("gitea_audit.write_event").start()
# gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216).
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start()
patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).start()
def tearDown(self):
patch.stopall()
@@ -2311,119 +2312,252 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
class TestVerifyMutationAuthority(unittest.TestCase):
"""In-process mutation authority (#199, refs #194).
The authority record lives in mcp_server._MUTATION_AUTHORITY (per
process, reset between tests by conftest); the CLI side-channel is
covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
lock file — nothing here touches /tmp.
"""
"""Test verification lock logic under various configurations."""
def setUp(self):
self.patch_profile = patch("mcp_server.get_profile")
self.mock_profile = self.patch_profile.start()
self.patch_username = patch("mcp_server._authenticated_username")
self.mock_username = self.patch_username.start()
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
# Restore real function for these tests
self._old_verify = mcp_server.verify_mutation_authority
mcp_server.verify_mutation_authority = _real_verify
def tearDown(self):
self.patch_profile.stop()
self.patch_username.stop()
mcp_server.verify_mutation_authority = self._old_verify
# Clean up lock file
if os.path.exists("/tmp/gitea_mutation_authority.lock"):
os.remove("/tmp/gitea_mutation_authority.lock")
def _authority(self, **overrides):
data = {
"initial_profile": "prgs-reviewer",
"initial_identity": "sysadmin",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin",
"remote": "prgs",
"task": "review_pr",
"role_pivot_authorized": False,
"role_pivot_record": None,
"pid": os.getpid(),
}
data.update(overrides)
mcp_server._MUTATION_AUTHORITY = data
def test_missing_authority_seeds_from_live_context(self):
# Approved preflight path (whoami → eligibility → mutation): the
# first mutation gate seeds the authority instead of failing closed,
# so the standard reviewer workflow keeps working.
mcp_server._MUTATION_AUTHORITY = None
mcp_server.verify_mutation_authority("prgs")
seeded = mcp_server._MUTATION_AUTHORITY
self.assertIsNotNone(seeded)
self.assertEqual(seeded["current_profile"], "prgs-reviewer")
self.assertEqual(seeded["current_identity"], "sysadmin")
self.assertEqual(seeded["remote"], "prgs")
def test_unresolved_profile_fails_closed(self):
self.mock_profile.return_value = {}
mcp_server._MUTATION_AUTHORITY = None
def test_missing_lock_fails_closed(self):
if os.path.exists("/tmp/gitea_mutation_authority.lock"):
os.remove("/tmp/gitea_mutation_authority.lock")
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
self.assertIn("profile unresolved", str(ctx.exception))
_real_verify("prgs")
self.assertIn("lock is missing", str(ctx.exception))
def test_mismatched_remote_fails(self):
self._authority(remote="dadeschools")
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "dadeschools",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin"
}, f)
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
_real_verify("prgs")
self.assertIn("does not match locked remote", str(ctx.exception))
def test_profile_flip_after_record_fails(self):
# Authority was recorded as author; the active profile now resolves
# as reviewer (e.g. an env-var flip mid-session) — refuse.
self._authority(
initial_profile="prgs-author",
initial_identity="jcwalker3",
current_profile="prgs-author",
current_identity="jcwalker3",
)
def test_mismatched_profile_fails(self):
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "prgs",
"current_profile": "prgs-author",
"current_identity": "jcwalker3"
}, f)
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
_real_verify("prgs")
self.assertIn("does not match locked authority", str(ctx.exception))
def test_session_lock_env_mismatch_fails(self):
# The launching session locked the environment to the author
# profile; the active profile resolves as reviewer — side-channel
# override rejected even with a matching in-process authority.
self._authority()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
self.assertIn("side-channel override rejected", str(ctx.exception))
def test_foreign_pid_authority_is_not_trusted(self):
# An authority record from another process (fork leftovers) is
# discarded and reseeded from the live context, never reused.
self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
mcp_server.verify_mutation_authority("prgs")
self.assertEqual(
mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
)
self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
def test_author_to_reviewer_pivot_blocked_without_authorization(self):
self._authority(
initial_profile="prgs-author",
initial_identity="jcwalker3",
)
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "prgs",
"initial_profile": "prgs-author",
"initial_identity": "jcwalker3",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin",
"role_pivot_authorized": False
}, f)
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
_real_verify("prgs", required_role="reviewer")
self.assertIn("without authorized role pivot", str(ctx.exception))
def test_authorized_pivot_is_allowed(self):
self._authority(
initial_profile="prgs-author",
initial_identity="jcwalker3",
role_pivot_authorized=True,
)
mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
def test_allowed_when_match(self):
self._authority()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
mcp_server.verify_mutation_authority("prgs")
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "prgs",
"initial_profile": "prgs-reviewer",
"initial_identity": "sysadmin",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin",
"role_pivot_authorized": False
}, f)
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
# Should pass without exception
_real_verify("prgs")
class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def tearDown(self):
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_success(self, _auth, mock_api):
mock_api.return_value = [] # no open PRs
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-195-mutations", remote="prgs")
self.assertIn("must contain locked issue pattern", str(ctx.exception))
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_open_pr_branch(self, _auth, mock_api):
mock_api.return_value = [{
"number": 200,
"head": {"ref": "feat/issue-196-boundary"},
"title": "Some PR",
"body": "No closes ref"
}]
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_lock_issue_reused_by_open_pr_closes_ref(self, _auth, mock_api):
mock_api.return_value = [{
"number": 200,
"head": {"ref": "feat/other-branch"},
"title": "Some PR",
"body": "fixes #196"
}]
with self.assertRaises(ValueError) as ctx:
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertIn("already tied to an open PR", str(ctx.exception))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_lock_fails(self, _auth):
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("Issue lock is missing", str(ctx.exception))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_branch_mismatch_fails(self, _auth):
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
self.assertIn("does not match locked branch", str(ctx.exception))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_forbidden_terms_fails(self, _auth):
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("contains forbidden term", str(ctx.exception))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth):
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
class TestPreflightVerification(unittest.TestCase):
"""Test workspace edits and pre-flight ordering verification."""
def setUp(self):
self.preflight_path = "/tmp/gitea_preflight_check.json"
if os.path.exists(self.preflight_path):
os.remove(self.preflight_path)
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
def tearDown(self):
if os.path.exists(self.preflight_path):
os.remove(self.preflight_path)
os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
@patch("subprocess.run")
def test_record_preflight_detects_violation_whoami(self, mock_run):
mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
from mcp_server import record_preflight_check, verify_preflight_purity
record_preflight_check("whoami")
with open(self.preflight_path, "r") as f:
data = json.load(f)
self.assertTrue(data.get("whoami_preflight_violation"))
self.assertTrue(data.get("whoami_called"))
data["capability_called"] = True
with open(self.preflight_path, "w") as f:
json.dump(data, f)
with self.assertRaises(RuntimeError) as ctx:
verify_preflight_purity("prgs")
self.assertIn("Workspace file edits occurred before gitea_whoami", str(ctx.exception))
@patch("subprocess.run")
def test_record_preflight_detects_violation_capability(self, mock_run):
mock_run.return_value = MagicMock(stdout="M mcp_server.py\n")
from mcp_server import record_preflight_check, verify_preflight_purity
record_preflight_check("capability", resolved_role="author")
with open(self.preflight_path, "r") as f:
data = json.load(f)
self.assertTrue(data.get("capability_preflight_violation"))
self.assertTrue(data.get("capability_called"))
self.assertEqual(data.get("role"), "author")
data["whoami_called"] = True
with open(self.preflight_path, "w") as f:
json.dump(data, f)
with self.assertRaises(RuntimeError) as ctx:
verify_preflight_purity("prgs")
self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability", str(ctx.exception))
@patch("subprocess.run")
def test_verify_preflight_reviewer_edits_blocked(self, mock_run):
with open(self.preflight_path, "w") as f:
json.dump({
"whoami_called": True,
"capability_called": True,
"role": "reviewer"
}, f)
mock_run.return_value = MagicMock(stdout="M review_proofs.py\n")
from mcp_server import verify_preflight_purity
with self.assertRaises(RuntimeError) as ctx:
verify_preflight_purity("prgs")
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
@patch("subprocess.run")
def test_verify_preflight_clean_reviewer_allowed(self, mock_run):
with open(self.preflight_path, "w") as f:
json.dump({
"whoami_called": True,
"capability_called": True,
"role": "reviewer"
}, f)
mock_run.return_value = MagicMock(stdout="")
from mcp_server import verify_preflight_purity
verify_preflight_purity("prgs")
def test_verify_preflight_skipped_fails(self):
from mcp_server import verify_preflight_purity
with self.assertRaises(RuntimeError) as ctx:
verify_preflight_purity("prgs")
self.assertIn("verification were skipped", str(ctx.exception))
-62
View File
@@ -203,67 +203,5 @@ class TestResolveTaskCapability(unittest.TestCase):
reasons = res.get("reasons", [])
self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0)
# Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close.
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api):
# Default author profile has PR create/comment but not close: the
# close capability must never be implied by broader author ops.
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
self.assertEqual(res["requested_task"], "close_pr")
self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
self.assertEqual(res["required_role_kind"], "author")
self.assertFalse(res["allowed_in_current_session"])
self.assertTrue(res["stop_required"])
self.assertEqual(res["matching_configured_profile"], [])
self.assertTrue(res["different_mcp_namespace_required"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api):
# Operator-granted close capability resolves cleanly under an author profile.
config = json.loads(json.dumps(CONFIG_RESOLVER))
config["profiles"]["author-closer"] = {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.pr.close"],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
"execution_profile": "author-closer",
}
self._write_config(config)
with patch.dict(os.environ, self._env("author-closer")):
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
self.assertTrue(res["allowed_in_current_session"])
self.assertFalse(res["stop_required"])
self.assertIn("author-closer", res["matching_configured_profile"])
self.assertIn("ready for operations", res["exact_safe_next_action"])
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api):
# close_pr is author-side: a reviewer profile must be told to stop.
with patch.dict(os.environ, self._env("reviewer-profile")):
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
self.assertFalse(res["allowed_in_current_session"])
self.assertTrue(res["stop_required"])
self.assertEqual(res["required_role_kind"], "author")
self.assertIn("author", res["exact_safe_next_action"].lower())
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api):
# close_pr resolves (no Unknown-task error); near-miss spellings keep
# failing closed so no untracked close fallback can be rationalized.
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs")
self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
for unknown in ("close_pull_request", "close", "pr_close"):
with self.assertRaises(ValueError):
mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs")
if __name__ == "__main__":
unittest.main()
+42 -27
View File
@@ -112,23 +112,39 @@ class TestAPIPayload(unittest.TestCase):
class TestMutationAuthorityLock(unittest.TestCase):
"""#199 (refs #194): the CLI refuses to run under a profile that differs
from the session profile lock exported by the launching MCP session."""
"""Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
@patch("review_pr.get_profile")
def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
# An author-bound session exported the lock; the CLI resolves a
# reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
# before any API call.
def test_cli_blocked_on_profile_mismatch(self, mock_get_profile):
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
original_exists = os.path.exists
def conditional_exists(path):
if "gitea_mutation_authority.lock" in str(path):
return True
return original_exists(path)
original_open = open
def conditional_open(file, *args, **kwargs):
if "gitea_mutation_authority.lock" in str(file):
return io.StringIO('{"current_profile": "prgs-author"}')
return original_open(file, *args, **kwargs)
from _pytest.monkeypatch import MonkeyPatch
import io
buf = io.StringIO()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}), \
patch.object(sys, "stderr", buf):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
monkeypatch = MonkeyPatch()
monkeypatch.setattr(sys, "stderr", buf)
with patch("os.path.exists", side_effect=conditional_exists), \
patch("builtins.open", side_effect=conditional_open):
try:
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
finally:
monkeypatch.undo()
self.assertEqual(rc, 3)
msg = buf.getvalue().lower()
self.assertIn("cli override rejected", msg)
@@ -139,22 +155,21 @@ class TestMutationAuthorityLock(unittest.TestCase):
def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
mock_api.side_effect = [FAKE_PR_DATA, {}]
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 0)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_cli_allowed_without_session_lock(self, _auth, mock_api):
# No lock in the environment = direct operator CLI use; the wall
# does not apply and the normal flow proceeds.
mock_api.side_effect = [FAKE_PR_DATA, {}]
env = {k: v for k, v in os.environ.items()
if k != "GITEA_SESSION_PROFILE_LOCK"}
with patch.dict(os.environ, env, clear=True):
original_exists = os.path.exists
def conditional_exists(path):
if "gitea_mutation_authority.lock" in str(path):
return True
return original_exists(path)
original_open = open
def conditional_open(file, *args, **kwargs):
if "gitea_mutation_authority.lock" in str(file):
return io.StringIO('{"current_profile": "prgs-reviewer"}')
return original_open(file, *args, **kwargs)
with patch("os.path.exists", side_effect=conditional_exists), \
patch("builtins.open", side_effect=conditional_open):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
+57 -267
View File
@@ -21,13 +21,10 @@ import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from review_proofs import ( # noqa: E402
assess_capability_evidence,
assess_controller_handoff,
assess_inventory_completeness,
assess_live_state_recheck,
assess_role_boundary,
assess_self_review_contamination,
assess_sweep_evidence,
assess_validation_report,
build_final_report,
pr_inventory_trust_gate,
@@ -119,63 +116,6 @@ def _good_role_boundary():
)
def _good_capability_evidence():
return assess_capability_evidence([
{
"task": "review_pr",
"allowed": True,
"evidence_source": (
"gitea_resolve_task_capability(review_pr) output: "
"allowed_in_current_session=true, profile prgs-reviewer"
),
},
{
"task": "merge_pr",
"allowed": True,
"evidence_source": (
"gitea_resolve_task_capability(merge_pr) output: "
"allowed_in_current_session=true, profile prgs-reviewer"
),
},
])
def _good_sweep(**overrides):
sweep = {
"command": (
"git diff prgs/master...HEAD | grep -inE "
"'password|token|secret|api[_-]?key|authorization|bearer|https?://'"
),
"scope": "full PR diff against prgs/master",
"clean": True,
}
sweep.update(overrides)
return assess_sweep_evidence(sweep)
def _good_live_state(**overrides):
recheck = {
"pr_state": "open",
"pinned_head_sha": PINNED,
"live_head_sha": PINNED,
"pinned_base_ref": "master",
"live_base_ref": "master",
"blocking_change_requests": False,
}
recheck.update(overrides)
return assess_live_state_recheck(recheck)
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
"namespaces_used": ["gitea-reviewer"],
"justification": None,
}
kwargs.update(overrides)
return assess_role_boundary(**kwargs)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -526,9 +466,6 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"capability_evidence": _good_capability_evidence(),
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
}
kwargs.update(overrides)
@@ -773,6 +710,7 @@ class TestControllerHandoff(unittest.TestCase):
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: one PR opened",
"- Workspace mutations: none",
"- Current status: PR open",
"- Blockers: none",
"- Next: review PR #999",
@@ -838,6 +776,40 @@ class TestControllerHandoff(unittest.TestCase):
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_rejects_equivalent_or_multiple_issues(self):
# 1. equivalent reference blocked
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: Issue #194 / #196 equivalent",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_eq, role="author")
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Selected issue", res["missing_fields"])
# 2. multiple issues blocked
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #194, #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_multi, role="author")
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Selected issue", res["missing_fields"])
def test_author_role_rejects_fuzzy_pr_number(self):
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: PR #203 / #204 equivalent",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_pr, role="author")
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("PR number opened", res["missing_fields"])
def test_inventory_role_requires_inventory_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Repositories checked: Gitea-Tools, mcp-control-plane",
@@ -859,6 +831,28 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("assess_controller_handoff", skill)
self.assertIn("issue #182", skill)
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
class TestPRInventoryTrustGate(unittest.TestCase):
"""Issue #194: unit tests for the PR inventory trust gate."""
@@ -942,209 +936,5 @@ class TestPRInventoryTrustGate(unittest.TestCase):
self.assertTrue(res["corroborated"])
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""
def test_evidence_backed_claims_are_proven(self):
result = _good_capability_evidence()
self.assertTrue(result["proven"])
self.assertEqual(result["reasons"], [])
def test_claim_without_evidence_source_is_not_proven(self):
result = assess_capability_evidence([
{"task": "review_pr", "allowed": True, "evidence_source": ""},
])
self.assertFalse(result["proven"])
self.assertTrue(any("evidence" in r.lower() for r in result["reasons"]))
def test_no_claims_at_all_fails_closed(self):
result = assess_capability_evidence([])
self.assertFalse(result["proven"])
def test_disallowed_task_is_not_proven(self):
result = assess_capability_evidence([
{
"task": "merge_pr",
"allowed": False,
"evidence_source": "gitea_resolve_task_capability output",
},
])
self.assertFalse(result["proven"])
class TestSweepEvidence(unittest.TestCase):
"""#179 gap 2: secret/provenance sweep must be exact."""
def test_exact_sweep_is_proven(self):
result = _good_sweep()
self.assertEqual(result["verdict"], "exact")
self.assertTrue(result["proven"])
def test_vague_sweep_without_command_is_downgraded(self):
result = _good_sweep(command="")
self.assertEqual(result["verdict"], "vague")
self.assertFalse(result["proven"])
def test_sweep_without_scope_is_downgraded(self):
result = _good_sweep(scope="")
self.assertEqual(result["verdict"], "vague")
self.assertFalse(result["proven"])
def test_missing_sweep_fails_closed(self):
result = assess_sweep_evidence(None)
self.assertEqual(result["verdict"], "missing")
self.assertFalse(result["proven"])
def test_unstated_result_is_downgraded(self):
result = _good_sweep(clean=None)
self.assertFalse(result["proven"])
class TestLiveStateRecheck(unittest.TestCase):
"""#179 gap 3: explicit pre-mutation live-state recheck."""
def test_clean_recheck_is_proven(self):
result = _good_live_state()
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_missing_recheck_fails_closed(self):
result = assess_live_state_recheck(None)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_closed_pr_blocks(self):
result = _good_live_state(pr_state="closed")
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_moved_head_blocks(self):
result = _good_live_state(live_head_sha=OTHER)
self.assertFalse(result["proven"])
self.assertTrue(any("head" in r.lower() for r in result["reasons"]))
def test_changed_base_blocks(self):
result = _good_live_state(live_base_ref="develop")
self.assertFalse(result["proven"])
def test_unresolved_blocking_reviews_block(self):
result = _good_live_state(blocking_change_requests=True)
self.assertFalse(result["proven"])
def test_unchecked_blocking_state_fails_closed(self):
result = _good_live_state(blocking_change_requests=None)
self.assertFalse(result["proven"])
class TestRoleBoundary179(unittest.TestCase):
"""#179 gap 4: reviewer flows avoid unjustified author-namespace use."""
def test_native_namespace_only_is_clean(self):
result = _good_role_boundary_179()
self.assertTrue(result["proven"])
def test_foreign_namespace_without_justification_is_downgraded(self):
result = _good_role_boundary_179(
namespaces_used=["gitea-reviewer", "gitea-author"]
)
self.assertFalse(result["proven"])
self.assertTrue(any("justif" in r.lower() for r in result["reasons"]))
def test_foreign_namespace_with_justification_is_clean(self):
result = _good_role_boundary_179(
namespaces_used=["gitea-reviewer", "gitea-author"],
justification=(
"author namespace read-only whoami used to evidence "
"self-review contamination status"
),
)
self.assertTrue(result["proven"])
def test_unreported_namespaces_fail_closed(self):
result = _good_role_boundary_179(namespaces_used=None)
self.assertFalse(result["proven"])
class TestFinalReport179Bar(unittest.TestCase):
"""#179 acceptance adds capability, sweep, live-state, and role proofs."""
def _report(self, **overrides):
kwargs = {
"checkout_proof": _good_checkout(),
"inventory": _good_inventory(),
"validation": _good_validation(),
"contamination": _good_contamination(),
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"capability_evidence": _good_capability_evidence(),
"sweep": _good_sweep(),
"live_state": _good_live_state(),
"role_boundary": _good_role_boundary(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
def test_all_179_proofs_present_is_grade_a(self):
report = self._report()
self.assertEqual(report["grade"], "A")
self.assertTrue(report["capability_evidence_proven"])
self.assertEqual(report["sweep_verdict"], "exact")
self.assertTrue(report["live_state_recheck_proven"])
self.assertTrue(report["role_boundary_clean"])
def test_missing_capability_evidence_downgrades(self):
report = self._report(capability_evidence=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["capability_evidence_proven"])
def test_unevidenced_capability_claim_downgrades(self):
report = self._report(
capability_evidence=assess_capability_evidence([
{"task": "review_pr", "allowed": True, "evidence_source": ""},
])
)
self.assertNotEqual(report["grade"], "A")
def test_vague_sweep_downgrades(self):
report = self._report(sweep=_good_sweep(command=""))
self.assertNotEqual(report["grade"], "A")
self.assertEqual(report["sweep_verdict"], "vague")
def test_missing_sweep_downgrades(self):
report = self._report(sweep=None)
self.assertNotEqual(report["grade"], "A")
def test_missing_live_state_recheck_downgrades_and_blocks_merge(self):
report = self._report(live_state=None)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
self.assertFalse(report["live_state_recheck_proven"])
def test_stale_live_state_blocks_merge(self):
report = self._report(live_state=_good_live_state(live_head_sha=OTHER))
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
def test_merge_claim_without_live_recheck_is_a_violation(self):
report = self._report(live_state=None, merge_performed=True)
self.assertEqual(report["grade"], "blocked")
self.assertTrue(report["violations"])
def test_unjustified_author_namespace_downgrades(self):
report = self._report(
role_boundary=_good_role_boundary_179(
namespaces_used=["gitea-reviewer", "gitea-author"]
)
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["role_boundary_clean"])
def test_positive_baseline_from_173_still_holds(self):
report = self._report()
self.assertTrue(report["inventory_complete"])
self.assertTrue(report["validated_on_pinned_head"])
if __name__ == "__main__":
unittest.main()
-189
View File
@@ -1,189 +0,0 @@
"""Tests for pre-task role/session router (#206)."""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_config
import mcp_server
import role_session_router
from review_proofs import assess_role_route_handoff
CONFIG = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
}
},
"profiles": {
"prgs-author": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "jcwalker3",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.pr.create",
"gitea.branch.push", "gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
],
"execution_profile": "prgs-author",
},
"prgs-reviewer": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "sysadmin",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": [
"gitea.read", "gitea.pr.review", "gitea.pr.approve",
"gitea.pr.merge", "gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
],
"execution_profile": "prgs-reviewer",
},
},
"rules": {"allow_runtime_switching": False},
}
class TestRoleSessionRouter(unittest.TestCase):
def setUp(self):
role_session_router.clear_route_state()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {
"host": "gitea.example.com",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG))
def tearDown(self):
self._remotes.stop()
role_session_router.clear_route_state()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir.cleanup()
def _env(self, profile):
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
}
def test_reviewer_task_under_author_profile_wrong_role_stop(self):
with patch.dict(os.environ, self._env("prgs-author")):
route = mcp_server.gitea_route_task_session(
task_type="review_pr", remote="prgs"
)
self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
self.assertFalse(route["downstream_allowed"])
self.assertIn("Wrong role/session for reviewer task", route["message"])
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_issue_creation_blocked_after_reviewer_wrong_role_stop(
self, _auth, mock_api
):
with patch.dict(os.environ, self._env("prgs-author")):
mcp_server.gitea_route_task_session(task_type="review_pr", remote="prgs")
result = mcp_server.gitea_create_issue(
title="process issue fallback",
body="should not post",
remote="prgs",
)
self.assertFalse(result.get("success", True))
self.assertFalse(result.get("performed", True))
issue_posts = [
c for c in mock_api.call_args_list
if c.args[0] == "POST" and str(c.args[1]).endswith("/issues")
]
self.assertEqual(issue_posts, [])
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_request", return_value={"number": 999})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_author_task_allowed_after_explicit_author_route(
self, _auth, _api, _get_all
):
with patch.dict(os.environ, self._env("prgs-author")):
route = mcp_server.gitea_route_task_session(
task_type="create_issue", remote="prgs"
)
self.assertEqual(
route["route_result"], role_session_router.ROUTE_ALLOWED
)
result = mcp_server.gitea_create_issue(
title="legit author task",
remote="prgs",
)
self.assertEqual(result.get("number"), 999)
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_reviewer_task_under_reviewer_profile_allowed(self, _auth, _api):
with patch.dict(os.environ, self._env("prgs-reviewer")):
route = mcp_server.gitea_route_task_session(
task_type="review_pr", remote="prgs"
)
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
self.assertTrue(route["downstream_allowed"])
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
with patch.dict(os.environ, self._env("prgs-reviewer")):
route = mcp_server.gitea_route_task_session(
task_type="create_issue", remote="prgs"
)
self.assertEqual(
route["route_result"], role_session_router.ROUTE_TO_AUTHOR
)
self.assertFalse(route["downstream_allowed"])
def test_activate_profile_blocked_in_static_mode(self):
with patch.dict(os.environ, self._env("prgs-author")):
result = mcp_server.gitea_activate_profile(
profile_name="prgs-reviewer", remote="prgs"
)
self.assertFalse(result["success"])
self.assertIn("switching is disabled", result["message"])
def test_handoff_requires_route_fields(self):
incomplete = assess_role_route_handoff("Task done.")
self.assertFalse(incomplete["complete"])
self.assertIn("Task type", incomplete["missing_fields"])
complete = assess_role_route_handoff(
"\n".join([
"Task type: review_pr",
"Required role: reviewer",
"Active role: author",
"Route result: wrong_role_stop",
]),
route_result={"route_result": "wrong_role_stop"},
)
self.assertTrue(complete["complete"])
if __name__ == "__main__":
unittest.main()
+33 -5
View File
@@ -14,11 +14,39 @@ BRANCHES = REPO / "branches"
def run(script, *args):
proc = subprocess.run(
["bash", str(SCRIPTS / script), *args],
capture_output=True, text=True, cwd=str(REPO),
)
return proc.returncode, proc.stdout, proc.stderr
branch = None
for arg in args:
if not arg.startswith("-"):
branch = arg
break
lock_file = Path("/tmp/gitea_issue_lock.json")
created_lock = False
if script == "worktree-start" and branch:
import re
import json
m = re.search(r"issue-(\d+)", branch)
if not m:
m = re.search(r"pr-(\d+)", branch)
issue_num = int(m.group(1)) if m else 999
lock_file.write_text(json.dumps({
"issue_number": issue_num,
"branch_name": branch,
"remote": "prgs",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools"
}), encoding="utf-8")
created_lock = True
try:
proc = subprocess.run(
["bash", str(SCRIPTS / script), *args],
capture_output=True, text=True, cwd=str(REPO),
)
return proc.returncode, proc.stdout, proc.stderr
finally:
if created_lock and lock_file.exists():
lock_file.unlink()
class TestWorktreeStart(unittest.TestCase):