sysadmin
57b2023611
#530's remote/repo guard blocked mcp_get_control_plane_guide(remote=prgs) because the bare default resolves to Timesheet while local git is Gitea-Tools, and remediation told callers to pass org/repo on a tool that did not accept those parameters. - Accept optional org/repo on mcp_get_control_plane_guide - Prefer local git remote org/repo when bare defaults mismatch (read-only guide) - Keep mutation tools fail-closed on bare mismatch - Align remediation text with tools that accept org/repo Closes #588
123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
"""Remote/repo mismatch guard (#530).
|
|
|
|
Bare ``remote`` names resolve to a default ``org``/``repo`` via the ``REMOTES``
|
|
table in :mod:`gitea_auth`. For the ``prgs`` instance the hardcoded default repo
|
|
is ``Timesheet``, but the tools in this project operate on
|
|
``Scaled-Tech-Consulting/Gitea-Tools``. When a session runs inside a Gitea-Tools
|
|
worktree and calls a tool with a bare ``remote=prgs`` (no explicit ``org``/``repo``),
|
|
the resolved target silently points at the wrong repository, producing false 404s
|
|
and risking mutation of a different repo.
|
|
|
|
This module provides a pure assessment that compares the MCP-resolved ``org/repo``
|
|
against the local git remote URL and fails closed on a genuine mismatch, unless the
|
|
caller supplied explicit ``org``/``repo`` (in which case their intent is authoritative)
|
|
or the local remote URL is unavailable (best-effort corroboration only).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
REMEDIATION = (
|
|
"Pass explicit org= and repo= matching the local git remote on tools that "
|
|
"accept those parameters (including mcp_get_control_plane_guide), "
|
|
"e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools. "
|
|
"Do not pass org/repo to tools whose schema does not list them."
|
|
)
|
|
|
|
# https://host/org/repo.git or git@host:org/repo.git
|
|
_REMOTE_URL_SLUG_RE = re.compile(
|
|
r"(?:[:/])(?P<org>[^/]+)/(?P<repo>[^/]+?)(?:\.git)?/*$"
|
|
)
|
|
|
|
|
|
def parse_org_repo_from_remote_url(remote_url: str | None) -> tuple[str, str] | None:
|
|
"""Best-effort parse of org/repo from a git remote URL."""
|
|
url = (remote_url or "").strip()
|
|
if not url:
|
|
return None
|
|
match = _REMOTE_URL_SLUG_RE.search(url)
|
|
if not match:
|
|
return None
|
|
org = (match.group("org") or "").strip()
|
|
repo = (match.group("repo") or "").strip()
|
|
if not org or not repo:
|
|
return None
|
|
return org, repo
|
|
|
|
|
|
def assess_remote_repo_match(
|
|
*,
|
|
remote: str,
|
|
resolved_org: str,
|
|
resolved_repo: str,
|
|
local_remote_url: str | None,
|
|
org_explicit: bool,
|
|
repo_explicit: bool,
|
|
) -> dict:
|
|
"""Fail closed when the resolved org/repo disagrees with the local git remote.
|
|
|
|
The guard is intentionally conservative:
|
|
|
|
* When the caller passed both ``org`` and ``repo`` explicitly, their intent is
|
|
authoritative and the guard never blocks.
|
|
* When the local git remote URL is unavailable (``None``/empty), corroboration
|
|
is impossible, so the guard does not block (best-effort only).
|
|
* Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL
|
|
(case-insensitive); if it does not, the guard blocks.
|
|
"""
|
|
reasons: list[str] = []
|
|
|
|
if org_explicit and repo_explicit:
|
|
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
|
|
|
|
url = (local_remote_url or "").strip()
|
|
if not url:
|
|
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
|
|
|
|
expected_slug = f"{resolved_org}/{resolved_repo}".lower()
|
|
if expected_slug in url.lower():
|
|
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
|
|
|
|
reasons.append(
|
|
f"MCP-resolved repository '{resolved_org}/{resolved_repo}' for remote "
|
|
f"'{remote}' does not match the local git remote URL '{url}'"
|
|
)
|
|
return _assessment(False, reasons, remote, resolved_org, resolved_repo, local_remote_url)
|
|
|
|
|
|
def format_remote_repo_guard_error(assessment: dict) -> str:
|
|
"""Single RuntimeError message for the MCP resolver gate."""
|
|
reasons = "; ".join(
|
|
assessment.get("reasons") or ["remote/repo resolution mismatch"]
|
|
)
|
|
resolved = (
|
|
f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
|
|
)
|
|
local = assessment.get("local_remote_url") or "(unknown)"
|
|
return (
|
|
f"Remote/repo guard (#530): {reasons}. "
|
|
f"Resolved target: {resolved}; local git remote: {local}. "
|
|
f"{REMEDIATION}"
|
|
)
|
|
|
|
|
|
def _assessment(
|
|
proven: bool,
|
|
reasons: list[str],
|
|
remote: str,
|
|
resolved_org: str,
|
|
resolved_repo: str,
|
|
local_remote_url: str | None,
|
|
) -> dict:
|
|
return {
|
|
"proven": proven,
|
|
"block": not proven,
|
|
"reasons": reasons,
|
|
"remote": remote,
|
|
"resolved_org": resolved_org,
|
|
"resolved_repo": resolved_repo,
|
|
"local_remote_url": local_remote_url,
|
|
"remediation": REMEDIATION,
|
|
}
|