fix: redact Gitea web links from PR/issue MCP tool output (#125) #133
+30
-17
@@ -54,6 +54,17 @@ def _reveal_endpoints() -> bool:
|
|||||||
return (os.environ.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() \
|
return (os.environ.get("GITEA_MCP_REVEAL_ENDPOINTS") or "").strip().lower() \
|
||||||
in ("1", "true", "yes")
|
in ("1", "true", "yes")
|
||||||
|
|
||||||
|
|
||||||
|
def _with_optional_url(result: dict, url: str | None) -> dict:
|
||||||
|
"""Attach web links only under the explicit endpoint reveal opt-in.
|
||||||
|
|
||||||
|
Mutates *result* in place and returns it; pass a freshly-built dict,
|
||||||
|
never a shared/aliased one.
|
||||||
|
"""
|
||||||
|
if _reveal_endpoints() and url:
|
||||||
|
result["url"] = url
|
||||||
|
return result
|
||||||
|
|
||||||
mcp = FastMCP("gitea-tools", instructions=(
|
mcp = FastMCP("gitea-tools", instructions=(
|
||||||
"Gitea issue tracker and PR management for dadeschools and prgs instances. "
|
"Gitea issue tracker and PR management for dadeschools and prgs instances. "
|
||||||
"Use the gitea_ prefixed tools to create issues, PRs, list issues, etc."
|
"Use the gitea_ prefixed tools to create issues, PRs, list issues, etc."
|
||||||
@@ -310,7 +321,7 @@ def gitea_create_issue(
|
|||||||
repo: Override the repository name.
|
repo: Override the repository name.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict with 'number' and 'url' of the created issue.
|
dict with 'number' of the created issue ('url' only with the reveal opt-in).
|
||||||
"""
|
"""
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
@@ -325,7 +336,7 @@ def gitea_create_issue(
|
|||||||
_audit("create_issue", host=h, remote=remote, org=o, repo=r,
|
_audit("create_issue", host=h, remote=remote, org=o, repo=r,
|
||||||
result=gitea_audit.SUCCEEDED, issue_number=data["number"],
|
result=gitea_audit.SUCCEEDED, issue_number=data["number"],
|
||||||
request_metadata={"title": title})
|
request_metadata={"title": title})
|
||||||
return {"number": data["number"], "url": data["html_url"]}
|
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -352,7 +363,7 @@ def gitea_create_pr(
|
|||||||
repo: Override the repository name.
|
repo: Override the repository name.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict with 'number' and 'url' of the created PR.
|
dict with 'number' of the created PR ('url' only with the reveal opt-in).
|
||||||
"""
|
"""
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
@@ -369,7 +380,7 @@ def gitea_create_pr(
|
|||||||
_audit("create_pr", host=h, remote=remote, org=o, repo=r,
|
_audit("create_pr", host=h, remote=remote, org=o, repo=r,
|
||||||
result=gitea_audit.SUCCEEDED, pr_number=data["number"],
|
result=gitea_audit.SUCCEEDED, pr_number=data["number"],
|
||||||
target_branch=head, request_metadata=meta)
|
target_branch=head, request_metadata=meta)
|
||||||
return {"number": data["number"], "url": data["html_url"]}
|
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -390,24 +401,25 @@ def gitea_list_prs(
|
|||||||
repo: Override the repository name.
|
repo: Override the repository name.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
List of dicts with 'number', 'title', 'state', 'head', 'base', 'url', 'mergeable'.
|
List of dicts with 'number', 'title', 'state', 'head', 'base', and
|
||||||
|
'mergeable' ('url' only with the reveal opt-in).
|
||||||
"""
|
"""
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
|
url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
|
||||||
prs = api_get_all(url, auth)
|
prs = api_get_all(url, auth)
|
||||||
return [
|
results = []
|
||||||
{
|
for pr in prs:
|
||||||
|
entry = {
|
||||||
"number": pr["number"],
|
"number": pr["number"],
|
||||||
"title": pr["title"],
|
"title": pr["title"],
|
||||||
"state": pr["state"],
|
"state": pr["state"],
|
||||||
"head": pr["head"]["ref"],
|
"head": pr["head"]["ref"],
|
||||||
"base": pr["base"]["ref"],
|
"base": pr["base"]["ref"],
|
||||||
"url": pr["html_url"],
|
|
||||||
"mergeable": pr.get("mergeable"),
|
"mergeable": pr.get("mergeable"),
|
||||||
}
|
}
|
||||||
for pr in prs
|
results.append(_with_optional_url(entry, pr.get("html_url")))
|
||||||
]
|
return results
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -434,17 +446,17 @@ def gitea_view_pr(
|
|||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
|
url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
|
||||||
pr = api_request("GET", url, auth)
|
pr = api_request("GET", url, auth)
|
||||||
return {
|
result = {
|
||||||
"number": pr["number"],
|
"number": pr["number"],
|
||||||
"title": pr["title"],
|
"title": pr["title"],
|
||||||
"body": pr.get("body", ""),
|
"body": pr.get("body", ""),
|
||||||
"state": pr["state"],
|
"state": pr["state"],
|
||||||
"head": pr["head"]["ref"],
|
"head": pr["head"]["ref"],
|
||||||
"base": pr["base"]["ref"],
|
"base": pr["base"]["ref"],
|
||||||
"url": pr["html_url"],
|
|
||||||
"mergeable": pr.get("mergeable"),
|
"mergeable": pr.get("mergeable"),
|
||||||
"user": pr.get("user", {}).get("login", ""),
|
"user": pr.get("user", {}).get("login", ""),
|
||||||
}
|
}
|
||||||
|
return _with_optional_url(result, pr.get("html_url"))
|
||||||
|
|
||||||
|
|
||||||
# Actions whose eligibility this tool can evaluate.
|
# Actions whose eligibility this tool can evaluate.
|
||||||
@@ -921,15 +933,15 @@ def gitea_edit_pr(
|
|||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
return {
|
result = {
|
||||||
"success": True,
|
"success": True,
|
||||||
"number": data["number"],
|
"number": data["number"],
|
||||||
"title": data["title"],
|
"title": data["title"],
|
||||||
"body": data.get("body", ""),
|
"body": data.get("body", ""),
|
||||||
"state": data["state"],
|
"state": data["state"],
|
||||||
"url": data["html_url"],
|
|
||||||
"cleanup_status": cleanup_status,
|
"cleanup_status": cleanup_status,
|
||||||
}
|
}
|
||||||
|
return _with_optional_url(result, data.get("html_url"))
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
@@ -1424,21 +1436,22 @@ def gitea_view_issue(
|
|||||||
repo: Override the repository name.
|
repo: Override the repository name.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict with 'number', 'title', 'body', 'state', 'labels', 'assignee', 'url'.
|
dict with 'number', 'title', 'body', 'state', 'labels', and 'assignee'
|
||||||
|
('url' only with the reveal opt-in).
|
||||||
"""
|
"""
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
|
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
|
||||||
i = api_request("GET", url, auth)
|
i = api_request("GET", url, auth)
|
||||||
return {
|
result = {
|
||||||
"number": i["number"],
|
"number": i["number"],
|
||||||
"title": i["title"],
|
"title": i["title"],
|
||||||
"body": i.get("body", ""),
|
"body": i.get("body", ""),
|
||||||
"state": i["state"],
|
"state": i["state"],
|
||||||
"labels": [lb["name"] for lb in i.get("labels", [])],
|
"labels": [lb["name"] for lb in i.get("labels", [])],
|
||||||
"assignee": (i.get("assignee") or {}).get("login", ""),
|
"assignee": (i.get("assignee") or {}).get("login", ""),
|
||||||
"url": i["html_url"],
|
|
||||||
}
|
}
|
||||||
|
return _with_optional_url(result, i.get("html_url"))
|
||||||
|
|
||||||
|
|
||||||
def _issue_comment_gate(op: str) -> list[str]:
|
def _issue_comment_gate(op: str) -> list[str]:
|
||||||
|
|||||||
@@ -48,15 +48,24 @@ class TestCreateIssue(unittest.TestCase):
|
|||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_creates_issue(self, _auth, mock_api):
|
def test_creates_issue(self, _auth, mock_api):
|
||||||
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
||||||
result = gitea_create_issue(title="Test issue", body="body text")
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
result = gitea_create_issue(title="Test issue", body="body text")
|
||||||
self.assertEqual(result["number"], 1)
|
self.assertEqual(result["number"], 1)
|
||||||
self.assertIn("issues/1", result["url"])
|
self.assertNotIn("url", result)
|
||||||
mock_api.assert_called_once()
|
mock_api.assert_called_once()
|
||||||
# Verify payload
|
# Verify payload
|
||||||
call_args = mock_api.call_args
|
call_args = mock_api.call_args
|
||||||
self.assertEqual(call_args[0][0], "POST")
|
self.assertEqual(call_args[0][0], "POST")
|
||||||
self.assertEqual(call_args[0][3]["title"], "Test issue")
|
self.assertEqual(call_args[0][3]["title"], "Test issue")
|
||||||
|
|
||||||
|
@patch("mcp_server.api_request")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
|
||||||
|
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
||||||
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
|
result = gitea_create_issue(title="Test issue", body="body text")
|
||||||
|
self.assertIn("issues/1", result["url"])
|
||||||
|
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_creates_on_prgs(self, _auth, mock_api):
|
def test_creates_on_prgs(self, _auth, mock_api):
|
||||||
@@ -77,12 +86,22 @@ class TestCreatePR(unittest.TestCase):
|
|||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
def test_creates_pr(self, _auth, mock_api):
|
def test_creates_pr(self, _auth, mock_api):
|
||||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||||
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
|
||||||
self.assertEqual(result["number"], 3)
|
self.assertEqual(result["number"], 3)
|
||||||
|
self.assertNotIn("url", result)
|
||||||
payload = mock_api.call_args[0][3]
|
payload = mock_api.call_args[0][3]
|
||||||
self.assertEqual(payload["head"], "feat/x")
|
self.assertEqual(payload["head"], "feat/x")
|
||||||
self.assertEqual(payload["base"], "main")
|
self.assertEqual(payload["base"], "main")
|
||||||
|
|
||||||
|
@patch("mcp_server.api_request")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
|
||||||
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||||
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
|
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
|
||||||
|
self.assertIn("pulls/3", result["url"])
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Close Issue
|
# Close Issue
|
||||||
@@ -158,11 +177,25 @@ class TestViewIssue(unittest.TestCase):
|
|||||||
"assignee": {"login": "jason"},
|
"assignee": {"login": "jason"},
|
||||||
"html_url": "https://gitea.prgs.cc/issues/7",
|
"html_url": "https://gitea.prgs.cc/issues/7",
|
||||||
}
|
}
|
||||||
result = gitea_view_issue(issue_number=7, remote="prgs")
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
result = gitea_view_issue(issue_number=7, remote="prgs")
|
||||||
self.assertEqual(result["number"], 7)
|
self.assertEqual(result["number"], 7)
|
||||||
self.assertEqual(result["body"], "Build it")
|
self.assertEqual(result["body"], "Build it")
|
||||||
self.assertEqual(result["labels"], ["important"])
|
self.assertEqual(result["labels"], ["important"])
|
||||||
self.assertEqual(result["assignee"], "jason")
|
self.assertEqual(result["assignee"], "jason")
|
||||||
|
self.assertNotIn("url", result)
|
||||||
|
|
||||||
|
@patch("mcp_server.api_request")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
def test_view_issue_reveal_opt_in_includes_url(self, _auth, mock_api):
|
||||||
|
mock_api.return_value = {
|
||||||
|
"number": 7, "title": "MCP server", "body": "Build it",
|
||||||
|
"state": "open", "labels": [], "assignee": None,
|
||||||
|
"html_url": "https://gitea.prgs.cc/issues/7",
|
||||||
|
}
|
||||||
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
|
result = gitea_view_issue(issue_number=7, remote="prgs")
|
||||||
|
self.assertIn("issues/7", result["url"])
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -272,10 +305,26 @@ class TestListPRs(unittest.TestCase):
|
|||||||
"html_url": "http://url1", "mergeable": True
|
"html_url": "http://url1", "mergeable": True
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
result = gitea_list_prs()
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
result = gitea_list_prs()
|
||||||
self.assertEqual(len(result), 1)
|
self.assertEqual(len(result), 1)
|
||||||
self.assertEqual(result[0]["number"], 1)
|
self.assertEqual(result[0]["number"], 1)
|
||||||
self.assertEqual(result[0]["head"], "branch1")
|
self.assertEqual(result[0]["head"], "branch1")
|
||||||
|
self.assertNotIn("url", result[0])
|
||||||
|
|
||||||
|
@patch("mcp_server.api_get_all")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
def test_list_prs_reveal_opt_in_includes_url(self, _auth, mock_api):
|
||||||
|
mock_api.return_value = [
|
||||||
|
{
|
||||||
|
"number": 1, "title": "PR 1", "state": "open",
|
||||||
|
"head": {"ref": "branch1"}, "base": {"ref": "main"},
|
||||||
|
"html_url": "http://url1", "mergeable": True
|
||||||
|
}
|
||||||
|
]
|
||||||
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
|
result = gitea_list_prs()
|
||||||
|
self.assertEqual(result[0]["url"], "http://url1")
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -292,10 +341,25 @@ class TestViewPR(unittest.TestCase):
|
|||||||
"html_url": "http://url1", "mergeable": True, "body": "description",
|
"html_url": "http://url1", "mergeable": True, "body": "description",
|
||||||
"user": {"login": "user1"}
|
"user": {"login": "user1"}
|
||||||
}
|
}
|
||||||
result = gitea_view_pr(pr_number=1)
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
result = gitea_view_pr(pr_number=1)
|
||||||
self.assertEqual(result["number"], 1)
|
self.assertEqual(result["number"], 1)
|
||||||
self.assertEqual(result["body"], "description")
|
self.assertEqual(result["body"], "description")
|
||||||
self.assertEqual(result["user"], "user1")
|
self.assertEqual(result["user"], "user1")
|
||||||
|
self.assertNotIn("url", result)
|
||||||
|
|
||||||
|
@patch("mcp_server.api_request")
|
||||||
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
|
def test_view_pr_reveal_opt_in_includes_url(self, _auth, mock_api):
|
||||||
|
mock_api.return_value = {
|
||||||
|
"number": 1, "title": "PR 1", "state": "open",
|
||||||
|
"head": {"ref": "branch1"}, "base": {"ref": "main"},
|
||||||
|
"html_url": "http://url1", "mergeable": True, "body": "description",
|
||||||
|
"user": {"login": "user1"}
|
||||||
|
}
|
||||||
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
|
result = gitea_view_pr(pr_number=1)
|
||||||
|
self.assertEqual(result["url"], "http://url1")
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
@@ -1674,9 +1738,30 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
|
|||||||
return {}
|
return {}
|
||||||
self.mock_api.side_effect = api_side_effect
|
self.mock_api.side_effect = api_side_effect
|
||||||
|
|
||||||
res = gitea_edit_pr(pr_number=1, state="closed")
|
with patch.dict(os.environ, {}, clear=True):
|
||||||
|
res = gitea_edit_pr(pr_number=1, state="closed")
|
||||||
self.assertTrue(res["success"])
|
self.assertTrue(res["success"])
|
||||||
self.assertEqual(res["cleanup_status"].get(123), "released")
|
self.assertEqual(res["cleanup_status"].get(123), "released")
|
||||||
|
self.assertNotIn("url", res)
|
||||||
|
|
||||||
|
def test_edit_pr_reveal_opt_in_includes_url(self):
|
||||||
|
def api_side_effect(method, url, auth, payload=None):
|
||||||
|
if method == "PATCH" and "pulls/1" in url:
|
||||||
|
return {
|
||||||
|
"number": 1,
|
||||||
|
"title": "My PR",
|
||||||
|
"state": "open",
|
||||||
|
"html_url": "http://url1",
|
||||||
|
"body": "No issue link",
|
||||||
|
"head": {"ref": "feat/my-branch"}
|
||||||
|
}
|
||||||
|
return {}
|
||||||
|
self.mock_api.side_effect = api_side_effect
|
||||||
|
|
||||||
|
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||||
|
res = gitea_edit_pr(pr_number=1, title="Updated")
|
||||||
|
self.assertTrue(res["success"])
|
||||||
|
self.assertEqual(res["url"], "http://url1")
|
||||||
|
|
||||||
def test_multiple_linked_issues(self):
|
def test_multiple_linked_issues(self):
|
||||||
def api_side_effect(method, url, auth, payload=None):
|
def api_side_effect(method, url, auth, payload=None):
|
||||||
|
|||||||
Reference in New Issue
Block a user