Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6e6ca94338 | ||
|
|
2f4dec8323 |
+296
-70
@@ -9552,15 +9552,13 @@ def gitea_edit_pr(
|
||||
if closing:
|
||||
gate_reasons = _profile_operation_gate("gitea.pr.close")
|
||||
if gate_reasons:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"pr_number": pr_number,
|
||||
"requested_state": "closed",
|
||||
"required_permission": "gitea.pr.close",
|
||||
"reasons": gate_reasons,
|
||||
"permission_report": _permission_block_report("gitea.pr.close"),
|
||||
}
|
||||
return _build_operation_gate_refusal(
|
||||
"gitea.pr.close",
|
||||
gate_reasons,
|
||||
pr_number=pr_number,
|
||||
requested_state="closed",
|
||||
required_permission="gitea.pr.close",
|
||||
)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
@@ -13823,13 +13821,17 @@ def gitea_view_issue(
|
||||
|
||||
def _permission_block_report(required_operation: str,
|
||||
identity: str | None = None) -> dict:
|
||||
"""Structured, LLM-safe explanation of a permission denial (#142).
|
||||
"""Structured, LLM-safe explanation of a permission denial (#142, #897).
|
||||
|
||||
Built only after a gate has already refused; it adds guidance to the
|
||||
refusal and never widens any permission, performs network I/O, or
|
||||
raises (fail-soft: degrades to a minimal fail-closed report). Names
|
||||
configured profiles only — never auth references, tokens, endpoint
|
||||
URLs, or keychain IDs.
|
||||
|
||||
#897: never fabricate a missing permission when the active profile
|
||||
already allows the operation. That path is a diagnostic defect (the
|
||||
refusal was not a permission denial), not a cue to switch profiles.
|
||||
"""
|
||||
report = {
|
||||
"requested_operation": required_operation,
|
||||
@@ -13841,6 +13843,7 @@ def _permission_block_report(required_operation: str,
|
||||
"matching_configured_profiles": [],
|
||||
"runtime_switching_supported": False,
|
||||
"different_mcp_namespace_required": True,
|
||||
"diagnostic_defect": False,
|
||||
"exact_safe_next_action": (
|
||||
"Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; "
|
||||
"the active profile could not be resolved (fail closed)."),
|
||||
@@ -13855,6 +13858,32 @@ def _permission_block_report(required_operation: str,
|
||||
report["active_allowed_operations"] = (
|
||||
profile.get("allowed_operations") or [])
|
||||
|
||||
# #897: fail closed as a diagnostic defect when the active profile
|
||||
# already holds the operation — callers must not invent "missing".
|
||||
try:
|
||||
holds, _hold_reason = gitea_config.check_operation(
|
||||
required_operation,
|
||||
profile.get("allowed_operations") or [],
|
||||
profile.get("forbidden_operations") or [],
|
||||
)
|
||||
except Exception:
|
||||
holds = False
|
||||
if holds:
|
||||
report["missing_permission"] = None
|
||||
report["required_permission"] = required_operation
|
||||
report["diagnostic_defect"] = True
|
||||
report["different_mcp_namespace_required"] = False
|
||||
report["exact_safe_next_action"] = (
|
||||
"Diagnostic defect: the active profile already allows "
|
||||
f"{required_operation}. This is not a permission denial — "
|
||||
"inspect blocker_kind / reasons (stale-runtime or runtime-mode). "
|
||||
"Do not call gitea_activate_profile or switch MCP sessions."
|
||||
)
|
||||
report["matching_configured_profiles"] = [
|
||||
p for p in [profile.get("profile_name")] if p
|
||||
]
|
||||
return report
|
||||
|
||||
matching = []
|
||||
try:
|
||||
config = gitea_config.load_config() or {}
|
||||
@@ -13903,6 +13932,205 @@ def _permission_block_report(required_operation: str,
|
||||
return report
|
||||
|
||||
|
||||
def _reason_is_stale_runtime(reason: str) -> bool:
|
||||
"""True when *reason* is a master-parity / stale-daemon refusal (#897)."""
|
||||
r = (reason or "").lower()
|
||||
if not r:
|
||||
return False
|
||||
if "stale relative to live master" in r:
|
||||
return True
|
||||
if "server code is stale" in r:
|
||||
return True
|
||||
if "daemon is stale" in r:
|
||||
return True
|
||||
if "started at commit" in r and "workspace master is now" in r:
|
||||
return True
|
||||
if "mcp server started at" in r and "stale" in r:
|
||||
return True
|
||||
if "restart the server to load the current capability gates" in r:
|
||||
return True
|
||||
if "restart/reconnect before mutating" in r:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _reason_is_runtime_mode(reason: str) -> bool:
|
||||
"""True when *reason* is a stable-control / runtime-mode refusal (#897)."""
|
||||
r = (reason or "").lower()
|
||||
if not r:
|
||||
return False
|
||||
if _reason_is_stale_runtime(reason):
|
||||
return False
|
||||
if "runtime mode could not be assessed" in r:
|
||||
return True
|
||||
if "runtime mode is" in r:
|
||||
return True
|
||||
if "stable control runtime" in r:
|
||||
return True
|
||||
if "dev-test" in r and ("runtime" in r or "production" in r):
|
||||
return True
|
||||
if "development worktree" in r or "dev worktree" in r:
|
||||
return True
|
||||
if "launched from a 'branches/" in r or "launched from a \"branches/" in r:
|
||||
return True
|
||||
if "process-root / active-workspace alignment" in r:
|
||||
return True
|
||||
if "namespace" in r and "reproof" in r:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _reason_is_permission(reason: str) -> bool:
|
||||
"""True when *reason* is a genuine profile-permission denial (#897)."""
|
||||
r = (reason or "").lower()
|
||||
if not r:
|
||||
return False
|
||||
if _reason_is_stale_runtime(reason) or _reason_is_runtime_mode(reason):
|
||||
return False
|
||||
if "profile could not be resolved" in r:
|
||||
return True
|
||||
if "profile has no configured allowed operations" in r:
|
||||
return True
|
||||
if "profile forbids" in r:
|
||||
return True
|
||||
if "profile is not allowed to" in r:
|
||||
return True
|
||||
if "unrecognized forbidden operation" in r:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _classify_operation_gate_reasons(reasons: list[str]) -> dict:
|
||||
"""Partition gate reasons into stale / runtime-mode / permission (#897)."""
|
||||
stale: list[str] = []
|
||||
runtime_mode: list[str] = []
|
||||
permission: list[str] = []
|
||||
other: list[str] = []
|
||||
for reason in reasons or []:
|
||||
if _reason_is_stale_runtime(reason):
|
||||
stale.append(reason)
|
||||
elif _reason_is_runtime_mode(reason):
|
||||
runtime_mode.append(reason)
|
||||
elif _reason_is_permission(reason):
|
||||
permission.append(reason)
|
||||
else:
|
||||
other.append(reason)
|
||||
return {
|
||||
"stale_runtime": stale,
|
||||
"runtime_mode": runtime_mode,
|
||||
"permission": permission,
|
||||
"other": other,
|
||||
}
|
||||
|
||||
|
||||
def _stale_runtime_reconnect_action() -> str:
|
||||
"""Sanctioned recovery for a stale daemon — reconnect only (#685/#897)."""
|
||||
return (
|
||||
"Reconnect the IDE/client MCP session so the server reloads at the "
|
||||
"current master head. Do not call gitea_activate_profile or switch "
|
||||
"MCP role sessions — profile switching does not clear a stale daemon."
|
||||
)
|
||||
|
||||
|
||||
def _build_operation_gate_refusal(
|
||||
required_operation: str,
|
||||
reasons: list[str],
|
||||
**extra_fields,
|
||||
) -> dict:
|
||||
"""Structured gate refusal with typed blockers (#897).
|
||||
|
||||
Stale-runtime and runtime-mode refusals never attach a
|
||||
``permission_report`` and never recommend profile switching. True
|
||||
permission denials still get ``permission_report``. When both apply,
|
||||
causes are reported separately under distinct fields.
|
||||
"""
|
||||
classified = _classify_operation_gate_reasons(reasons)
|
||||
stale = classified["stale_runtime"]
|
||||
runtime_mode = classified["runtime_mode"]
|
||||
permission = classified["permission"]
|
||||
other = classified["other"]
|
||||
|
||||
blocked: dict = {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"reasons": list(reasons),
|
||||
"mutation_performed": False,
|
||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||
"gate_reason_classes": {
|
||||
"stale_runtime": list(stale),
|
||||
"runtime_mode": list(runtime_mode),
|
||||
"permission": list(permission),
|
||||
"other": list(other),
|
||||
},
|
||||
}
|
||||
|
||||
if stale:
|
||||
parity = _current_master_parity()
|
||||
blocked["blocker_kind"] = "runtime_reconnect_required"
|
||||
blocked["restart_required"] = True
|
||||
blocked["stop_required"] = True
|
||||
blocked["startup_head"] = parity.get("startup_head")
|
||||
blocked["current_head"] = parity.get("current_head")
|
||||
blocked["daemon_start_head"] = (
|
||||
parity.get("daemon_start_head") or parity.get("startup_head")
|
||||
)
|
||||
blocked["local_head"] = (
|
||||
parity.get("local_head") or parity.get("current_head")
|
||||
)
|
||||
blocked["live_remote_head"] = parity.get("live_remote_head")
|
||||
blocked["live_stale"] = bool(parity.get("live_stale"))
|
||||
blocked["live_known"] = bool(parity.get("live_known"))
|
||||
blocked["exact_safe_next_action"] = _stale_runtime_reconnect_action()
|
||||
if permission or other:
|
||||
blocked["permission_block_reasons"] = list(permission) + list(other)
|
||||
blocked["stale_runtime_reasons"] = list(stale)
|
||||
# Never attach permission_report for a staleness refusal.
|
||||
blocked.update(extra_fields)
|
||||
return blocked
|
||||
|
||||
if runtime_mode:
|
||||
blocked["blocker_kind"] = "runtime_mode_blocked"
|
||||
blocked["restart_required"] = False
|
||||
blocked["stop_required"] = True
|
||||
blocked["exact_safe_next_action"] = (
|
||||
"Real workflow mutations run only on the promoted stable control "
|
||||
"runtime. Promote/reload the stable runtime; do not call "
|
||||
"gitea_activate_profile or switch MCP role sessions to clear a "
|
||||
"runtime-mode block."
|
||||
)
|
||||
if permission or other:
|
||||
blocked["permission_block_reasons"] = list(permission) + list(other)
|
||||
blocked["runtime_mode_reasons"] = list(runtime_mode)
|
||||
blocked.update(extra_fields)
|
||||
return blocked
|
||||
|
||||
# Pure permission (or unclassified-as-permission) denial.
|
||||
blocked["blocker_kind"] = "permission_denied"
|
||||
blocked["permission_report"] = _permission_block_report(required_operation)
|
||||
blocked.update(extra_fields)
|
||||
return blocked
|
||||
|
||||
|
||||
def _permission_report_for_gate_reasons(
|
||||
required_operation: str,
|
||||
reasons: list[str] | None,
|
||||
) -> dict | None:
|
||||
"""Attach ``permission_report`` only for true permission denials (#897).
|
||||
|
||||
Call sites that historically always attached a permission report after
|
||||
``_profile_operation_gate`` should use this so stale/runtime refusals
|
||||
do not emit a fabricated missing-permission payload.
|
||||
"""
|
||||
if not reasons:
|
||||
return None
|
||||
classified = _classify_operation_gate_reasons(reasons)
|
||||
if classified["stale_runtime"] or classified["runtime_mode"]:
|
||||
return None
|
||||
if not (classified["permission"] or classified["other"]):
|
||||
return None
|
||||
return _permission_block_report(required_operation)
|
||||
|
||||
|
||||
def _role_for_operation(op: str) -> str | None:
|
||||
# Normalize op first
|
||||
try:
|
||||
@@ -14079,7 +14307,7 @@ def _master_parity_block(op: str) -> list[str]:
|
||||
|
||||
|
||||
def _profile_operation_gate(op: str) -> list[str]:
|
||||
"""Profile permission check for a single gated operation (#126, #216, #420).
|
||||
"""Profile permission check for a single gated operation (#126, #216, #420, #897).
|
||||
|
||||
Issue discussion comments are gated separately from the gitea.pr.*
|
||||
review/merge family: listing requires ``gitea.read``, creating requires
|
||||
@@ -14092,21 +14320,26 @@ def _profile_operation_gate(op: str) -> list[str]:
|
||||
capability gate that has since been merged, and when the runtime itself is
|
||||
not the promoted stable control runtime (#615) -- a dev/test or unknown
|
||||
runtime holds production credentials but has not been promoted.
|
||||
|
||||
#897: collect *all* independent refusal classes (stale, runtime-mode,
|
||||
permission) rather than short-circuiting after the first. Callers that
|
||||
only need a boolean still treat any non-empty list as blocked; typed
|
||||
consumers (``_build_operation_gate_refusal``) can separate causes.
|
||||
"""
|
||||
stale_reasons = _master_parity_block(op)
|
||||
if stale_reasons:
|
||||
return stale_reasons
|
||||
runtime_reasons = _runtime_mode_block(op)
|
||||
if runtime_reasons:
|
||||
return runtime_reasons
|
||||
reasons: list[str] = []
|
||||
reasons.extend(_master_parity_block(op))
|
||||
reasons.extend(_runtime_mode_block(op))
|
||||
try:
|
||||
profile = get_profile()
|
||||
except Exception as exc:
|
||||
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
|
||||
reasons.append(
|
||||
f"profile could not be resolved (fail closed): {_redact(str(exc))}"
|
||||
)
|
||||
return reasons
|
||||
op_ok, op_reason = gitea_config.check_operation(
|
||||
op, profile["allowed_operations"], profile["forbidden_operations"])
|
||||
if op_ok:
|
||||
return []
|
||||
return reasons
|
||||
|
||||
if _try_auto_switch_for_operation(op):
|
||||
try:
|
||||
@@ -14114,17 +14347,26 @@ def _profile_operation_gate(op: str) -> list[str]:
|
||||
op_ok, op_reason = gitea_config.check_operation(
|
||||
op, profile["allowed_operations"], profile["forbidden_operations"])
|
||||
if op_ok:
|
||||
return []
|
||||
return reasons
|
||||
except Exception as exc:
|
||||
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
|
||||
reasons.append(
|
||||
f"profile could not be resolved (fail closed): {_redact(str(exc))}"
|
||||
)
|
||||
return reasons
|
||||
|
||||
if op_reason == "no-allowed-operations":
|
||||
return ["profile has no configured allowed operations (fail closed)"]
|
||||
if op_reason == "forbidden":
|
||||
return [f"profile forbids '{op}'"]
|
||||
if op_reason == "invalid-forbidden-entry":
|
||||
return ["profile has an unrecognized forbidden operation entry (fail closed)"]
|
||||
return [f"profile is not allowed to {op}"]
|
||||
reasons.append(
|
||||
"profile has no configured allowed operations (fail closed)"
|
||||
)
|
||||
elif op_reason == "forbidden":
|
||||
reasons.append(f"profile forbids '{op}'")
|
||||
elif op_reason == "invalid-forbidden-entry":
|
||||
reasons.append(
|
||||
"profile has an unrecognized forbidden operation entry (fail closed)"
|
||||
)
|
||||
else:
|
||||
reasons.append(f"profile is not allowed to {op}")
|
||||
return reasons
|
||||
|
||||
|
||||
def _mutation_config_authority_block(required_operation: str) -> dict | None:
|
||||
@@ -14344,10 +14586,14 @@ def _session_context_mutation_block(
|
||||
|
||||
|
||||
def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
|
||||
"""Structured permission denial for gated tools (#69, #142).
|
||||
"""Structured operation-gate denial for gated tools (#69, #142, #897).
|
||||
|
||||
Returns a block dict when the active profile forbids *required_operation*,
|
||||
or ``None`` when the gate passes. Never performs network I/O.
|
||||
the daemon is stale, or the runtime mode is not mutation-safe — or
|
||||
``None`` when the gate passes. Never performs network I/O.
|
||||
|
||||
#897: stale-runtime and runtime-mode refusals are typed
|
||||
(``blocker_kind``) and never carry a ``permission_report``.
|
||||
"""
|
||||
req_role = "reviewer" if any(required_operation.startswith(p) for p in (
|
||||
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
|
||||
@@ -14357,15 +14603,9 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
|
||||
|
||||
reasons = _profile_operation_gate(required_operation)
|
||||
if reasons:
|
||||
blocked = {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"reasons": reasons,
|
||||
"permission_report": _permission_block_report(required_operation),
|
||||
"session_context_audit": session_ctx.mutation_context_audit_fields(),
|
||||
}
|
||||
blocked.update(extra_fields)
|
||||
return blocked
|
||||
return _build_operation_gate_refusal(
|
||||
required_operation, reasons, **extra_fields
|
||||
)
|
||||
|
||||
auth_block = _mutation_config_authority_block(required_operation)
|
||||
if auth_block is not None:
|
||||
@@ -14494,20 +14734,14 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
"""Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
|
||||
read_block = _profile_operation_gate("gitea.read")
|
||||
if read_block:
|
||||
return {
|
||||
"success": False,
|
||||
"acquired": False,
|
||||
"reasons": read_block,
|
||||
"permission_report": _permission_block_report("gitea.read"),
|
||||
}
|
||||
return _build_operation_gate_refusal(
|
||||
"gitea.read", read_block, acquired=False
|
||||
)
|
||||
comment_block = _profile_operation_gate("gitea.pr.comment")
|
||||
if comment_block:
|
||||
return {
|
||||
"success": False,
|
||||
"acquired": False,
|
||||
"reasons": comment_block,
|
||||
"permission_report": _permission_block_report("gitea.pr.comment"),
|
||||
}
|
||||
return _build_operation_gate_refusal(
|
||||
"gitea.pr.comment", comment_block, acquired=False
|
||||
)
|
||||
|
||||
# task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
|
||||
# anti-stomp for the declared lease-acquire mutation inventory entry.
|
||||
@@ -14623,20 +14857,14 @@ def gitea_acquire_merger_pr_lease(
|
||||
"""
|
||||
read_block = _profile_operation_gate("gitea.read")
|
||||
if read_block:
|
||||
return {
|
||||
"success": False,
|
||||
"acquired": False,
|
||||
"reasons": read_block,
|
||||
"permission_report": _permission_block_report("gitea.read"),
|
||||
}
|
||||
return _build_operation_gate_refusal(
|
||||
"gitea.read", read_block, acquired=False
|
||||
)
|
||||
comment_block = _profile_operation_gate("gitea.pr.comment")
|
||||
if comment_block:
|
||||
return {
|
||||
"success": False,
|
||||
"acquired": False,
|
||||
"reasons": comment_block,
|
||||
"permission_report": _permission_block_report("gitea.pr.comment"),
|
||||
}
|
||||
return _build_operation_gate_refusal(
|
||||
"gitea.pr.comment", comment_block, acquired=False
|
||||
)
|
||||
merge_block = _profile_operation_gate("gitea.pr.merge")
|
||||
if merge_block:
|
||||
return {
|
||||
@@ -19375,14 +19603,12 @@ def gitea_update_pr_branch_by_merge(
|
||||
# Permission: author branch push / PR mutation surface.
|
||||
push_block = _profile_operation_gate("gitea.branch.push")
|
||||
if push_block:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"mutation_allowed": False,
|
||||
"reasons": push_block,
|
||||
"permission_report": _permission_block_report("gitea.branch.push"),
|
||||
"role_kind": role,
|
||||
}
|
||||
return _build_operation_gate_refusal(
|
||||
"gitea.branch.push",
|
||||
push_block,
|
||||
mutation_allowed=False,
|
||||
role_kind=role,
|
||||
)
|
||||
|
||||
if role != "author":
|
||||
pre = pr_sync_status.assess_update_pr_branch_preflight(
|
||||
|
||||
@@ -0,0 +1,453 @@
|
||||
"""#897: stale-runtime / runtime-mode refusals must not look like permission denials.
|
||||
|
||||
Acceptance criteria (issue #897):
|
||||
|
||||
* Stale-runtime and runtime-mode refusals are typed distinctly from
|
||||
profile-permission refusals (distinct ``blocker_kind``).
|
||||
* A refusal caused by staleness or runtime mode never emits a
|
||||
``permission_report`` and never names a permission the active profile holds.
|
||||
* ``_permission_block_report`` verifies the active profile actually lacks the
|
||||
operation before reporting it missing.
|
||||
* A stale-runtime refusal reports reconnect-only recovery and never recommends
|
||||
``gitea_activate_profile`` or an MCP session switch.
|
||||
* The blocker payload states the observed heads (parity fields).
|
||||
* Matrix across author / reviewer / merger / reconciler profiles.
|
||||
* Regression: ``gitea_create_issue`` on a stale daemon under ``prgs-author``
|
||||
never returns ``missing_permission: gitea.issue.create``.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_config # noqa: E402
|
||||
import gitea_mcp_server as mcp_server # noqa: E402
|
||||
|
||||
SHA_START = "7af40fb5ff7debd5e9165fe97d9c7c279358e175"
|
||||
SHA_LIVE = "2f4dec832327513118f2fe92b74da25d124a01cb"
|
||||
|
||||
ROLE_MATRIX = (
|
||||
(
|
||||
"prgs-author",
|
||||
"author",
|
||||
"gitea.issue.create",
|
||||
[
|
||||
"gitea.read",
|
||||
"gitea.issue.create",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.comment",
|
||||
"gitea.repo.commit",
|
||||
],
|
||||
["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes"],
|
||||
"gitea.pr.merge", # forbidden op for pure-permission case
|
||||
),
|
||||
(
|
||||
"prgs-reviewer",
|
||||
"reviewer",
|
||||
"gitea.pr.review",
|
||||
[
|
||||
"gitea.read",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
],
|
||||
["gitea.branch.push", "gitea.pr.create"],
|
||||
"gitea.branch.push",
|
||||
),
|
||||
(
|
||||
"prgs-merger",
|
||||
"merger",
|
||||
"gitea.pr.merge",
|
||||
[
|
||||
"gitea.read",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
],
|
||||
["gitea.pr.approve", "gitea.branch.push", "gitea.pr.create"],
|
||||
"gitea.branch.push",
|
||||
),
|
||||
(
|
||||
"prgs-reconciler",
|
||||
"reconciler",
|
||||
"gitea.branch.delete",
|
||||
[
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
"gitea.pr.close",
|
||||
"gitea.issue.close",
|
||||
],
|
||||
["gitea.pr.approve", "gitea.pr.merge"],
|
||||
"gitea.pr.merge",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _profile(name: str, role: str, allowed: list[str], forbidden: list[str]) -> dict:
|
||||
return {
|
||||
"profile_name": name,
|
||||
"role": role,
|
||||
"role_kind": role,
|
||||
"allowed_operations": list(allowed),
|
||||
"forbidden_operations": list(forbidden),
|
||||
"identity": "test-user",
|
||||
}
|
||||
|
||||
|
||||
def _config(profiles: dict) -> dict:
|
||||
return {
|
||||
"version": 2,
|
||||
"profiles": {
|
||||
name: {
|
||||
"role": p["role"],
|
||||
"allowed_operations": p["allowed_operations"],
|
||||
"forbidden_operations": p["forbidden_operations"],
|
||||
}
|
||||
for name, p in profiles.items()
|
||||
},
|
||||
"rules": {"allow_runtime_switching": True},
|
||||
}
|
||||
|
||||
|
||||
class Issue897Helpers(unittest.TestCase):
|
||||
def test_classify_stale_reason_strings(self):
|
||||
stale = (
|
||||
f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
|
||||
f"at {SHA_START[:12]}; the daemon is stale relative to live master "
|
||||
"-- restart/reconnect before mutating"
|
||||
)
|
||||
classified = mcp_server._classify_operation_gate_reasons([stale])
|
||||
self.assertEqual(classified["stale_runtime"], [stale])
|
||||
self.assertEqual(classified["permission"], [])
|
||||
self.assertEqual(classified["runtime_mode"], [])
|
||||
|
||||
def test_classify_permission_reason(self):
|
||||
reason = "profile is not allowed to gitea.pr.merge"
|
||||
classified = mcp_server._classify_operation_gate_reasons([reason])
|
||||
self.assertEqual(classified["permission"], [reason])
|
||||
self.assertEqual(classified["stale_runtime"], [])
|
||||
|
||||
def test_classify_runtime_mode_reason(self):
|
||||
reason = (
|
||||
"runtime mode is 'dev-test' and the mutation targets the "
|
||||
"production repository; dev/test runtimes must not mutate real "
|
||||
"issues or PRs (ADR: stable control runtime vs dev runtime)"
|
||||
)
|
||||
classified = mcp_server._classify_operation_gate_reasons([reason])
|
||||
self.assertEqual(classified["runtime_mode"], [reason])
|
||||
self.assertEqual(classified["stale_runtime"], [])
|
||||
|
||||
|
||||
class Issue897PermissionBlockReport(unittest.TestCase):
|
||||
def test_holds_op_is_diagnostic_defect_not_missing_permission(self):
|
||||
profile = _profile(
|
||||
"prgs-author",
|
||||
"author",
|
||||
["gitea.read", "gitea.issue.create", "gitea.issue.comment"],
|
||||
[],
|
||||
)
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server.gitea_config, "load_config", return_value=_config({"prgs-author": profile})
|
||||
), patch.object(
|
||||
mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
|
||||
):
|
||||
report = mcp_server._permission_block_report("gitea.issue.create")
|
||||
self.assertTrue(report.get("diagnostic_defect"), report)
|
||||
self.assertIsNone(report.get("missing_permission"), report)
|
||||
action = (report.get("exact_safe_next_action") or "").lower()
|
||||
# Must not *recommend* profile switching; mentioning the forbidden
|
||||
# action in a "do not call" instruction is fine.
|
||||
self.assertNotIn("call gitea_activate_profile with", action)
|
||||
self.assertNotIn("switch to the author mcp session", action)
|
||||
self.assertNotIn("switch to the reviewer mcp session", action)
|
||||
self.assertIn("diagnostic defect", action)
|
||||
|
||||
def test_true_missing_permission_still_reports(self):
|
||||
profile = _profile(
|
||||
"prgs-author",
|
||||
"author",
|
||||
["gitea.read", "gitea.issue.create"],
|
||||
["gitea.pr.merge"],
|
||||
)
|
||||
reviewer = _profile(
|
||||
"prgs-reviewer",
|
||||
"reviewer",
|
||||
["gitea.read", "gitea.pr.merge", "gitea.pr.approve"],
|
||||
[],
|
||||
)
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server.gitea_config,
|
||||
"load_config",
|
||||
return_value=_config({"prgs-author": profile, "prgs-reviewer": reviewer}),
|
||||
), patch.object(
|
||||
mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
|
||||
):
|
||||
report = mcp_server._permission_block_report("gitea.pr.merge")
|
||||
self.assertFalse(report.get("diagnostic_defect"), report)
|
||||
self.assertEqual(report.get("missing_permission"), "gitea.pr.merge")
|
||||
self.assertIn("prgs-reviewer", report.get("matching_configured_profiles") or [])
|
||||
|
||||
|
||||
class Issue897GateRefusalMatrix(unittest.TestCase):
|
||||
def _stale_parity(self) -> dict:
|
||||
return {
|
||||
"in_parity": True,
|
||||
"stale": False,
|
||||
"restart_required": True,
|
||||
"determinable": True,
|
||||
"startup_head": SHA_START,
|
||||
"current_head": SHA_START,
|
||||
"daemon_start_head": SHA_START,
|
||||
"local_head": SHA_START,
|
||||
"live_remote_head": SHA_LIVE,
|
||||
"live_known": True,
|
||||
"live_stale": True,
|
||||
"mutation_safe": False,
|
||||
"reasons": [
|
||||
f"live remote master is {SHA_LIVE[:12]} but the MCP server "
|
||||
f"started at {SHA_START[:12]}; the daemon is stale relative "
|
||||
"to live master -- restart/reconnect before mutating"
|
||||
],
|
||||
}
|
||||
|
||||
def test_stale_plus_permitted_op_all_roles(self):
|
||||
for name, role, permitted_op, allowed, forbidden, _forbidden_op in ROLE_MATRIX:
|
||||
with self.subTest(profile=name, op=permitted_op):
|
||||
profile = _profile(name, role, allowed, forbidden)
|
||||
parity = self._stale_parity()
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server, "_current_master_parity", return_value=parity
|
||||
), patch.object(
|
||||
mcp_server, "_master_parity_block", return_value=list(parity["reasons"])
|
||||
), patch.object(
|
||||
mcp_server, "_runtime_mode_block", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
), patch.object(
|
||||
mcp_server.session_ctx,
|
||||
"mutation_context_audit_fields",
|
||||
return_value={"session_profile": name},
|
||||
):
|
||||
blocked = mcp_server._profile_permission_block(permitted_op)
|
||||
self.assertIsNotNone(blocked, name)
|
||||
assert blocked is not None
|
||||
self.assertEqual(
|
||||
blocked.get("blocker_kind"),
|
||||
"runtime_reconnect_required",
|
||||
blocked,
|
||||
)
|
||||
self.assertNotIn("permission_report", blocked, blocked)
|
||||
self.assertTrue(blocked.get("restart_required"), blocked)
|
||||
self.assertEqual(blocked.get("startup_head"), SHA_START, blocked)
|
||||
self.assertEqual(blocked.get("live_remote_head"), SHA_LIVE, blocked)
|
||||
action = (blocked.get("exact_safe_next_action") or "").lower()
|
||||
self.assertIn("reconnect", action)
|
||||
self.assertNotIn("call gitea_activate_profile with", action)
|
||||
self.assertNotIn("switch to the author mcp session", action)
|
||||
self.assertNotIn("switch to the reviewer mcp session", action)
|
||||
|
||||
def test_fresh_plus_forbidden_op_all_roles(self):
|
||||
for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
|
||||
with self.subTest(profile=name, op=forbidden_op):
|
||||
profile = _profile(name, role, allowed, forbidden)
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server, "_master_parity_block", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "_runtime_mode_block", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
), patch.object(
|
||||
mcp_server.session_ctx,
|
||||
"mutation_context_audit_fields",
|
||||
return_value={"session_profile": name},
|
||||
), patch.object(
|
||||
mcp_server.gitea_config,
|
||||
"load_config",
|
||||
return_value=_config({name: profile}),
|
||||
), patch.object(
|
||||
mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=False
|
||||
):
|
||||
blocked = mcp_server._profile_permission_block(forbidden_op)
|
||||
self.assertIsNotNone(blocked, name)
|
||||
assert blocked is not None
|
||||
self.assertEqual(blocked.get("blocker_kind"), "permission_denied", blocked)
|
||||
self.assertIn("permission_report", blocked, blocked)
|
||||
report = blocked["permission_report"]
|
||||
self.assertEqual(report.get("missing_permission"), forbidden_op, report)
|
||||
self.assertFalse(report.get("diagnostic_defect"), report)
|
||||
# No runtime reconnect fields for pure permission denial
|
||||
self.assertNotEqual(
|
||||
blocked.get("blocker_kind"), "runtime_reconnect_required"
|
||||
)
|
||||
|
||||
def test_stale_plus_forbidden_op_both_causes_separated(self):
|
||||
for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
|
||||
with self.subTest(profile=name, op=forbidden_op):
|
||||
profile = _profile(name, role, allowed, forbidden)
|
||||
parity = self._stale_parity()
|
||||
stale_reason = parity["reasons"][0]
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server, "_current_master_parity", return_value=parity
|
||||
), patch.object(
|
||||
mcp_server, "_master_parity_block", return_value=[stale_reason]
|
||||
), patch.object(
|
||||
mcp_server, "_runtime_mode_block", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
), patch.object(
|
||||
mcp_server.session_ctx,
|
||||
"mutation_context_audit_fields",
|
||||
return_value={"session_profile": name},
|
||||
):
|
||||
# Gate collects both classes; force permission reason too.
|
||||
with patch.object(
|
||||
mcp_server,
|
||||
"_profile_operation_gate",
|
||||
return_value=[
|
||||
stale_reason,
|
||||
f"profile is not allowed to {forbidden_op}",
|
||||
],
|
||||
):
|
||||
blocked = mcp_server._profile_permission_block(forbidden_op)
|
||||
self.assertIsNotNone(blocked)
|
||||
assert blocked is not None
|
||||
self.assertEqual(
|
||||
blocked.get("blocker_kind"), "runtime_reconnect_required", blocked
|
||||
)
|
||||
self.assertNotIn("permission_report", blocked, blocked)
|
||||
self.assertIn("permission_block_reasons", blocked, blocked)
|
||||
self.assertIn("stale_runtime_reasons", blocked, blocked)
|
||||
classes = blocked.get("gate_reason_classes") or {}
|
||||
self.assertTrue(classes.get("stale_runtime"), classes)
|
||||
self.assertTrue(classes.get("permission"), classes)
|
||||
|
||||
def test_runtime_mode_block_no_permission_report(self):
|
||||
profile = _profile(
|
||||
"prgs-author",
|
||||
"author",
|
||||
["gitea.read", "gitea.issue.create"],
|
||||
[],
|
||||
)
|
||||
runtime_reason = (
|
||||
"runtime mode is 'dev-test' and the mutation targets the "
|
||||
"production repository; dev/test runtimes must not mutate real "
|
||||
"issues or PRs (ADR: stable control runtime vs dev runtime)"
|
||||
)
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server, "_master_parity_block", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "_runtime_mode_block", return_value=[runtime_reason]
|
||||
), patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
), patch.object(
|
||||
mcp_server.session_ctx,
|
||||
"mutation_context_audit_fields",
|
||||
return_value={"session_profile": "prgs-author"},
|
||||
):
|
||||
blocked = mcp_server._profile_permission_block("gitea.issue.create")
|
||||
self.assertIsNotNone(blocked)
|
||||
assert blocked is not None
|
||||
self.assertEqual(blocked.get("blocker_kind"), "runtime_mode_blocked", blocked)
|
||||
self.assertNotIn("permission_report", blocked, blocked)
|
||||
action = (blocked.get("exact_safe_next_action") or "").lower()
|
||||
self.assertNotIn("call gitea_activate_profile with", action)
|
||||
self.assertIn("stable control runtime", action)
|
||||
|
||||
|
||||
class Issue897CreateIssueRegression(unittest.TestCase):
|
||||
def test_create_issue_stale_daemon_never_missing_issue_create(self):
|
||||
"""Regression AC: stale prgs-author create_issue must not claim missing create."""
|
||||
profile = _profile(
|
||||
"prgs-author",
|
||||
"author",
|
||||
[
|
||||
"gitea.read",
|
||||
"gitea.issue.create",
|
||||
"gitea.issue.comment",
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.comment",
|
||||
"gitea.repo.commit",
|
||||
],
|
||||
[],
|
||||
)
|
||||
stale_reason = (
|
||||
f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
|
||||
f"at {SHA_START[:12]}; the daemon is stale relative to live master "
|
||||
"-- restart/reconnect before mutating"
|
||||
)
|
||||
parity = {
|
||||
"in_parity": True,
|
||||
"stale": False,
|
||||
"restart_required": True,
|
||||
"determinable": True,
|
||||
"startup_head": SHA_START,
|
||||
"current_head": SHA_START,
|
||||
"daemon_start_head": SHA_START,
|
||||
"local_head": SHA_START,
|
||||
"live_remote_head": SHA_LIVE,
|
||||
"live_known": True,
|
||||
"live_stale": True,
|
||||
"mutation_safe": False,
|
||||
"reasons": [stale_reason],
|
||||
}
|
||||
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
|
||||
mcp_server, "_current_master_parity", return_value=parity
|
||||
), patch.object(
|
||||
mcp_server, "_master_parity_block", return_value=[stale_reason]
|
||||
), patch.object(
|
||||
mcp_server, "_runtime_mode_block", return_value=[]
|
||||
), patch.object(
|
||||
mcp_server, "_ensure_matching_profile", return_value=None
|
||||
), patch.object(
|
||||
mcp_server.session_ctx,
|
||||
"mutation_context_audit_fields",
|
||||
return_value={"session_profile": "prgs-author"},
|
||||
), patch.object(
|
||||
mcp_server, "_mutation_config_authority_block", return_value=None
|
||||
), patch.object(
|
||||
mcp_server, "_session_context_mutation_block", return_value=None
|
||||
):
|
||||
blocked = mcp_server._profile_permission_block(
|
||||
"gitea.issue.create", remote="prgs"
|
||||
)
|
||||
|
||||
self.assertIsNotNone(blocked)
|
||||
assert blocked is not None
|
||||
self.assertEqual(blocked.get("blocker_kind"), "runtime_reconnect_required")
|
||||
self.assertNotIn("permission_report", blocked)
|
||||
# Even if a caller still built a raw report, holds-check must not claim missing.
|
||||
with patch.object(mcp_server, "get_profile", return_value=profile):
|
||||
raw = mcp_server._permission_block_report("gitea.issue.create")
|
||||
self.assertIsNone(raw.get("missing_permission"), raw)
|
||||
self.assertNotEqual(raw.get("missing_permission"), "gitea.issue.create")
|
||||
|
||||
def test_permission_report_for_gate_reasons_skips_stale(self):
|
||||
stale = (
|
||||
f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
|
||||
f"at {SHA_START[:12]}; the daemon is stale relative to live master "
|
||||
"-- restart/reconnect before mutating"
|
||||
)
|
||||
self.assertIsNone(
|
||||
mcp_server._permission_report_for_gate_reasons(
|
||||
"gitea.issue.comment", [stale]
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user