feat: block reviewer local Gitea fallbacks during normal review (Closes #324)
This commit is contained in:
@@ -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",
|
||||
|
||||
@@ -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