Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix(reviewer-workflow): address review — in-process mutation authority, no /tmp lock (#199)

Browse Source 
Addresses the sysadmin REQUEST_CHANGES on PR #203 (reviewed head
10d2644790):

1. Lock redesigned; /tmp file removed entirely. The mutation authority is
   now an in-process record (_MUTATION_AUTHORITY) plus an environment
   session lock (GITEA_SESSION_PROFILE_LOCK) exported at server launch:
   - in-process record cannot be spoofed by other local processes, cannot
     go stale across sessions, and cannot race concurrent agents;
   - the env lock is inherited by child CLI processes, so review_pr.py can
     refuse an ad-hoc GITEA_MCP_PROFILE role escalation without any shared
     file; a missing env lock (direct operator CLI use) stays allowed;
   - silent except-pass writes are gone; an unresolvable profile fails
     closed.
2. Standard reviewer workflow unbroken: verify_mutation_authority seeds
   itself from the live config-resolved context at the first mutation gate
   (approved preflight path whoami -> eligibility -> review/merge), and now
   runs as the final gate after eligibility, reusing the identity that
   eligibility proved (no extra /user call).
3. Trailing whitespace removed from review_pr.py (git diff --check clean).
4. Module-global verify_mutation_authority no-op bypass removed from
   tests/test_mcp_server.py; replaced with a tests/conftest.py autouse
   fixture that only resets per-process state (_MUTATION_AUTHORITY,
   _IDENTITY_CACHE, session lock env) between tests — the gate itself
   stays live in every test.
5. Tests rewritten for the new design: seeding on first verify, unresolved
   profile fails closed, remote/profile/identity mismatches fail closed,
   session-lock env mismatch rejected, foreign-pid authority reseeded,
   unauthorized author->reviewer pivot blocked, authorized pivot allowed;
   CLI: mismatch blocked, match allowed, no-lock allowed.
6. Rebased onto current master (c6fd0fd).

