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

Merge branch 'master' into feat/issue-605-workflow-dashboard

Browse Source
This commit is contained in:
jcwalker3
2026-07-19 18:54:17 -05:00
parent 7f2b9f36de b7a5284b98
commit d181d499d3
3 changed files with 454 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+29 -2
View File
@@ -701,6 +701,18 @@ def _clear_preflight_capability_state() -> None:
_preflight_reviewer_violation_files = []
def _invalidate_preflight_identity_state() -> None:
"""Fail closed when whoami cannot prove the configured identity."""
global _preflight_whoami_called, _preflight_whoami_violation
global _preflight_whoami_baseline_porcelain, _preflight_whoami_violation_files
_preflight_whoami_called = False
_preflight_whoami_violation = False
_preflight_whoami_baseline_porcelain = None
_preflight_whoami_violation_files = []
_clear_preflight_capability_state()
def record_preflight_check(
type_name: str,
resolved_role: str | None = None,
@@ -1101,6 +1113,8 @@ def _run_anti_stomp_preflight(
# Workflow-hash facts when the task is review/merge oriented.
if workflow_hash_valid is None and task in {
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"review_pr",
"submit_pr_review",
"approve_pr",
@@ -1236,7 +1250,9 @@ def verify_preflight_purity(
if (
task is not None
and _preflight_resolved_task is not None
and task != _preflight_resolved_task
and not task_capability_map.preflight_task_matches(
_preflight_resolved_task, task
)
):
raise _PreflightOrderError(
"Pre-flight task mismatch: "
@@ -14163,6 +14179,8 @@ def gitea_whoami(
"identity_match": not id_match.get("block"),
"identity_match_reasons": id_match.get("reasons") or [],
}
if id_match.get("block"):
_invalidate_preflight_identity_state()
if _reveal_endpoints():
result["server"] = gitea_url(h, "").rstrip("/")
return result
@@ -17731,7 +17749,12 @@ def gitea_resolve_task_capability(
result["stale_binding_recovery"] = stale_binding
if reason_msg:
result["reason"] = reason_msg
if task in ("review_pr", "merge_pr"):
if task in (
"acquire_reviewer_pr_lease",
"gitea_acquire_reviewer_pr_lease",
"review_pr",
"merge_pr",
):
result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
PROJECT_ROOT)
if not result["workflow_load_proof"].get("workflow_load_valid"):
@@ -17754,6 +17777,10 @@ def gitea_resolve_task_capability(
)
elif was_terminal and not (stop_required or restart_required):
result["cleared_stale_denial"] = True
if stop_required or restart_required:
# A denied or stale resolver result is diagnostic evidence, never a
# consumable capability token for a later mutation (#763).
_clear_preflight_capability_state()
return result
task_capability_map.py
+30
View File
@@ -432,6 +432,36 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
},
}
# A reviewer lease is the first mutation in the canonical ``review_pr``
# workflow, so the already-resolved review capability is valid for that one
# narrower transition. Keep this directed and explicit: lease acquisition
# does not authorize a review verdict, and reviewer proof never authorizes a
# merger lease (#763).
_PREFLIGHT_TASK_TRANSITIONS = frozenset({
("review_pr", "acquire_reviewer_pr_lease"),
})
def _canonical_preflight_task(task: str | None) -> str:
"""Normalize only declared ``gitea_`` aliases for preflight comparison."""
value = (task or "").strip()
if value.startswith("gitea_") and value[6:] in TASK_CAPABILITY_MAP:
return value[6:]
return value
def preflight_task_matches(
resolved_task: str | None,
mutation_task: str | None,
) -> bool:
"""Return whether capability proof authorizes this mutation transition."""
resolved = _canonical_preflight_task(resolved_task)
mutation = _canonical_preflight_task(mutation_task)
if not resolved or not mutation:
return False
return resolved == mutation or (resolved, mutation) in _PREFLIGHT_TASK_TRANSITIONS
# Issue-mutating MCP tools and their resolver task keys.
ISSUE_MUTATION_TOOL_TASKS: dict[str, str] = {
"gitea_create_issue": "create_issue",
tests/test_issue_763_reviewer_lease_preflight_order.py
+395
View File
@@ -0,0 +1,395 @@
"""Regression coverage for reviewer-lease preflight ordering (#763)."""
from __future__ import annotations
from datetime import datetime, timezone
from unittest.mock import patch
import pytest
import anti_stomp_preflight
import gitea_mcp_server as server
import merger_lease_adoption
import reviewer_pr_lease
import task_capability_map
def _prime_clean_reviewer_preflight(monkeypatch, resolved_task: str) -> None:
"""Install a clean reviewer preflight without bypassing task matching."""
monkeypatch.setenv("GITEA_TEST_PORCELAIN", "")
monkeypatch.delenv("GITEA_TEST_FORCE_DIRTY", raising=False)
monkeypatch.setattr(server, "_preflight_in_test_mode", lambda: False)
monkeypatch.setattr(server, "_process_start_porcelain", "")
monkeypatch.setattr(server, "_preflight_whoami_called", False)
monkeypatch.setattr(server, "_preflight_capability_called", False)
monkeypatch.setattr(server, "_preflight_whoami_violation", False)
monkeypatch.setattr(server, "_preflight_capability_violation", False)
monkeypatch.setattr(server, "_preflight_resolved_role", None)
monkeypatch.setattr(server, "_preflight_resolved_task", None)
monkeypatch.setattr(server, "_preflight_whoami_baseline_porcelain", None)
monkeypatch.setattr(server, "_preflight_capability_baseline_porcelain", None)
monkeypatch.setattr(server, "_preflight_whoami_violation_files", [])
monkeypatch.setattr(server, "_preflight_capability_violation_files", [])
monkeypatch.setattr(server, "_preflight_reviewer_violation_files", [])
monkeypatch.setattr(
server,
"_resolve_namespace_mutation_context",
lambda _worktree=None: {
"workspace_path": server.PROJECT_ROOT,
"canonical_repo_root": server.PROJECT_ROOT,
"process_project_root": server.PROJECT_ROOT,
"workspace_role_kind": "reviewer",
"workspace_binding_source": "test reviewer binding",
"ignored_bindings": [],
},
)
monkeypatch.setattr(server, "_enforce_stable_branch_contamination_gate", lambda *_a: None)
monkeypatch.setattr(server, "_enforce_canonical_repository_root", lambda *_a, **_k: None)
monkeypatch.setattr(server, "_enforce_root_checkout_guard", lambda *_a: None)
monkeypatch.setattr(server, "_enforce_branches_only_author_mutation", lambda *_a, **_k: None)
monkeypatch.setattr(server, "_enforce_issue_scope_guard", lambda *_a, **_k: None)
monkeypatch.setattr(server, "_create_issue_bootstrap_assessment", lambda *_a: None)
monkeypatch.setattr(server, "_run_anti_stomp_preflight", lambda *_a, **_k: None)
server.record_preflight_check("whoami")
server.record_preflight_check(
"capability", resolved_role="reviewer", resolved_task=resolved_task
)
def test_documented_review_capability_allows_reviewer_lease_acquire(monkeypatch):
"""whoami -> resolve(review_pr) -> acquire reviewer lease is canonical."""
_prime_clean_reviewer_preflight(monkeypatch, "review_pr")
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
assert server._preflight_capability_called is False
def test_exact_lease_capability_without_intervening_call_still_succeeds(monkeypatch):
_prime_clean_reviewer_preflight(monkeypatch, "acquire_reviewer_pr_lease")
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
assert server._preflight_capability_called is False
def test_missing_wrong_and_consumed_capability_fail_closed(monkeypatch):
_prime_clean_reviewer_preflight(monkeypatch, "create_issue")
with pytest.raises(RuntimeError, match="task mismatch"):
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
_prime_clean_reviewer_preflight(monkeypatch, "acquire_reviewer_pr_lease")
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
with pytest.raises(RuntimeError, match="has not been resolved"):
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
def test_documented_intervening_whoami_read_preserves_capability(monkeypatch):
_prime_clean_reviewer_preflight(monkeypatch, "review_pr")
with patch.object(server, "_get_workspace_porcelain", return_value=""):
server.record_preflight_check("whoami")
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
def test_reviewer_transition_is_narrow_alias_aware_and_one_way():
assert task_capability_map.preflight_task_matches(
"review_pr", "gitea_acquire_reviewer_pr_lease"
)
assert task_capability_map.preflight_task_matches(
"gitea_acquire_reviewer_pr_lease", "acquire_reviewer_pr_lease"
)
assert not task_capability_map.preflight_task_matches(
"acquire_reviewer_pr_lease", "review_pr"
)
assert not task_capability_map.preflight_task_matches(
"review_pr", "acquire_merger_pr_lease"
)
assert not task_capability_map.preflight_task_matches(
"merge_pr", "acquire_reviewer_pr_lease"
)
def test_dirty_reviewer_workspace_still_fails_closed(monkeypatch):
_prime_clean_reviewer_preflight(monkeypatch, "review_pr")
monkeypatch.setenv("GITEA_TEST_PORCELAIN", " M gitea_mcp_server.py\n")
with pytest.raises(RuntimeError, match="Reviewer role violation"):
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
def test_mismatched_reviewer_workspace_still_fails_closed(monkeypatch):
_prime_clean_reviewer_preflight(monkeypatch, "review_pr")
monkeypatch.setattr(
server,
"_resolve_namespace_mutation_context",
lambda _worktree=None: {
"workspace_path": "/outside/review-pr-762",
"canonical_repo_root": "/repo",
"process_project_root": "/repo",
"workspace_role_kind": "reviewer",
"workspace_binding_source": "test reviewer binding",
"ignored_bindings": [],
},
)
monkeypatch.setattr(
server.author_mutation_worktree,
"assess_workspace_repo_membership",
lambda **_kwargs: {"block": True, "reasons": ["workspace mismatch"]},
)
monkeypatch.setattr(
server.author_mutation_worktree,
"format_workspace_repo_membership_error",
lambda _assessment: "workspace mismatch (fail closed)",
)
with pytest.raises(RuntimeError, match="workspace mismatch"):
server.verify_preflight_purity(task="acquire_reviewer_pr_lease")
def test_reviewer_lease_acquire_requires_workflow_load_proof(monkeypatch):
sha = "a" * 40
monkeypatch.setattr(server, "_anti_stomp_in_test_mode", lambda: False)
monkeypatch.setattr(
server,
"get_profile",
lambda: {
"profile_name": "prgs-reviewer",
"role": "reviewer",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.pr.review",
],
},
)
monkeypatch.setattr(server, "_actual_profile_role", lambda: "reviewer")
monkeypatch.setattr(
server,
"_resolve_namespace_mutation_context",
lambda _worktree=None: {
"workspace_path": "/repo/branches/review-pr-762",
"canonical_repo_root": "/repo",
"process_project_root": "/repo",
},
)
monkeypatch.setattr(
server.issue_lock_worktree,
"read_worktree_git_state",
lambda _path: {
"current_branch": "master",
"head_sha": sha,
"porcelain_status": "",
},
)
monkeypatch.setattr(
server.root_checkout_guard,
"resolve_remote_master_sha",
lambda _path: sha,
)
monkeypatch.setattr(
server,
"_current_master_parity",
lambda: {"startup_head": sha, "current_head": sha},
)
monkeypatch.setattr(server, "_local_git_remote_url", lambda _remote: None)
monkeypatch.setattr(server, "_load_stable_contamination_marker", lambda _remote: None)
monkeypatch.setattr(
server,
"_review_workflow_load_gate_reasons",
lambda: ["canonical review workflow proof missing"],
)
with pytest.raises(RuntimeError, match="workflow"):
server._run_anti_stomp_preflight(
"acquire_reviewer_pr_lease",
remote="prgs",
worktree_path="/repo/branches/review-pr-762",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
def test_whoami_identity_mismatch_invalidates_preflight(monkeypatch):
monkeypatch.setenv("GITEA_TEST_PORCELAIN", "")
monkeypatch.setattr(server, "_process_start_porcelain", "")
monkeypatch.setattr(server, "_preflight_whoami_called", False)
monkeypatch.setattr(server, "_preflight_capability_called", True)
monkeypatch.setattr(server, "_auth", lambda _host: "redacted")
monkeypatch.setattr(
server,
"api_request",
lambda *_args, **_kwargs: {"login": "wrong-reviewer", "id": 7},
)
monkeypatch.setattr(
server,
"get_profile",
lambda: {
"profile_name": "prgs-reviewer",
"role": "reviewer",
"username": "sysadmin",
"allowed_operations": ["gitea.read", "gitea.pr.review"],
"forbidden_operations": [],
},
)
monkeypatch.setattr(server, "_seed_session_context", lambda **_kwargs: None)
monkeypatch.setattr(server.session_ctx, "mutation_context_audit_fields", lambda: {})
monkeypatch.setattr(server, "_reveal_endpoints", lambda: False)
result = server.gitea_whoami(remote="prgs")
assert result["identity_match"] is False
assert server._preflight_whoami_called is False
assert server._preflight_capability_called is False
def test_denied_reviewer_profile_does_not_leave_capability_proof(monkeypatch):
profile = {
"profile_name": "prgs-author",
"role": "author",
"username": "jcwalker3",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.pr.review",
],
"forbidden_operations": [],
}
monkeypatch.setenv("GITEA_TEST_PORCELAIN", "")
monkeypatch.setattr(server, "_process_start_porcelain", "")
monkeypatch.setattr(server, "get_profile", lambda: profile)
monkeypatch.setattr(
server.gitea_config,
"load_config",
lambda: {"profiles": {"prgs-author": profile}},
)
monkeypatch.setattr(server.gitea_config, "is_runtime_switching_enabled", lambda: False)
monkeypatch.setattr(server, "_authenticated_username", lambda _host: "jcwalker3")
monkeypatch.setattr(server, "_seed_session_context", lambda **_kwargs: None)
monkeypatch.setattr(
server.session_ctx,
"assess_session_context",
lambda **_kwargs: {"block": False, "reasons": []},
)
monkeypatch.setattr(
server.session_ctx,
"assess_identity_match",
lambda **_kwargs: {"block": False, "reasons": []},
)
monkeypatch.setattr(
server.session_ctx,
"profile_allowed_for_remote",
lambda *_args, **_kwargs: {"block": False, "reasons": []},
)
monkeypatch.setattr(server.session_ctx, "mutation_context_audit_fields", lambda: {})
monkeypatch.setattr(
server.role_session_router,
"assess_infra_stop",
lambda _root: {"infra_stop": False, "infra_stop_reasons": []},
)
monkeypatch.setattr(server, "_check_mcp_runtimes_diagnostics", lambda *_a: [])
monkeypatch.setattr(
server,
"_assess_stale_active_binding",
lambda **_kwargs: {"classification": "unbound"},
)
monkeypatch.setattr(server, "record_mutation_authority", lambda *_args: None)
monkeypatch.setattr(server, "init_review_decision_lock", lambda *_a, **_k: None)
monkeypatch.setattr(server.capability_stop_terminal, "is_active", lambda: False)
monkeypatch.setattr(
server.capability_stop_terminal,
"sync_from_capability_result",
lambda _result: False,
)
result = server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
assert result["allowed_in_current_session"] is False
assert result["required_role_kind"] == "reviewer"
assert server._preflight_capability_called is False
def test_head_and_foreign_lease_protections_remain_enforced():
now = datetime.now(timezone.utc)
head = "a" * 40
moved_head = "b" * 40
body = reviewer_pr_lease.format_lease_body(
repo="Scaled-Tech-Consulting/Gitea-Tools",
pr_number=762,
issue_number=605,
reviewer_identity="other-reviewer",
profile="prgs-reviewer",
session_id="foreign-session",
worktree="/repo/branches/review-pr-762",
phase="claimed",
candidate_head=head,
target_branch="master",
target_branch_sha="c" * 40,
last_activity=now,
)
comments = [{"id": 10, "author": "other-reviewer", "body": body}]
acquire = reviewer_pr_lease.assess_acquire_lease(
comments,
pr_number=762,
reviewer_identity="sysadmin",
profile="prgs-reviewer",
session_id="my-session",
repo="Scaled-Tech-Consulting/Gitea-Tools",
issue_number=605,
worktree="/repo/branches/review-pr-762-mine",
candidate_head=head,
target_branch="master",
target_branch_sha="c" * 40,
now=now,
)
assert acquire["acquire_allowed"] is False
reviewer_pr_lease.clear_session_lease()
reviewer_pr_lease.record_session_lease(
{
"pr_number": 762,
"session_id": "foreign-session",
"candidate_head": head,
"comment_id": 10,
},
lease_provenance=merger_lease_adoption.build_lease_provenance(
source=merger_lease_adoption.SOURCE_ACQUIRE,
comment_id=10,
),
)
try:
gate = reviewer_pr_lease.assess_mutation_lease_gate(
pr_number=762,
comments=comments,
reviewer_identity="other-reviewer",
session_id="foreign-session",
mutation="approve",
live_head_sha=moved_head,
pinned_head_sha=head,
now=now,
)
finally:
reviewer_pr_lease.clear_session_lease()
assert gate["block"] is True
assert any("head changed" in reason for reason in gate["reasons"])
def test_reviewer_lease_role_gate_is_not_weakened():
result = anti_stomp_preflight.assess_anti_stomp_preflight(
task="acquire_reviewer_pr_lease",
profile_name="prgs-author",
profile_role="author",
required_role="reviewer",
required_permission="gitea.pr.comment",
allowed_operations=["gitea.read"],
check_repo=False,
check_root_checkout=False,
check_worktree=False,
check_stale_runtime=False,
)
assert result["block"] is True
assert result["blocker_kind"] == anti_stomp_preflight.BLOCKER_WRONG_ROLE