213 lines
7.1 KiB
Python
213 lines
7.1 KiB
Python
"""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"
|
|
)
|
|
),
|
|
} |