sysadminandClaude Opus 4.8
795f544047
Add lock_provenance metadata on gitea_lock_issue writes and fail closed at gitea_create_pr when provenance is missing. Final-report validation now requires explicit External-state mutations disclosure for issue-lock read/write/delete and blocks mixed author PR creation with reviewer approval. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
275 lines
10 KiB
Python
275 lines
10 KiB
Python
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))
|
|
|
|
repo_root = os.path.realpath(
|
|
os.path.join(os.path.dirname(__file__), os.pardir)
|
|
)
|
|
branches_root = os.path.join(repo_root, "branches")
|
|
os.makedirs(branches_root, exist_ok=True)
|
|
self.locked_worktree_dir = tempfile.TemporaryDirectory(
|
|
dir=branches_root,
|
|
prefix="test-commit-payloads-",
|
|
)
|
|
self.locked_worktree_path = os.path.realpath(self.locked_worktree_dir.name)
|
|
|
|
self.lock_file_path = "/tmp/gitea_issue_lock.json"
|
|
import issue_lock_provenance
|
|
|
|
work_lease = {
|
|
"operation_type": "author_issue_work",
|
|
"issue_number": 263,
|
|
"branch": "feat/issue-263-native-commit-payloads",
|
|
"claimant": {"username": "test-user", "profile": "test-author"},
|
|
"expires_at": "2999-01-01T00:00:00Z",
|
|
}
|
|
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,
|
|
"work_lease": work_lease,
|
|
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
|
|
tool="gitea_lock_issue",
|
|
claimant=work_lease.get("claimant"),
|
|
),
|
|
}
|
|
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": "",
|
|
"GITEA_AUTHOR_WORKTREE": self.locked_worktree_path,
|
|
}
|
|
|
|
@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))
|