diff --git a/docs/gitea-execution-profiles.md b/docs/gitea-execution-profiles.md index 61324bd..e4ec36e 100644 --- a/docs/gitea-execution-profiles.md +++ b/docs/gitea-execution-profiles.md @@ -203,6 +203,29 @@ 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: diff --git a/mcp_server.py b/mcp_server.py index 164cb1f..21a6fa5 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -17,10 +17,135 @@ import json import os import re import sys +import json import functools 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 + +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 = { + "initial_profile": profile_name, + "initial_identity": identity, + "current_profile": profile_name, + "current_identity": identity, + "remote": remote, + "task": task, + "role_pivot_authorized": False, + "role_pivot_record": None, + "pid": os.getpid(), + } + + +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 + + if data.get("remote") != remote: + raise RuntimeError( + f"Mutation remote '{remote}' does not match locked remote " + f"'{data.get('remote')}' (fail closed)" + ) + + 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)" + ) + + # 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()): + if not data.get("role_pivot_authorized"): + raise RuntimeError( + "Attempted reviewer mutation from author session without " + "authorized role pivot (fail closed)" + ) + # Resolve the project root. MCP clients must launch this script directly with # the venv interpreter (venv/bin/python3) — see the config example above. We do # NOT os.execv() to re-point the interpreter: replacing the process after the @@ -46,6 +171,7 @@ from gitea_auth import ( # noqa: E402 ) import gitea_audit # noqa: E402 import gitea_config # noqa: E402 +import role_session_router # noqa: E402 def _reveal_endpoints() -> bool: @@ -325,6 +451,16 @@ def gitea_create_issue( 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) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/issues" @@ -957,7 +1093,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 = _issue_comment_gate("gitea.read") + reasons = _profile_operation_gate("gitea.read") if reasons: return { "success": False, @@ -1133,6 +1269,18 @@ def _evaluate_pr_review_submission( ) 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: auth = _auth(h) @@ -1289,11 +1437,20 @@ 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'. + state: New state — 'open' or 'closed'. 'closed' requires the + ``gitea.pr.close`` capability. base: Target branch name. remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. @@ -1301,7 +1458,10 @@ def gitea_edit_pr( repo: Override the repository name. Returns: - dict with success status and details of the edited PR. + 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. """ # Validate inputs BEFORE any auth/profile resolution or API setup: a # no-fields call is a pure validation error and must not depend on @@ -1312,6 +1472,9 @@ 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 @@ -1319,12 +1482,33 @@ 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}" - with _audited("edit_pr", host=h, remote=remote, org=o, repo=r, - pr_number=pr_number, request_metadata={"fields": sorted(payload)}): + 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): data = api_request("PATCH", url, auth, payload) cleanup_status = None @@ -1605,6 +1789,17 @@ 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) @@ -2085,13 +2280,14 @@ def _permission_block_report(required_operation: str, return report -def _issue_comment_gate(op: str) -> list[str]: - """Profile permission check for issue-comment tools (#126). +def _profile_operation_gate(op: str) -> list[str]: + """Profile permission check for a single gated operation (#126, #216). Issue discussion comments are gated separately from the gitea.pr.* review/merge family: listing requires ``gitea.read``, creating requires - ``gitea.issue.comment``. Returns a list of block reasons (empty = allowed); - an unreadable profile fails closed. + ``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. """ try: profile = get_profile() @@ -2143,7 +2339,7 @@ def gitea_list_issue_comments( 'success' False, 'reasons', and a structured 'permission_report' (#142) with no API call made. """ - reasons = _issue_comment_gate("gitea.read") + reasons = _profile_operation_gate("gitea.read") if reasons: return {"success": False, "issue_number": issue_number, "reasons": reasons, @@ -2205,7 +2401,7 @@ def gitea_create_issue_comment( (permission blocks also carry a structured 'permission_report', #142). """ - gate_reasons = _issue_comment_gate("gitea.issue.comment") + gate_reasons = _profile_operation_gate("gitea.issue.comment") reasons = list(gate_reasons) if not (body or "").strip(): reasons.append("comment body must be a non-empty string") @@ -3203,6 +3399,22 @@ 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 + # 5. Audit the switch if auditing is on _audit( "activate_profile", @@ -3520,6 +3732,60 @@ 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, @@ -3569,6 +3835,14 @@ 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", @@ -3581,6 +3855,18 @@ 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", @@ -3698,6 +3984,8 @@ def gitea_resolve_task_capability( task, ) + record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task) + return { "requested_task": task, "required_operation_permission": required_permission, @@ -3718,4 +4006,8 @@ def gitea_resolve_task_capability( # ── 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") diff --git a/review_pr.py b/review_pr.py index 464d348..b4fe4dc 100755 --- a/review_pr.py +++ b/review_pr.py @@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3") if os.path.exists(venv_python) and sys.executable != venv_python: os.execv(venv_python, [venv_python] + sys.argv) -from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url +from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile def main(argv=None): @@ -60,6 +60,32 @@ 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: + try: + cli_profile = (get_profile().get("profile_name") or "").strip() + 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: if args.body_file == "-": diff --git a/review_proofs.py b/review_proofs.py index 43b831d..90f48c5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -330,22 +330,167 @@ def assess_self_review_contamination(reviewer_identity, pr_author, } -def assess_role_boundary(proof): +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 "" + 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): """Assess reviewer/author role separation for blind queue workflows. Issue #175 blocks a reviewer queue task from silently becoming author - 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. + implementation work. Issue #179 also requires reviewer workflows to + report namespace use and justify any foreign namespace calls. This helper + accepts both forms: - *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``. + - the #175 dict proof with mutation details, or + - the #179 keyword form: ``task_role``, ``namespaces_used``, + ``justification``. """ - proof = proof or {} + 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 {}) + 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 []) @@ -364,6 +509,8 @@ def assess_role_boundary(proof): 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: @@ -409,12 +556,27 @@ def assess_role_boundary(proof): 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), } @@ -468,7 +630,9 @@ def assess_review_mutation_final_report(report_text, review_decision_lock): def build_final_report(checkout_proof, inventory, validation, contamination, identity_eligible, merge_performed, - issue_status_verified, role_boundary=None): + issue_status_verified, + capability_evidence=None, sweep=None, live_state=None, + role_boundary=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -479,6 +643,13 @@ 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")) @@ -491,6 +662,29 @@ 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") @@ -514,6 +708,23 @@ 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 @@ -522,6 +733,8 @@ 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 = [] @@ -556,6 +769,10 @@ 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, } @@ -683,6 +900,43 @@ 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": [], + } + + # ── PR Inventory Trust Gate (Issue #194) ────────────────────────────────────── # # A reviewer agent may not convert an empty PR list response into a definitive diff --git a/role_session_router.py b/role_session_router.py new file mode 100644 index 0000000..b9a0406 --- /dev/null +++ b/role_session_router.py @@ -0,0 +1,195 @@ +"""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, [] \ No newline at end of file diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 0b8aa1f..c24670c 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -241,19 +241,35 @@ Worktree folder = branch with `/` replaced by `-` (`review_proofs.assess_review_mutation_final_report`) unless an operator-approved correction flow was invoked and explained. 10. **Do not merge if checks fail. Do not merge if the reviewer is the author.** -11. The final report must distinguish (`review_proofs.build_final_report`): +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`): identity eligible; PR author different from reviewer; session - 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. + 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. ## G. Merge / cleanup workflow Only an eligible (non-author) reviewer merges. Before merging: always verify -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 +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 **expected changed-file set**, so a moved head or widened diff refuses the merge. After a real merge: diff --git a/skills/llm-project-workflow/templates/merge-pr.md b/skills/llm-project-workflow/templates/merge-pr.md index 4155fb6..bee3b75 100644 --- a/skills/llm-project-workflow/templates/merge-pr.md +++ b/skills/llm-project-workflow/templates/merge-pr.md @@ -20,10 +20,21 @@ 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 #: 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. If any gate fails → STOP and report. -4. Merge with explicit confirmation (e.g. confirmation="MERGE 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). +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 "), + 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). *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): diff --git a/skills/llm-project-workflow/templates/review-pr.md b/skills/llm-project-workflow/templates/review-pr.md index 8485d6f..3a2dfae 100644 --- a/skills/llm-project-workflow/templates/review-pr.md +++ b/skills/llm-project-workflow/templates/review-pr.md @@ -36,6 +36,11 @@ 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). @@ -58,12 +63,23 @@ 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 #; 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 ''`); "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. 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. 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. Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md): Review Metadata: diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e08e2df --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,28 @@ +"""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", {}) + yield diff --git a/tests/test_close_pr_capability.py b/tests/test_close_pr_capability.py new file mode 100644 index 0000000..8a2385e --- /dev/null +++ b/tests/test_close_pr_capability.py @@ -0,0 +1,154 @@ +"""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() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 6ac2f9f..66158b0 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -42,6 +42,8 @@ from mcp_server import ( # noqa: E402 from gitea_auth import get_profile # noqa: E402 import gitea_config # noqa: E402 +import mcp_server + FAKE_AUTH = "Basic dGVzdDp0ZXN0" @@ -1776,7 +1778,8 @@ 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() - patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).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() def tearDown(self): patch.stopall() @@ -2451,3 +2454,122 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase): "gitea.issue.comment", reviewer["allowed_operations"], reviewer.get("forbidden_operations", [])) self.assertTrue(ok) + + +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. + """ + + 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" + + def tearDown(self): + self.patch_profile.stop() + self.patch_username.stop() + + 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 + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_mutation_authority("prgs") + self.assertIn("profile unresolved", str(ctx.exception)) + + def test_mismatched_remote_fails(self): + self._authority(remote="dadeschools") + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_mutation_authority("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", + ) + with self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_mutation_authority("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 self.assertRaises(RuntimeError) as ctx: + mcp_server.verify_mutation_authority("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") diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 358febe..991e141 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -203,5 +203,67 @@ 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() diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py index d28ff01..1344745 100644 --- a/tests/test_review_pr.py +++ b/tests/test_review_pr.py @@ -2,6 +2,8 @@ Mocks api_request and credentials. """ +import io +import os import sys import unittest from unittest.mock import patch @@ -27,6 +29,11 @@ FAKE_PR_DATA = { class TestArgParsing(unittest.TestCase): + def setUp(self): + self.exists_patcher = patch("os.path.exists", return_value=False) + self.exists_patcher.start() + self.addCleanup(self.exists_patcher.stop) + @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) def test_missing_pr_number_exits(self, _auth): with self.assertRaises(SystemExit): @@ -35,6 +42,11 @@ class TestArgParsing(unittest.TestCase): class TestAPIPayload(unittest.TestCase): + def setUp(self): + self.exists_patcher = patch("os.path.exists", return_value=False) + self.exists_patcher.start() + self.addCleanup(self.exists_patcher.stop) + @patch("review_pr.api_request") @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) def test_payload_fields_and_workflow(self, _auth, mock_api): @@ -99,5 +111,55 @@ class TestAPIPayload(unittest.TestCase): self.assertIn("gitea_merge_pr", msg) +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.""" + + @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. + mock_get_profile.return_value = {"profile_name": "prgs-reviewer"} + 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", + ]) + self.assertEqual(rc, 3) + msg = buf.getvalue().lower() + self.assertIn("cli override rejected", msg) + + @patch("review_pr.get_profile") + @patch("review_pr.api_request") + @patch("review_pr.get_auth_header", return_value=FAKE_CREDS) + 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): + rc = review_pr.main([ + "--pr-number", "81", "--event", "APPROVE", + ]) + self.assertEqual(rc, 0) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 8bb9577..da712bc 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -21,11 +21,14 @@ 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_review_mutation_final_report, assess_role_boundary, assess_self_review_contamination, + assess_sweep_evidence, assess_validation_report, build_final_report, pr_inventory_trust_gate, @@ -117,6 +120,63 @@ 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.""" @@ -467,6 +527,9 @@ 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) @@ -929,5 +992,209 @@ 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() diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py new file mode 100644 index 0000000..3ba3db6 --- /dev/null +++ b/tests/test_role_session_router.py @@ -0,0 +1,188 @@ +"""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_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 + ): + 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() \ No newline at end of file