diff --git a/review_proofs.py b/review_proofs.py index 02fa543..29f3d88 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3444,6 +3444,13 @@ def assess_email_disclosure( } +def assess_reviewer_fallback_report(report_text, **kwargs): + """#324: block local Gitea fallbacks during normal reviewer workflows.""" + from reviewer_fallback import assess_reviewer_fallback_report as _assess + + return _assess(report_text, **kwargs) + + _QUEUE_STATUS_REPORT_HINT = re.compile( r"queue[- ]status|selected pr:\s*none|no pr selected|" r"queue[- ]status[- ]only", diff --git a/reviewer_fallback.py b/reviewer_fallback.py new file mode 100644 index 0000000..9b8ef6c --- /dev/null +++ b/reviewer_fallback.py @@ -0,0 +1,213 @@ +"""Reviewer local-fallback detection for normal PR review workflows (#324). + +Normal reviewer runs must use MCP tools. Reading profile secret files or +running local Gitea helper scripts while MCP is available is a fail-closed +violation. Explicit recovery mode may use local fallback only with full proof. +""" + +from __future__ import annotations + +import re +from typing import Any + +LOCAL_GITEA_SCRIPT_NAMES = ( + "list_prs.py", + "view_pr.py", + "create_pr.py", + "merge_pr.py", + "edit_pr.py", + "list_issues.py", + "delete_branch.py", + "create_issue.py", + "close_issue.py", + "review_pr.py", + "mark_issue.py", + "mirror_refs.sh", +) + +PROFILE_SECRET_MARKERS = ( + "profiles.json", + ".config/gitea-tools/profiles", + "gitea_auth.py", + "gitea_config.py", + "keychain", + "credential fill", + "token store", +) + +_RECOVERY_MODE_RE = re.compile( + r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|" + r"mcp tools unavailable|no mcp path)\b", + re.IGNORECASE, +) +_FALLBACK_CLASSIFICATION_RE = re.compile( + r"\b(?:local fallback|fallback mode|used local gitea|ran local script)\b", + re.IGNORECASE, +) +_MCP_TOOL_USE_RE = re.compile( + r"\b(?:gitea_list_prs|gitea_view_pr|gitea_review_pr|gitea_merge_pr|" + r"gitea_resolve_task_capability|gitea_whoami|mcp tool)\b", + re.IGNORECASE, +) +_LOCAL_SCRIPT_RE = re.compile( + r"(?:^|[\s\"'`/])(?:" + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + r")", + re.IGNORECASE, +) +_PROFILE_ACCESS_RE = re.compile( + r"(?:read|open|inspect|cat|view|access(?:ed)?|loaded?)\s+(?:file\s+)?[`'\"]?" + r"[^`'\"]*profiles\.json|profiles\.json[`'\"]?\s+(?:read|opened|inspected|accessed)|" + r"~/?\.config/gitea-tools/profiles", + re.IGNORECASE, +) +_LOCAL_ENV_SCRIPT_RE = re.compile( + r"GITEA_MCP_(?:CONFIG|PROFILE)=.*\b(?:python\s+)?(?:list_prs|view_pr|create_pr|merge_pr)\.py", + re.IGNORECASE, +) +_RECOVERY_PROOF_FIELDS = ( + ("identity proof", ("identity proof", "exact identity proof")), + ("profile proof", ("profile proof", "exact profile proof")), + ("repo proof", ("repo proof", "exact repo proof")), + ("capability proof", ("capability proof", "exact capability proof")), + ("mcp unavailable reason", ( + "why mcp was unavailable", + "mcp unavailable", + "mcp unavailability", + )), +) + + +def _collect_observed_text( + report_text: str, + action_log: list[dict] | None, +) -> str: + chunks = [report_text or ""] + for entry in action_log or []: + for key in ("command", "action", "detail", "path", "script"): + value = entry.get(key) + if value: + chunks.append(str(value)) + return "\n".join(chunks) + + +def _detect_profile_secret_access(text: str) -> list[str]: + reasons = [] + lower = text.lower() + if _PROFILE_ACCESS_RE.search(text): + reasons.append("report or action log shows profiles.json/profile secret access") + for marker in PROFILE_SECRET_MARKERS: + if marker == "profiles.json": + continue + if marker in lower and "do not" not in lower and "must not" not in lower: + if any(verb in lower for verb in ("read ", "open ", "inspect ", "cat ", "loaded ")): + reasons.append(f"profile secret surface '{marker}' accessed during review") + return reasons + + +def _detect_local_script_use(text: str) -> list[str]: + reasons = [] + match = _LOCAL_SCRIPT_RE.search(text) + if match: + reasons.append( + f"local Gitea helper script invoked ({match.group(0).strip()})" + ) + if _LOCAL_ENV_SCRIPT_RE.search(text): + reasons.append( + "local Gitea script run with GITEA_MCP_CONFIG/GITEA_MCP_PROFILE env overrides" + ) + return reasons + + +def _recovery_proof_missing(text: str) -> list[str]: + lower = text.lower() + missing = [] + for label, aliases in _RECOVERY_PROOF_FIELDS: + if not any(alias in lower for alias in aliases): + missing.append(label) + if not _FALLBACK_CLASSIFICATION_RE.search(text): + missing.append("fallback classification") + if not _RECOVERY_MODE_RE.search(text): + missing.append("recovery mode declaration") + return missing + + +def assess_reviewer_fallback_report( + report_text: str, + *, + action_log: list[dict] | None = None, + recovery_mode: bool = False, + mcp_available: bool = True, + mcp_tools_used: bool | None = None, +) -> dict[str, Any]: + """Fail closed when normal reviewer workflows use local Gitea fallbacks (#324).""" + text = _collect_observed_text(report_text, action_log) + lower = (report_text or "").lower() + + profile_violations = _detect_profile_secret_access(text) + script_violations = _detect_local_script_use(text) + violations = profile_violations + script_violations + + if mcp_tools_used is None: + mcp_tools_used = bool(_MCP_TOOL_USE_RE.search(report_text or "")) + elif mcp_tools_used is False and _MCP_TOOL_USE_RE.search(report_text or ""): + mcp_tools_used = True + + reasons: list[str] = [] + recovery_declared = recovery_mode or bool(_RECOVERY_MODE_RE.search(report_text or "")) + + if violations and mcp_available and not recovery_declared: + reasons.extend(violations) + if mcp_tools_used: + reasons.append( + "MCP tools were available and used; local fallback/profile access is forbidden" + ) + else: + reasons.append( + "MCP path is available; normal review must not use local Gitea fallbacks" + ) + + if violations and recovery_declared: + missing = _recovery_proof_missing(report_text or "") + if missing: + reasons.append( + "recovery-mode fallback missing proof fields: " + + ", ".join(missing) + ) + + if ( + not violations + and recovery_declared + and _FALLBACK_CLASSIFICATION_RE.search(report_text or "") + and not recovery_mode + ): + missing = _recovery_proof_missing(report_text or "") + if missing: + reasons.append( + "report claims local fallback but recovery proof is incomplete: " + + ", ".join(missing) + ) + + proven = not reasons + blocked = bool(reasons) and not recovery_declared or any( + "recovery-mode fallback missing" in r or "recovery proof is incomplete" in r + for r in reasons + ) + return { + "proven": proven, + "block": blocked or (bool(reasons) and mcp_available and not recovery_declared), + "downgraded": bool(reasons) and not blocked, + "recovery_mode": recovery_declared, + "mcp_available": mcp_available, + "mcp_tools_used": mcp_tools_used, + "profile_violations": profile_violations, + "script_violations": script_violations, + "reasons": reasons, + "safe_next_action": ( + "use MCP reviewer tools only; do not read profiles.json or run local Gitea scripts" + if reasons and not recovery_declared + else ( + "complete recovery-mode fallback proof before claiming local fallback" + if reasons + else "proceed with MCP tools" + ) + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 7e0fe23..0481ed5 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -118,6 +118,12 @@ def test_start_issue_template_references_work_issue_workflow(): assert "workflows/work-issue.md" in text +def test_reviewer_fallback_verifier_exported(): + from review_proofs import assess_reviewer_fallback_report + + assert callable(assess_reviewer_fallback_report) + + def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report diff --git a/tests/test_reviewer_fallback.py b/tests/test_reviewer_fallback.py new file mode 100644 index 0000000..59fce4e --- /dev/null +++ b/tests/test_reviewer_fallback.py @@ -0,0 +1,112 @@ +"""Tests for reviewer local-fallback verifier (#324).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_fallback import assess_reviewer_fallback_report # noqa: E402 + + +class TestReviewerFallbackNormalMode(unittest.TestCase): + def test_clean_mcp_report_passes(self): + report = ( + "Used gitea_list_prs and gitea_view_pr via MCP reviewer namespace.\n" + "Capability evidence: gitea_resolve_task_capability(review_pr)." + ) + result = assess_reviewer_fallback_report(report, mcp_available=True) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_profiles_json_access_blocks(self): + report = ( + "Inspected ~/.config/gitea-tools/profiles.json for reviewer credentials.\n" + "Then used gitea_view_pr." + ) + result = assess_reviewer_fallback_report(report, mcp_available=True) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("profiles.json" in r for r in result["reasons"])) + + def test_local_list_prs_script_blocks_when_mcp_available(self): + report = ( + "Ran `python list_prs.py --remote prgs` with GITEA_MCP_CONFIG set.\n" + "MCP tools were also available." + ) + result = assess_reviewer_fallback_report( + report, + mcp_available=True, + action_log=[{"command": "GITEA_MCP_PROFILE=prgs-reviewer python list_prs.py"}], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(result["script_violations"]) + + def test_action_log_profile_access_blocks(self): + result = assess_reviewer_fallback_report( + "Review complete via MCP.", + action_log=[{"path": "/Users/me/.config/gitea-tools/profiles.json", "action": "read"}], + mcp_available=True, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + +class TestReviewerFallbackRecoveryMode(unittest.TestCase): + def _recovery_report(self, **extra): + base = "\n".join([ + "Recovery mode: MCP tools unavailable in this session.", + "Local fallback: ran python view_pr.py after capability proof.", + "Identity proof: sysadmin / prgs-reviewer", + "Profile proof: prgs-reviewer", + "Repo proof: Scaled-Tech-Consulting/Gitea-Tools", + "Capability proof: gitea.pr.review allowed for recovery command", + "Why MCP was unavailable: reviewer MCP namespace failed to start", + ]) + return base + ("\n" + extra.get("append", "") if extra.get("append") else "") + + def test_recovery_fallback_with_proof_passes(self): + result = assess_reviewer_fallback_report( + self._recovery_report(), + action_log=[{"command": "python view_pr.py --remote prgs"}], + mcp_available=False, + recovery_mode=True, + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_recovery_fallback_without_proof_blocks(self): + report = ( + "Recovery mode engaged.\n" + "Local fallback: python list_prs.py\n" + ) + result = assess_reviewer_fallback_report( + report, + mcp_available=False, + recovery_mode=True, + action_log=[{"command": "python list_prs.py"}], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue( + any("recovery" in r.lower() for r in result["reasons"]) + ) + + def test_mcp_unavailable_allows_recovery_attempt(self): + result = assess_reviewer_fallback_report( + self._recovery_report(), + mcp_available=False, + recovery_mode=True, + ) + self.assertTrue(result["proven"]) + + +class TestReviewerFallbackExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_reviewer_fallback_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file