fix: guard remote/repo mismatch vs local git remote (Closes #530) #555
+43
-4
@@ -748,6 +748,7 @@ import merge_approval_gate # noqa: E402
|
||||
import already_landed_reconcile # noqa: E402
|
||||
import author_mutation_worktree # noqa: E402
|
||||
import root_checkout_guard # noqa: E402
|
||||
import remote_repo_guard # noqa: E402
|
||||
import issue_claim_heartbeat # noqa: E402
|
||||
import issue_work_duplicate_gate # noqa: E402
|
||||
import reviewer_pr_lease # noqa: E402
|
||||
@@ -1162,11 +1163,49 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
|
||||
if remote not in REMOTES:
|
||||
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
|
||||
profile = REMOTES[remote]
|
||||
return (
|
||||
host or profile["host"],
|
||||
org or profile["org"],
|
||||
repo or profile["repo"],
|
||||
resolved_host = host or profile["host"]
|
||||
resolved_org = org or profile["org"]
|
||||
resolved_repo = repo or profile["repo"]
|
||||
_enforce_remote_repo_guard(
|
||||
remote,
|
||||
resolved_org,
|
||||
resolved_repo,
|
||||
org_explicit=org is not None,
|
||||
repo_explicit=repo is not None,
|
||||
)
|
||||
return (resolved_host, resolved_org, resolved_repo)
|
||||
|
||||
|
||||
def _enforce_remote_repo_guard(
|
||||
remote: str,
|
||||
resolved_org: str,
|
||||
resolved_repo: str,
|
||||
*,
|
||||
org_explicit: bool,
|
||||
repo_explicit: bool,
|
||||
) -> None:
|
||||
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530).
|
||||
|
||||
Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
|
||||
set, so the unit suite (which calls tools with bare remotes against mocked APIs)
|
||||
is unaffected. In production it protects every read/lookup/mutation tool because
|
||||
they all resolve targets through :func:`_resolve`.
|
||||
"""
|
||||
if "pytest" in sys.modules and not os.environ.get(
|
||||
"GITEA_FORCE_REMOTE_REPO_CHECK"
|
||||
):
|
||||
return
|
||||
local_remote_url = _local_git_remote_url(remote)
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote=remote,
|
||||
resolved_org=resolved_org,
|
||||
resolved_repo=resolved_repo,
|
||||
local_remote_url=local_remote_url,
|
||||
org_explicit=org_explicit,
|
||||
repo_explicit=repo_explicit,
|
||||
)
|
||||
if assessment["block"]:
|
||||
raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
|
||||
|
||||
|
||||
def _auth(host: str) -> str:
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
"""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
|
||||
|
||||
REMEDIATION = (
|
||||
"Pass explicit org= and repo= matching the local git remote, "
|
||||
"e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools."
|
||||
)
|
||||
|
||||
|
||||
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,
|
||||
}
|
||||
@@ -10,6 +10,10 @@ Load the canonical workflow first:
|
||||
Final report schema: `schemas/review-merge-final-report.md`.
|
||||
|
||||
Rules (llm-project-workflow):
|
||||
- Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
|
||||
every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
|
||||
repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo
|
||||
and is blocked when it disagrees with the local git remote URL.
|
||||
- Only an eligible, NON-author reviewer merges. If authenticated user == PR
|
||||
author → STOP.
|
||||
- Do not merge unless the PR is open, mergeable, and its checks/review pass.
|
||||
|
||||
@@ -35,6 +35,10 @@ Load the canonical workflow first:
|
||||
Final report schema: `schemas/review-merge-final-report.md`.
|
||||
|
||||
Rules (llm-project-workflow):
|
||||
- Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
|
||||
every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
|
||||
repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo
|
||||
and is blocked when it disagrees with the local git remote URL.
|
||||
- Review in a SEPARATE detached review worktree, never the author's folder.
|
||||
- Worktree safety (#233): before checkout, diff, validation, review, or merge,
|
||||
report the starting worktree path and whether it was dirty. If unrelated
|
||||
|
||||
@@ -15,6 +15,10 @@ Rules (llm-project-workflow):
|
||||
- Work only in an isolated branch worktree under branches/. The main checkout
|
||||
is orchestration/status only.
|
||||
- Do not self-review or self-merge.
|
||||
- Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
|
||||
every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
|
||||
repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo
|
||||
and is blocked when it disagrees with the local git remote URL.
|
||||
|
||||
Steps:
|
||||
0. Work Selection Rule — before any claim, branch, or file edits, acquire or
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
"""Regression coverage for the remote/repo mismatch guard (#530).
|
||||
|
||||
Bare ``remote=prgs`` historically resolved to ``Scaled-Tech-Consulting/Timesheet``
|
||||
(the hardcoded ``REMOTES`` default) instead of ``Scaled-Tech-Consulting/Gitea-Tools``,
|
||||
which is what the local git remote actually points at. That silent mismatch caused
|
||||
false 404s and risked mutating the wrong repository. The guard fails closed when the
|
||||
MCP-resolved org/repo disagrees with the local git remote URL and the caller did not
|
||||
pass explicit ``org``/``repo``.
|
||||
"""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import remote_repo_guard
|
||||
|
||||
|
||||
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
|
||||
|
||||
|
||||
class TestAssessRemoteRepoMatch(unittest.TestCase):
|
||||
def test_explicit_org_and_repo_skips_guard(self):
|
||||
# Caller supplied both org and repo explicitly: never block, even on mismatch.
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="Scaled-Tech-Consulting",
|
||||
resolved_repo="Timesheet",
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=True,
|
||||
repo_explicit=True,
|
||||
)
|
||||
self.assertTrue(assessment["proven"])
|
||||
self.assertFalse(assessment["block"])
|
||||
self.assertEqual(assessment["reasons"], [])
|
||||
|
||||
def test_missing_local_remote_url_skips_guard(self):
|
||||
# Best-effort lookup: no local remote URL (non-checkout usage) => do not block.
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="Scaled-Tech-Consulting",
|
||||
resolved_repo="Timesheet",
|
||||
local_remote_url=None,
|
||||
org_explicit=False,
|
||||
repo_explicit=False,
|
||||
)
|
||||
self.assertTrue(assessment["proven"])
|
||||
self.assertFalse(assessment["block"])
|
||||
|
||||
def test_matching_repo_is_proven(self):
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="Scaled-Tech-Consulting",
|
||||
resolved_repo="Gitea-Tools",
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=False,
|
||||
repo_explicit=False,
|
||||
)
|
||||
self.assertTrue(assessment["proven"])
|
||||
self.assertFalse(assessment["block"])
|
||||
|
||||
def test_regression_prgs_default_timesheet_vs_local_gitea_tools(self):
|
||||
# The exact #530 scenario: default repo Timesheet, local remote Gitea-Tools.
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="Scaled-Tech-Consulting",
|
||||
resolved_repo="Timesheet",
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=False,
|
||||
repo_explicit=False,
|
||||
)
|
||||
self.assertFalse(assessment["proven"])
|
||||
self.assertTrue(assessment["block"])
|
||||
self.assertTrue(assessment["reasons"])
|
||||
self.assertEqual(assessment["resolved_repo"], "Timesheet")
|
||||
self.assertEqual(assessment["resolved_org"], "Scaled-Tech-Consulting")
|
||||
|
||||
def test_only_org_explicit_still_checks_repo(self):
|
||||
# Repo left as default => still guarded even if org was explicit.
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="Scaled-Tech-Consulting",
|
||||
resolved_repo="Timesheet",
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=True,
|
||||
repo_explicit=False,
|
||||
)
|
||||
self.assertTrue(assessment["block"])
|
||||
|
||||
def test_case_insensitive_match(self):
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="scaled-tech-consulting",
|
||||
resolved_repo="gitea-tools",
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=False,
|
||||
repo_explicit=False,
|
||||
)
|
||||
self.assertTrue(assessment["proven"])
|
||||
|
||||
def test_format_error_mentions_resolved_and_local(self):
|
||||
assessment = remote_repo_guard.assess_remote_repo_match(
|
||||
remote="prgs",
|
||||
resolved_org="Scaled-Tech-Consulting",
|
||||
resolved_repo="Timesheet",
|
||||
local_remote_url=LOCAL_GITEA_TOOLS_URL,
|
||||
org_explicit=False,
|
||||
repo_explicit=False,
|
||||
)
|
||||
message = remote_repo_guard.format_remote_repo_guard_error(assessment)
|
||||
self.assertIn("Timesheet", message)
|
||||
self.assertIn("Gitea-Tools", message)
|
||||
self.assertIn("org=", message)
|
||||
self.assertIn("repo=", message)
|
||||
|
||||
|
||||
class TestResolveServerWiring(unittest.TestCase):
|
||||
"""The guard is wired into gitea_mcp_server._resolve, so every read/lookup/
|
||||
mutation tool that resolves a target fails closed on a repo mismatch.
|
||||
|
||||
Under pytest the guard is bypassed unless GITEA_FORCE_REMOTE_REPO_CHECK is set,
|
||||
so these tests set the force flag and patch the local remote lookup + REMOTES.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
import gitea_mcp_server
|
||||
|
||||
self.server = gitea_mcp_server
|
||||
|
||||
force = mock.patch.dict(
|
||||
os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"}, clear=False
|
||||
)
|
||||
force.start()
|
||||
self.addCleanup(force.stop)
|
||||
|
||||
url_patch = mock.patch.object(
|
||||
gitea_mcp_server,
|
||||
"_local_git_remote_url",
|
||||
return_value=LOCAL_GITEA_TOOLS_URL,
|
||||
)
|
||||
url_patch.start()
|
||||
self.addCleanup(url_patch.stop)
|
||||
|
||||
def _set_prgs_default_repo(self, repo):
|
||||
original = dict(self.server.REMOTES["prgs"])
|
||||
self.addCleanup(self.server.REMOTES.__setitem__, "prgs", original)
|
||||
self.server.REMOTES["prgs"] = {**original, "repo": repo}
|
||||
|
||||
def test_resolve_blocks_on_default_repo_mismatch(self):
|
||||
self._set_prgs_default_repo("Timesheet")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
self.server._resolve("prgs", None, None, None)
|
||||
self.assertIn("Timesheet", str(ctx.exception))
|
||||
self.assertIn("Gitea-Tools", str(ctx.exception))
|
||||
|
||||
def test_resolve_allows_explicit_org_and_repo(self):
|
||||
self._set_prgs_default_repo("Timesheet")
|
||||
# Explicit org/repo is authoritative even if it does not match local remote.
|
||||
host, org, repo = self.server._resolve(
|
||||
"prgs", None, "Scaled-Tech-Consulting", "Timesheet"
|
||||
)
|
||||
self.assertEqual((org, repo), ("Scaled-Tech-Consulting", "Timesheet"))
|
||||
|
||||
def test_resolve_allows_matching_default(self):
|
||||
self._set_prgs_default_repo("Gitea-Tools")
|
||||
host, org, repo = self.server._resolve("prgs", None, None, None)
|
||||
self.assertEqual((org, repo), ("Scaled-Tech-Consulting", "Gitea-Tools"))
|
||||
|
||||
def test_lookup_tool_cannot_silently_query_different_repo(self):
|
||||
# gitea_view_issue resolves the target via _resolve first, so a bare
|
||||
# remote=prgs pointing at the wrong default repo fails closed before any
|
||||
# API call is made.
|
||||
self._set_prgs_default_repo("Timesheet")
|
||||
with self.assertRaises(RuntimeError):
|
||||
self.server.gitea_view_issue(issue_number=1, remote="prgs")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user