Closes #199
Refs #194

Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
sysadmin
2026-07-05 16:51:34 -04:00
co-authored by Claude Fable 5
parent 871d9d8590
commit 30cff19a41
5 changed files with 303 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+28
View File
@@ -0,0 +1,28 @@
"""Shared pytest fixtures for the Gitea-Tools test suite."""
import sys
import pytest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@pytest.fixture(autouse=True)
def _reset_mutation_authority(monkeypatch):
"""Isolate the in-process mutation authority between tests (#199).
The mutation-authority gate stays LIVE in every test — this fixture only
clears the per-process record and the session profile lock so one test's
seeded authority (or an intentionally mismatched one) cannot leak into
the next test. It must never replace verify_mutation_authority with a
no-op: individual tests that need a specific authority state set it up
explicitly.
"""
monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
try:
import mcp_server
except Exception:
yield
return
monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
yield
tests/test_mcp_server.py
+91 -62
View File
@@ -38,9 +38,6 @@ from gitea_auth import get_profile # noqa: E402
import gitea_config # noqa: E402
import mcp_server
# Globally disable verification check for existing isolated unit tests
_real_verify = mcp_server.verify_mutation_authority
mcp_server.verify_mutation_authority = lambda *args, **kwargs: None
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
@@ -2304,87 +2301,119 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
class TestVerifyMutationAuthority(unittest.TestCase):
"""Test verification lock logic under various configurations."""
"""In-process mutation authority (#199, refs #194).
The authority record lives in mcp_server._MUTATION_AUTHORITY (per
process, reset between tests by conftest); the CLI side-channel is
covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
lock file — nothing here touches /tmp.
"""
def setUp(self):
self.patch_profile = patch("mcp_server.get_profile")
self.mock_profile = self.patch_profile.start()
self.patch_username = patch("mcp_server._authenticated_username")
self.mock_username = self.patch_username.start()
# Restore real function for these tests
self._old_verify = mcp_server.verify_mutation_authority
mcp_server.verify_mutation_authority = _real_verify
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
def tearDown(self):
self.patch_profile.stop()
self.patch_username.stop()
mcp_server.verify_mutation_authority = self._old_verify
# Clean up lock file
if os.path.exists("/tmp/gitea_mutation_authority.lock"):
os.remove("/tmp/gitea_mutation_authority.lock")
def test_missing_lock_fails_closed(self):
if os.path.exists("/tmp/gitea_mutation_authority.lock"):
os.remove("/tmp/gitea_mutation_authority.lock")
def _authority(self, **overrides):
data = {
"initial_profile": "prgs-reviewer",
"initial_identity": "sysadmin",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin",
"remote": "prgs",
"task": "review_pr",
"role_pivot_authorized": False,
"role_pivot_record": None,
"pid": os.getpid(),
}
data.update(overrides)
mcp_server._MUTATION_AUTHORITY = data
def test_missing_authority_seeds_from_live_context(self):
# Approved preflight path (whoami → eligibility → mutation): the
# first mutation gate seeds the authority instead of failing closed,
# so the standard reviewer workflow keeps working.
mcp_server._MUTATION_AUTHORITY = None
mcp_server.verify_mutation_authority("prgs")
seeded = mcp_server._MUTATION_AUTHORITY
self.assertIsNotNone(seeded)
self.assertEqual(seeded["current_profile"], "prgs-reviewer")
self.assertEqual(seeded["current_identity"], "sysadmin")
self.assertEqual(seeded["remote"], "prgs")
def test_unresolved_profile_fails_closed(self):
self.mock_profile.return_value = {}
mcp_server._MUTATION_AUTHORITY = None
with self.assertRaises(RuntimeError) as ctx:
_real_verify("prgs")
self.assertIn("lock is missing", str(ctx.exception))
mcp_server.verify_mutation_authority("prgs")
self.assertIn("profile unresolved", str(ctx.exception))
def test_mismatched_remote_fails(self):
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "dadeschools",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin"
}, f)
self._authority(remote="dadeschools")
with self.assertRaises(RuntimeError) as ctx:
_real_verify("prgs")
mcp_server.verify_mutation_authority("prgs")
self.assertIn("does not match locked remote", str(ctx.exception))
def test_mismatched_profile_fails(self):
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "prgs",
"current_profile": "prgs-author",
"current_identity": "jcwalker3"
}, f)
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
def test_profile_flip_after_record_fails(self):
# Authority was recorded as author; the active profile now resolves
# as reviewer (e.g. an env-var flip mid-session) — refuse.
self._authority(
initial_profile="prgs-author",
initial_identity="jcwalker3",
current_profile="prgs-author",
current_identity="jcwalker3",
)
with self.assertRaises(RuntimeError) as ctx:
_real_verify("prgs")
mcp_server.verify_mutation_authority("prgs")
self.assertIn("does not match locked authority", str(ctx.exception))
def test_author_to_reviewer_pivot_blocked_without_authorization(self):
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "prgs",
"initial_profile": "prgs-author",
"initial_identity": "jcwalker3",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin",
"role_pivot_authorized": False
}, f)
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
def test_session_lock_env_mismatch_fails(self):
# The launching session locked the environment to the author
# profile; the active profile resolves as reviewer — side-channel
# override rejected even with a matching in-process authority.
self._authority()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_mutation_authority("prgs")
self.assertIn("side-channel override rejected", str(ctx.exception))
def test_foreign_pid_authority_is_not_trusted(self):
# An authority record from another process (fork leftovers) is
# discarded and reseeded from the live context, never reused.
self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
mcp_server.verify_mutation_authority("prgs")
self.assertEqual(
mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
)
self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
def test_author_to_reviewer_pivot_blocked_without_authorization(self):
self._authority(
initial_profile="prgs-author",
initial_identity="jcwalker3",
)
with self.assertRaises(RuntimeError) as ctx:
_real_verify("prgs", required_role="reviewer")
mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
self.assertIn("without authorized role pivot", str(ctx.exception))
def test_allowed_when_match(self):
with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f:
json.dump({
"remote": "prgs",
"initial_profile": "prgs-reviewer",
"initial_identity": "sysadmin",
"current_profile": "prgs-reviewer",
"current_identity": "sysadmin",
"role_pivot_authorized": False
}, f)
self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
self.mock_username.return_value = "sysadmin"
def test_authorized_pivot_is_allowed(self):
self._authority(
initial_profile="prgs-author",
initial_identity="jcwalker3",
role_pivot_authorized=True,
)
mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
# Should pass without exception
_real_verify("prgs")
def test_allowed_when_match(self):
self._authority()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
mcp_server.verify_mutation_authority("prgs")
tests/test_review_pr.py
+27 -42
View File
@@ -112,39 +112,23 @@ class TestAPIPayload(unittest.TestCase):
class TestMutationAuthorityLock(unittest.TestCase):
"""Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock."""
"""#199 (refs #194): the CLI refuses to run under a profile that differs
from the session profile lock exported by the launching MCP session."""
@patch("review_pr.get_profile")
def test_cli_blocked_on_profile_mismatch(self, mock_get_profile):
def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
# An author-bound session exported the lock; the CLI resolves a
# reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
# before any API call.
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
original_exists = os.path.exists
def conditional_exists(path):
if "gitea_mutation_authority.lock" in str(path):
return True
return original_exists(path)
original_open = open
def conditional_open(file, *args, **kwargs):
if "gitea_mutation_authority.lock" in str(file):
return io.StringIO('{"current_profile": "prgs-author"}')
return original_open(file, *args, **kwargs)
from _pytest.monkeypatch import MonkeyPatch
import io
buf = io.StringIO()
monkeypatch = MonkeyPatch()
monkeypatch.setattr(sys, "stderr", buf)
with patch("os.path.exists", side_effect=conditional_exists), \
patch("builtins.open", side_effect=conditional_open):
try:
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
finally:
monkeypatch.undo()
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}), \
patch.object(sys, "stderr", buf):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 3)
msg = buf.getvalue().lower()
self.assertIn("cli override rejected", msg)
@@ -155,21 +139,22 @@ class TestMutationAuthorityLock(unittest.TestCase):
def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
mock_api.side_effect = [FAKE_PR_DATA, {}]
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 0)
original_exists = os.path.exists
def conditional_exists(path):
if "gitea_mutation_authority.lock" in str(path):
return True
return original_exists(path)
original_open = open
def conditional_open(file, *args, **kwargs):
if "gitea_mutation_authority.lock" in str(file):
return io.StringIO('{"current_profile": "prgs-reviewer"}')
return original_open(file, *args, **kwargs)
with patch("os.path.exists", side_effect=conditional_exists), \
patch("builtins.open", side_effect=conditional_open):
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_cli_allowed_without_session_lock(self, _auth, mock_api):
# No lock in the environment = direct operator CLI use; the wall
# does not apply and the normal flow proceeds.
mock_api.side_effect = [FAKE_PR_DATA, {}]
env = {k: v for k, v in os.environ.items()
if k != "GITEA_SESSION_PROFILE_LOCK"}
with patch.dict(os.environ, env, clear=True):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])