Merge branch 'feat/issue-263-native-commit-payloads' into docs/issue-261-agent-artifact-cleanup
This commit is contained in:
+130
-4
@@ -2295,6 +2295,106 @@ def gitea_get_file(
|
||||
}
|
||||
|
||||
|
||||
def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[dict]]:
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
|
||||
processed_files = []
|
||||
source_proofs = []
|
||||
|
||||
lock_data = {}
|
||||
if os.path.exists(ISSUE_LOCK_FILE):
|
||||
try:
|
||||
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
|
||||
lock_data = json.load(f)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
locked_worktree = lock_data.get("worktree_path")
|
||||
if locked_worktree:
|
||||
locked_worktree = os.path.realpath(locked_worktree)
|
||||
|
||||
for f in files:
|
||||
f_copy = dict(f)
|
||||
path = f_copy.get("path", "")
|
||||
|
||||
content_keys = [k for k in ["content", "content_plain", "workspace_path", "local_path"] if k in f_copy]
|
||||
if len(content_keys) > 1:
|
||||
raise ValueError(
|
||||
f"Multiple content sources specified for file '{path}'; provide exactly one of "
|
||||
f"'content', 'content_plain', 'workspace_path', or 'local_path' (fail closed)"
|
||||
)
|
||||
|
||||
source = "none"
|
||||
if "content_plain" in f_copy:
|
||||
content_plain = f_copy.pop("content_plain")
|
||||
if content_plain is not None:
|
||||
content_bytes = content_plain.encode("utf-8")
|
||||
f_copy["content"] = base64.b64encode(content_bytes).decode("utf-8")
|
||||
source = "inline_plain"
|
||||
elif "workspace_path" in f_copy:
|
||||
workspace_path = f_copy.pop("workspace_path")
|
||||
if not locked_worktree:
|
||||
raise RuntimeError(
|
||||
f"Issue lock is missing or does not define a worktree path. "
|
||||
f"Cannot resolve workspace_path '{workspace_path}' (fail closed)."
|
||||
)
|
||||
target_path = os.path.realpath(os.path.abspath(os.path.join(locked_worktree, workspace_path)))
|
||||
|
||||
is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
|
||||
is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
|
||||
|
||||
if not (is_inside or is_tmp):
|
||||
raise ValueError(
|
||||
f"workspace_path '{workspace_path}' resolves to '{target_path}' which falls outside "
|
||||
f"of locked worktree '{locked_worktree}' (fail closed)"
|
||||
)
|
||||
|
||||
if not os.path.exists(target_path):
|
||||
raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
|
||||
|
||||
with open(target_path, "rb") as fh:
|
||||
file_bytes = fh.read()
|
||||
f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
|
||||
source = "workspace_path"
|
||||
elif "local_path" in f_copy:
|
||||
local_path = f_copy.pop("local_path")
|
||||
if not locked_worktree:
|
||||
raise RuntimeError(
|
||||
f"Issue lock is missing or does not define a worktree path. "
|
||||
f"Cannot resolve local_path '{local_path}' (fail closed)."
|
||||
)
|
||||
target_path = os.path.realpath(os.path.abspath(local_path))
|
||||
|
||||
is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
|
||||
is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
|
||||
|
||||
if not (is_inside or is_tmp):
|
||||
raise ValueError(
|
||||
f"local_path '{local_path}' resolves to '{target_path}' which falls outside "
|
||||
f"of locked worktree '{locked_worktree}' (fail closed)"
|
||||
)
|
||||
|
||||
if not os.path.exists(target_path):
|
||||
raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
|
||||
|
||||
with open(target_path, "rb") as fh:
|
||||
file_bytes = fh.read()
|
||||
f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
|
||||
source = "local_path"
|
||||
elif "content" in f_copy:
|
||||
source = "inline_base64"
|
||||
|
||||
processed_files.append(f_copy)
|
||||
source_proofs.append({
|
||||
"path": path,
|
||||
"source": source
|
||||
})
|
||||
|
||||
return processed_files, source_proofs
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_commit_files(
|
||||
files: list[dict],
|
||||
@@ -2309,7 +2409,7 @@ def gitea_commit_files(
|
||||
"""Commit changes to multiple files in a Gitea repository in a single atomic commit.
|
||||
|
||||
Args:
|
||||
files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and 'content' (base64 encoded for create/update), and optionally 'sha' (required for update/delete) or 'from_path' (for rename).
|
||||
files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and one of content payload sources: 'content' (base64), 'content_plain', 'workspace_path', or 'local_path'.
|
||||
message: The commit message.
|
||||
branch: Optional existing branch to start/commit from.
|
||||
new_branch: Optional new branch name to create for this commit.
|
||||
@@ -2321,13 +2421,38 @@ def gitea_commit_files(
|
||||
Returns:
|
||||
dict with success status and commit/branch information.
|
||||
"""
|
||||
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
|
||||
"commit_files"
|
||||
)
|
||||
if not ok:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"commit": "",
|
||||
"branch": "",
|
||||
"reasons": block_reasons,
|
||||
}
|
||||
blocked = _namespace_mutation_block(
|
||||
"commit_files", commit="", branch="", remote=remote
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
blocked = _profile_permission_block(
|
||||
task_capability_map.required_permission("commit_files"),
|
||||
commit="", branch="",
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
processed_files, source_proofs = _prepare_commit_payload_files(files)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
url = f"{repo_api_url(h, o, r)}/contents"
|
||||
|
||||
payload = {
|
||||
"files": files,
|
||||
"files": processed_files,
|
||||
"message": message,
|
||||
}
|
||||
if branch is not None:
|
||||
@@ -2338,13 +2463,14 @@ def gitea_commit_files(
|
||||
with _audited("commit_files", host=h, remote=remote, org=o, repo=r,
|
||||
target_branch=(new_branch or branch),
|
||||
request_metadata={"message": message,
|
||||
"paths": [f.get("path") for f in files],
|
||||
"operations": [f.get("operation") for f in files]}):
|
||||
"paths": [f.get("path") for f in processed_files],
|
||||
"operations": [f.get("operation") for f in processed_files]}):
|
||||
data = api_request("POST", url, auth, payload)
|
||||
return {
|
||||
"success": True,
|
||||
"commit": data.get("commit", {}).get("sha", ""),
|
||||
"branch": data.get("branch", {}).get("name", ""),
|
||||
"content_source_proof": source_proofs,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -80,6 +80,14 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"permission": "gitea.branch.delete",
|
||||
"role": "author",
|
||||
},
|
||||
"commit_files": {
|
||||
"permission": "gitea.repo.commit",
|
||||
"role": "author",
|
||||
},
|
||||
"gitea_commit_files": {
|
||||
"permission": "gitea.repo.commit",
|
||||
"role": "author",
|
||||
},
|
||||
}
|
||||
|
||||
# Issue-mutating MCP tools and their resolver task keys.
|
||||
@@ -89,6 +97,7 @@ ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
|
||||
"gitea_create_issue_comment": "comment_issue",
|
||||
"gitea_mark_issue": "mark_issue",
|
||||
"gitea_set_issue_labels": "set_issue_labels",
|
||||
"gitea_commit_files": "commit_files",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
"""Tests for commit_files task capability resolver mapping (#262)."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
import task_capability_map
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"commit-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": ["gitea.read", "gitea.repo.commit"],
|
||||
"forbidden_operations": ["gitea.pr.create"],
|
||||
"execution_profile": "commit-author",
|
||||
},
|
||||
"pr-only-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "pr-author",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.create"],
|
||||
"forbidden_operations": ["gitea.repo.commit"],
|
||||
"execution_profile": "pr-only-author",
|
||||
},
|
||||
"reviewer-profile": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "reviewer-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.merge"],
|
||||
"forbidden_operations": [
|
||||
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push",
|
||||
],
|
||||
"execution_profile": "reviewer-profile",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
|
||||
class CommitFilesCapabilityBase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._preflight_snapshot = (
|
||||
mcp_server._preflight_whoami_called,
|
||||
mcp_server._preflight_capability_called,
|
||||
)
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
mcp_server._preflight_whoami_called, mcp_server._preflight_capability_called = (
|
||||
self._preflight_snapshot
|
||||
)
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
|
||||
|
||||
class TestCommitFilesResolver(CommitFilesCapabilityBase):
|
||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_resolves_to_repo_commit(self, _auth, _api):
|
||||
with patch.dict(os.environ, self._env("commit-author"), clear=True):
|
||||
for task in ("commit_files", "gitea_commit_files"):
|
||||
with self.subTest(task=task):
|
||||
res = mcp_server.gitea_resolve_task_capability(
|
||||
task=task, remote="prgs")
|
||||
self.assertEqual(res["required_operation_permission"],
|
||||
"gitea.repo.commit")
|
||||
self.assertEqual(res["required_role_kind"], "author")
|
||||
self.assertTrue(res["allowed_in_current_session"])
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "pr-author"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_create_pr_does_not_satisfy_commit_files(self, _auth, _api):
|
||||
with patch.dict(os.environ, self._env("pr-only-author"), clear=True):
|
||||
commit_res = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs")
|
||||
pr_res = mcp_server.gitea_resolve_task_capability(
|
||||
task="create_pr", remote="prgs")
|
||||
self.assertFalse(commit_res["allowed_in_current_session"])
|
||||
self.assertTrue(pr_res["allowed_in_current_session"])
|
||||
self.assertEqual(commit_res["required_operation_permission"],
|
||||
"gitea.repo.commit")
|
||||
self.assertEqual(pr_res["required_operation_permission"],
|
||||
"gitea.pr.create")
|
||||
|
||||
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_reviewer_denied_commit_files(self, _auth, _api):
|
||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
||||
res = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs")
|
||||
self.assertFalse(res["allowed_in_current_session"])
|
||||
self.assertEqual(res["required_operation_permission"],
|
||||
"gitea.repo.commit")
|
||||
self.assertTrue(res["different_mcp_namespace_required"])
|
||||
|
||||
|
||||
class TestCommitFilesToolGate(CommitFilesCapabilityBase):
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "abc"},
|
||||
"branch": {"name": "feat/x"},
|
||||
}
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
with patch.dict(os.environ, self._env("commit-author"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
|
||||
message="test",
|
||||
new_branch="feat/x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_reviewer_blocked_with_permission_report(self, _auth, _role):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
|
||||
message="test",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res.get("performed", True))
|
||||
self.assertIn("permission_report", res)
|
||||
self.assertEqual(res["permission_report"]["missing_permission"],
|
||||
"gitea.repo.commit")
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
def test_preflight_blocks_before_api(self, _role):
|
||||
env = {**self._env("commit-author"), "GITEA_TEST_FORCE_DIRTY": "1"}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "a.txt", "content": "YQ=="}],
|
||||
message="test",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("gitea_whoami", str(ctx.exception))
|
||||
|
||||
|
||||
class TestCommitFilesMapAlignment(unittest.TestCase):
|
||||
def test_tool_gate_matches_resolver(self):
|
||||
self.assertEqual(
|
||||
task_capability_map.tool_required_permission("gitea_commit_files"),
|
||||
task_capability_map.required_permission("commit_files"),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,251 @@
|
||||
"""Regression tests: gitea_commit_files tool gates match resolver and whoami verification.
|
||||
|
||||
Covers Issue #262 requirements.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
import task_capability_map
|
||||
import gitea_config
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"full-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.issue.create", "gitea.repo.commit"
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
"execution_profile": "full-author",
|
||||
},
|
||||
"reviewer-no-commit": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "reviewer",
|
||||
"username": "reviewer-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
|
||||
"allowed_operations": [
|
||||
"gitea.read", "gitea.pr.review", "gitea.pr.approve"
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
|
||||
],
|
||||
"execution_profile": "reviewer-no-commit",
|
||||
},
|
||||
},
|
||||
"rules": {"allow_runtime_switching": False},
|
||||
}
|
||||
|
||||
|
||||
class TestCommitFilesGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
gitea_config._active_profile_override = None
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
gitea_config._active_profile_override = None
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
|
||||
}
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_allowed_author_proceeds(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "abc123commit"},
|
||||
"branch": {"name": "some-branch"},
|
||||
}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs"
|
||||
)
|
||||
self.assertTrue(resolve["allowed_in_current_session"])
|
||||
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["commit"], "abc123commit")
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_denied_reviewer_blocked(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {"login": "reviewer-user"}
|
||||
with patch.dict(os.environ, self._env("reviewer-no-commit"), clear=True):
|
||||
resolve = mcp_server.gitea_resolve_task_capability(
|
||||
task="commit_files", remote="prgs"
|
||||
)
|
||||
self.assertFalse(resolve["allowed_in_current_session"])
|
||||
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res.get("performed", True))
|
||||
self.assertIn("permission_report", res)
|
||||
self.assertEqual(
|
||||
res["permission_report"]["missing_permission"], "gitea.repo.commit"
|
||||
)
|
||||
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
|
||||
self.assertFalse(post_calls)
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||
def test_unknown_profile_fails_closed(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {"login": "reviewer-user"}
|
||||
with patch.dict(os.environ, self._env("non-existent"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["success"])
|
||||
self.assertFalse(res.get("performed", True))
|
||||
post_calls = [c for c in mock_api.call_args_list if len(c.args) > 0 and c.args[0] == "POST"]
|
||||
self.assertFalse(post_calls)
|
||||
|
||||
|
||||
class TestPreflightCommitFilesGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
self.orig_whoami_called = mcp_server._preflight_whoami_called
|
||||
self.orig_capability_called = mcp_server._preflight_capability_called
|
||||
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
|
||||
self.orig_capability_violation = mcp_server._preflight_capability_violation
|
||||
self.orig_resolved_role = mcp_server._preflight_resolved_role
|
||||
self.orig_process_start = mcp_server._process_start_porcelain
|
||||
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
|
||||
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
|
||||
self.orig_whoami_files = mcp_server._preflight_whoami_violation_files
|
||||
self.orig_capability_files = mcp_server._preflight_capability_violation_files
|
||||
self.orig_reviewer_files = mcp_server._preflight_reviewer_violation_files
|
||||
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
mcp_server._preflight_capability_violation = False
|
||||
mcp_server._preflight_resolved_role = None
|
||||
mcp_server._process_start_porcelain = ""
|
||||
mcp_server._preflight_whoami_baseline_porcelain = None
|
||||
mcp_server._preflight_capability_baseline_porcelain = None
|
||||
mcp_server._preflight_whoami_violation_files = []
|
||||
mcp_server._preflight_capability_violation_files = []
|
||||
mcp_server._preflight_reviewer_violation_files = []
|
||||
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami_called
|
||||
mcp_server._preflight_capability_called = self.orig_capability_called
|
||||
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
|
||||
mcp_server._preflight_capability_violation = self.orig_capability_violation
|
||||
mcp_server._preflight_resolved_role = self.orig_resolved_role
|
||||
mcp_server._process_start_porcelain = self.orig_process_start
|
||||
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
|
||||
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
|
||||
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
|
||||
mcp_server._preflight_capability_violation_files = self.orig_capability_files
|
||||
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
|
||||
|
||||
self._dir.cleanup()
|
||||
os.environ.pop("GITEA_TEST_FORCE_DIRTY", None)
|
||||
os.environ.pop("GITEA_TEST_PORCELAIN", None)
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
}
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_preflight_not_called_blocks_commit(self, _auth, mock_api):
|
||||
mock_api.return_value = {"login": "author-user"}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_dirty_workspace_before_whoami_blocks_commit(self, _auth, mock_api):
|
||||
mock_api.return_value = {"login": "author-user"}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[{"operation": "create", "path": "x.txt", "content": "YQ=="}],
|
||||
message="Add x",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,251 @@
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import gitea_config
|
||||
import mcp_server
|
||||
import task_capability_map
|
||||
|
||||
CONFIG = {
|
||||
"version": 2,
|
||||
"contexts": {
|
||||
"ctx": {
|
||||
"enabled": True,
|
||||
"gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
|
||||
}
|
||||
},
|
||||
"profiles": {
|
||||
"full-author": {
|
||||
"enabled": True,
|
||||
"context": "ctx",
|
||||
"role": "author",
|
||||
"username": "author-user",
|
||||
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.repo.commit",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.review",
|
||||
],
|
||||
"execution_profile": "full-author",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class TestCommitPayloads(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._remotes = patch.dict(mcp_server.REMOTES, {
|
||||
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
|
||||
"repo": "Example-Repo"},
|
||||
})
|
||||
self._remotes.start()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
# Setup temp directories and lock files
|
||||
self._dir = tempfile.TemporaryDirectory()
|
||||
self.config_path = os.path.join(self._dir.name, "profiles.json")
|
||||
with open(self.config_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(CONFIG))
|
||||
|
||||
self.locked_worktree_dir = tempfile.TemporaryDirectory()
|
||||
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
|
||||
|
||||
self.lock_file_path = "/tmp/gitea_issue_lock.json"
|
||||
self.lock_data = {
|
||||
"issue_number": 263,
|
||||
"branch_name": "feat/issue-263-native-commit-payloads",
|
||||
"remote": "prgs",
|
||||
"org": "Example-Org",
|
||||
"repo": "Example-Repo",
|
||||
"worktree_path": self.locked_worktree_path,
|
||||
}
|
||||
with open(self.lock_file_path, "w", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps(self.lock_data))
|
||||
|
||||
# Reset preflight status to bypass/pass verification in tests
|
||||
self.orig_whoami_called = mcp_server._preflight_whoami_called
|
||||
self.orig_capability_called = mcp_server._preflight_capability_called
|
||||
mcp_server._preflight_whoami_called = True
|
||||
mcp_server._preflight_capability_called = True
|
||||
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami_called
|
||||
mcp_server._preflight_capability_called = self.orig_capability_called
|
||||
|
||||
self._dir.cleanup()
|
||||
self.locked_worktree_dir.cleanup()
|
||||
if os.path.exists(self.lock_file_path):
|
||||
os.remove(self.lock_file_path)
|
||||
|
||||
def _env(self, profile: str) -> dict:
|
||||
return {
|
||||
"GITEA_MCP_CONFIG": self.config_path,
|
||||
"GITEA_MCP_PROFILE": profile,
|
||||
"GITEA_TOKEN_AUTHOR": "author-pass",
|
||||
"GITEA_TEST_PORCELAIN": "",
|
||||
}
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_content_plain(self, _auth, mock_api):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "sha-123"},
|
||||
"branch": {"name": "feat/issue-263-native-commit-payloads"}
|
||||
}
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[
|
||||
{
|
||||
"operation": "create",
|
||||
"path": "hello.txt",
|
||||
"content_plain": "Hello World!",
|
||||
}
|
||||
],
|
||||
message="Add hello.txt",
|
||||
remote="prgs",
|
||||
)
|
||||
print("DEBUG RES IS:", res)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["commit"], "sha-123")
|
||||
self.assertEqual(res["content_source_proof"][0]["source"], "inline_plain")
|
||||
|
||||
# Verify posted base64 content
|
||||
post_args = mock_api.call_args[0]
|
||||
payload = post_args[3]
|
||||
self.assertEqual(payload["files"][0]["content"], "SGVsbG8gV29ybGQh")
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_workspace_path(self, _auth, mock_api):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "sha-456"},
|
||||
"branch": {"name": "feat/issue-263-native-commit-payloads"}
|
||||
}
|
||||
|
||||
# Create a workspace file
|
||||
file_relative = "src/code.py"
|
||||
file_abs = os.path.join(self.locked_worktree_path, file_relative)
|
||||
os.makedirs(os.path.dirname(file_abs), exist_ok=True)
|
||||
with open(file_abs, "wb") as fh:
|
||||
fh.write(b"print('hello')")
|
||||
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[
|
||||
{
|
||||
"operation": "create",
|
||||
"path": "src/code.py",
|
||||
"workspace_path": file_relative,
|
||||
}
|
||||
],
|
||||
message="Add code.py",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["content_source_proof"][0]["source"], "workspace_path")
|
||||
|
||||
post_args = mock_api.call_args[0]
|
||||
payload = post_args[3]
|
||||
# "print('hello')" in base64 is cHJpbnQoJ2hlbGxvJyk=
|
||||
self.assertEqual(payload["files"][0]["content"], "cHJpbnQoJ2hlbGxvJyk=")
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_local_path(self, _auth, mock_api):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "sha-789"},
|
||||
"branch": {"name": "feat/issue-263-native-commit-payloads"}
|
||||
}
|
||||
|
||||
file_abs = os.path.join(self.locked_worktree_path, "local.bin")
|
||||
with open(file_abs, "wb") as fh:
|
||||
fh.write(b"\x00\x01\x02\x03\xff")
|
||||
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
res = mcp_server.gitea_commit_files(
|
||||
files=[
|
||||
{
|
||||
"operation": "create",
|
||||
"path": "local.bin",
|
||||
"local_path": file_abs,
|
||||
}
|
||||
],
|
||||
message="Add local.bin",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res["success"])
|
||||
self.assertEqual(res["content_source_proof"][0]["source"], "local_path")
|
||||
|
||||
post_args = mock_api.call_args[0]
|
||||
payload = post_args[3]
|
||||
# b"\x00\x01\x02\x03\xff" in base64 is AAECA/8=
|
||||
self.assertEqual(payload["files"][0]["content"], "AAECA/8=")
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_traversal_blocked(self, _auth, mock_api):
|
||||
# Remove active lock file to ensure it fails on traversal/invalid locks
|
||||
os.remove(self.lock_file_path)
|
||||
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[
|
||||
{
|
||||
"operation": "create",
|
||||
"path": "hack.txt",
|
||||
"workspace_path": "any.txt",
|
||||
}
|
||||
],
|
||||
message="Add hack",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Issue lock is missing", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_outside_scope_blocked(self, _auth, mock_api):
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[
|
||||
{
|
||||
"operation": "create",
|
||||
"path": "hack.txt",
|
||||
"workspace_path": "../outside.txt",
|
||||
}
|
||||
],
|
||||
message="Add hack",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("falls outside of locked worktree", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||
def test_commit_files_multiple_sources_blocked(self, _auth, mock_api):
|
||||
with patch.dict(os.environ, self._env("full-author"), clear=True):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
mcp_server.gitea_commit_files(
|
||||
files=[
|
||||
{
|
||||
"operation": "create",
|
||||
"path": "hello.txt",
|
||||
"content_plain": "Hello!",
|
||||
"content": "SGVsbG8=",
|
||||
}
|
||||
],
|
||||
message="Add hello",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Multiple content sources specified", str(ctx.exception))
|
||||
@@ -1071,9 +1071,11 @@ class TestGetFile(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCommitFiles(unittest.TestCase):
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_commit_files_success(self, _auth, mock_api):
|
||||
def test_commit_files_success(self, _auth, mock_api, _role):
|
||||
mock_api.return_value = {
|
||||
"commit": {"sha": "commit-sha-123"},
|
||||
"branch": {"name": "test-branch"}
|
||||
@@ -1081,11 +1083,15 @@ class TestCommitFiles(unittest.TestCase):
|
||||
files = [
|
||||
{"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
|
||||
]
|
||||
result = gitea_commit_files(
|
||||
files=files,
|
||||
message="Initial commit",
|
||||
new_branch="test-branch"
|
||||
)
|
||||
env = {
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.repo.commit",
|
||||
}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
result = gitea_commit_files(
|
||||
files=files,
|
||||
message="Initial commit",
|
||||
new_branch="test-branch"
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["commit"], "commit-sha-123")
|
||||
self.assertEqual(result["branch"], "test-branch")
|
||||
|
||||
Reference in New Issue
Block a user