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

feat(preflight): block workspace edits before identity/capability verification (Closes #210)

Browse Source
This commit is contained in:
sysadmin
2026-07-05 17:49:40 -04:00
parent 5b9fcaefbf
commit cbeed5c159
4 changed files with 227 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_server.py
+99
View File
@@ -156,6 +156,89 @@ PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
if PROJECT_ROOT not in sys.path:
sys.path.insert(0, PROJECT_ROOT)
import json
_preflight_whoami_called = False
_preflight_capability_called = False
_preflight_whoami_violation = False
_preflight_capability_violation = False
_preflight_resolved_role = None
def record_preflight_check(type_name: str, resolved_role: str | None = None):
"""Record a pre-flight check (whoami or capability) and check for workspace edits."""
global _preflight_whoami_called, _preflight_capability_called
global _preflight_whoami_violation, _preflight_capability_violation
global _preflight_resolved_role
in_test = "pytest" in sys.modules or "unittest" in sys.modules
if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
is_dirty = True
elif in_test:
is_dirty = False
else:
is_dirty = False
try:
res = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True, text=True, cwd=PROJECT_ROOT
)
for line in res.stdout.splitlines():
if line and not line.startswith("??"):
is_dirty = True
break
except Exception:
pass
if is_dirty:
if type_name == "whoami" and not _preflight_whoami_called:
_preflight_whoami_violation = True
if type_name == "capability" and not _preflight_capability_called:
_preflight_capability_violation = True
if type_name == "whoami":
_preflight_whoami_called = True
elif type_name == "capability":
_preflight_capability_called = True
if resolved_role:
_preflight_resolved_role = resolved_role
def verify_preflight_purity(remote: str | None = None):
"""Verify that identity and capability were verified prior to edits, and that reviewers made no edits."""
in_test = "pytest" in sys.modules or "unittest" in sys.modules
if in_test and not os.environ.get("GITEA_TEST_FORCE_DIRTY"):
return
if not _preflight_whoami_called:
raise RuntimeError("Pre-flight order violation: Identity (gitea_whoami) has not been verified (fail closed)")
if not _preflight_capability_called:
raise RuntimeError("Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)")
if _preflight_whoami_violation:
raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_whoami verification (fail closed)")
if _preflight_capability_violation:
raise RuntimeError("Pre-flight order violation: Workspace file edits occurred before gitea_resolve_task_capability verification (fail closed)")
if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
is_dirty = True
elif in_test:
is_dirty = False
else:
is_dirty = False
try:
res = subprocess.run(
["git", "status", "--porcelain"],
capture_output=True, text=True, cwd=PROJECT_ROOT
)
for line in res.stdout.splitlines():
if line and not line.startswith("??"):
is_dirty = True
break
except Exception:
pass
if _preflight_resolved_role == "reviewer" and is_dirty:
raise RuntimeError("Reviewer role violation: Reviewer profile is forbidden from modifying tracked workspace files (fail closed)")
from mcp.server.fastmcp import FastMCP # noqa: E402
from gitea_auth import ( # noqa: E402
@@ -465,6 +548,7 @@ def gitea_create_issue(
"number": None,
"reasons": block_reasons,
}
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
@@ -531,6 +615,7 @@ def gitea_create_pr(
Returns:
dict with 'number' of the created PR ('url' only with the reveal opt-in).
"""
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls"
@@ -1141,6 +1226,7 @@ def gitea_submit_pr_review(
authenticated user, profile name, PR author, PR number, head SHA
checked, and the reasons/gates passed or blocked. Never secrets.
"""
verify_preflight_purity(remote)
action = (action or "").strip().lower()
result = {
"requested_action": action,
@@ -1294,6 +1380,8 @@ def gitea_edit_pr(
if not payload:
raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
verify_preflight_purity(remote)
# PR closure is a first-class capability, distinct from retitling or
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
# never touches the network.
@@ -1411,6 +1499,7 @@ def gitea_commit_files(
Returns:
dict with success status and commit/branch information.
"""
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/contents"
@@ -1505,6 +1594,7 @@ def gitea_merge_pr(
reasons/gates passed or blocked, and merge result / merge commit if
available. Never secrets.
"""
verify_preflight_purity(remote)
do = (do or "").strip().lower()
result = {
"performed": False,
@@ -1884,6 +1974,7 @@ def gitea_delete_branch(
Returns:
dict with 'success' and 'message'.
"""
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
import urllib.parse
@@ -1915,6 +2006,7 @@ def gitea_close_issue(
Returns:
dict with 'success' boolean and 'message'.
"""
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -2213,6 +2305,7 @@ def gitea_create_issue_comment(
(permission blocks also carry a structured 'permission_report',
#142).
"""
verify_preflight_purity(remote)
gate_reasons = _profile_operation_gate("gitea.issue.comment")
reasons = list(gate_reasons)
if not (body or "").strip():
@@ -2781,6 +2874,7 @@ def gitea_whoami(
'email', 'server', 'remote', and 'profile' (safe runtime profile
metadata: profile_name + allowed_operations; never the token).
"""
record_preflight_check("whoami")
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
h = host or REMOTES[remote]["host"]
@@ -3300,6 +3394,7 @@ def gitea_mark_issue(
if action not in ("start", "done"):
raise ValueError(f"action must be 'start' or 'done', got '{action}'")
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
@@ -3382,6 +3477,7 @@ def gitea_create_label(
Returns:
dict containing the created label details.
"""
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
@@ -3421,6 +3517,7 @@ def gitea_set_issue_labels(
Returns:
list of all labels currently applied to the issue.
"""
verify_preflight_purity(remote)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
@@ -3691,6 +3788,8 @@ def gitea_resolve_task_capability(
required_permission = TASK_MAP[task]["permission"]
required_role = TASK_MAP[task]["role"]
record_preflight_check("capability", required_role)
profile = get_profile()
config = gitea_config.load_config()
review_proofs.py
+20 -2
View File
@@ -749,6 +749,7 @@ HANDOFF_BASE_FIELDS = (
("Files changed", ("files changed", "changed", "files")),
("Validation", ("validation",)),
("Mutations", ("mutations",)),
("Workspace mutations", ("workspace mutations",)),
("Current status", ("current status", "status")),
("Blockers", ("blockers",)),
("Next", ("next",)),
@@ -799,7 +800,7 @@ def _handoff_section_lines(report_text):
return lines[start:]
def assess_controller_handoff(report_text, role=None):
def assess_controller_handoff(report_text, role=None, local_edits=False):
"""Issue #182: final reports without a Controller Handoff downgrade.
Verdicts:
@@ -824,10 +825,14 @@ def assess_controller_handoff(report_text, role=None):
}
labels = []
fields_dict = {}
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
k, v = stripped.split(":", 1)
label = k.strip().lower()
labels.append(label)
fields_dict[label] = v.strip()
required = list(HANDOFF_BASE_FIELDS)
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
@@ -846,6 +851,19 @@ def assess_controller_handoff(report_text, role=None):
"reasons": [f"handoff missing required field: {m}"
for m in missing],
}
if local_edits:
workspace_mutations_val = fields_dict.get("workspace mutations", "").strip().lower()
if not workspace_mutations_val or workspace_mutations_val == "none":
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": ["Workspace mutations"],
"reasons": [
"Workspace mutations cannot be 'none' or empty when local edits exist"
],
}
return {
"verdict": "complete",
"downgraded": False,
tests/test_mcp_server.py
+84
View File
@@ -2427,3 +2427,87 @@ class TestVerifyMutationAuthority(unittest.TestCase):
with patch.dict(os.environ,
{"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
mcp_server.verify_mutation_authority("prgs")
# ---------------------------------------------------------------------------
# Pre-flight ordering and workspace edit block (#210)
# ---------------------------------------------------------------------------
class TestPreflightVerification(unittest.TestCase):
"""Assert that pre-flight verification gates order correctly."""
def setUp(self):
# Save real global variables
import mcp_server
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
# Reset state for each test
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
def tearDown(self):
# Restore real global variables
import mcp_server
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
if "GITEA_TEST_FORCE_DIRTY" in os.environ:
del os.environ["GITEA_TEST_FORCE_DIRTY"]
def test_preflight_whoami_violation(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server._preflight_capability_called = True
mcp_server.record_preflight_check("whoami")
self.assertTrue(mcp_server._preflight_whoami_violation)
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Workspace file edits occurred before gitea_whoami verification", str(ctx.exception))
def test_preflight_capability_violation(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
mcp_server._preflight_whoami_called = True
mcp_server.record_preflight_check("capability", resolved_role="author")
self.assertTrue(mcp_server._preflight_capability_violation)
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Workspace file edits occurred before gitea_resolve_task_capability verification", str(ctx.exception))
def test_preflight_not_called_fails_closed(self):
import mcp_server
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Identity (gitea_whoami) has not been verified", str(ctx.exception))
mcp_server._preflight_whoami_called = True
with self.assertRaises(RuntimeError) as ctx2:
mcp_server.verify_preflight_purity()
self.assertIn("Task capability (gitea_resolve_task_capability) has not been resolved", str(ctx2.exception))
def test_preflight_reviewer_mutation_violation(self):
import mcp_server
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "reviewer"
# When dirty, reviewer edits are blocked
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
with self.assertRaises(RuntimeError) as ctx:
mcp_server.verify_preflight_purity()
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
# When clean, reviewer is allowed
del os.environ["GITEA_TEST_FORCE_DIRTY"]
mcp_server.verify_preflight_purity()
tests/test_review_proofs.py
+24
View File
@@ -773,6 +773,7 @@ class TestControllerHandoff(unittest.TestCase):
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: one PR opened",
"- Workspace mutations: none",
"- Current status: PR open",
"- Blockers: none",
"- Next: review PR #999",
@@ -859,6 +860,29 @@ class TestControllerHandoff(unittest.TestCase):
self.assertIn("assess_controller_handoff", skill)
self.assertIn("issue #182", skill)
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
class TestPRInventoryTrustGate(unittest.TestCase):
"""Issue #194: unit tests for the PR inventory trust gate."""