feat: block reviewer local Gitea fallbacks during normal review (Closes #324)
Add assess_reviewer_fallback_report to fail closed when reviewer final reports or action logs show profiles.json access or local helper scripts while MCP tools are available. Recovery mode requires explicit fallback classification and identity/profile/repo/capability proof. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -3099,3 +3099,10 @@ def assess_email_disclosure(
|
||||
for email in emails
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
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)
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
),
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user