fix(mcp): implement operation-scoped role selection and auto profile switching
This commit is contained in:
+112
-3
@@ -2758,6 +2758,76 @@ def _permission_block_report(required_operation: str,
|
||||
return report
|
||||
|
||||
|
||||
def _role_for_operation(op: str) -> str | None:
|
||||
# Normalize op first
|
||||
try:
|
||||
op_n = gitea_config.normalize_operation(op)
|
||||
except Exception:
|
||||
op_n = op
|
||||
if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"):
|
||||
return "reviewer"
|
||||
if op_n in (
|
||||
"gitea.issue.create",
|
||||
"gitea.issue.comment",
|
||||
"gitea.issue.close",
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.pr.create",
|
||||
"gitea.pr.comment",
|
||||
"gitea.pr.close",
|
||||
"gitea.branch.delete",
|
||||
"gitea.repo.commit"
|
||||
):
|
||||
return "author"
|
||||
return None
|
||||
|
||||
|
||||
def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
|
||||
"""Try to find a profile in config that allows op and has valid credentials.
|
||||
|
||||
If found, switch to it, clear identity cache, and return True.
|
||||
Otherwise return False.
|
||||
"""
|
||||
role = _role_for_operation(op)
|
||||
if not role:
|
||||
return False
|
||||
if not gitea_config.is_runtime_switching_enabled():
|
||||
return False
|
||||
config = gitea_config.load_config()
|
||||
if not config or "profiles" not in config:
|
||||
return False
|
||||
for p_name, p_data in config["profiles"].items():
|
||||
# Role classification matching
|
||||
p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
|
||||
if p_role != role:
|
||||
continue
|
||||
p_allowed = p_data.get("allowed_operations") or []
|
||||
p_forbidden = p_data.get("forbidden_operations") or []
|
||||
p_allowed_n = []
|
||||
for op_val in p_allowed:
|
||||
try:
|
||||
p_allowed_n.append(gitea_config.normalize_operation(op_val))
|
||||
except Exception:
|
||||
pass
|
||||
p_forbidden_n = []
|
||||
for op_val in p_forbidden:
|
||||
try:
|
||||
p_forbidden_n.append(gitea_config.normalize_operation(op_val))
|
||||
except Exception:
|
||||
pass
|
||||
ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
|
||||
if ok:
|
||||
try:
|
||||
tok = gitea_config.resolve_token(p_data)
|
||||
if tok:
|
||||
gitea_config._active_profile_override = p_name
|
||||
_IDENTITY_CACHE.clear()
|
||||
return True
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def _profile_operation_gate(op: str) -> list[str]:
|
||||
"""Profile permission check for a single gated operation (#126, #216).
|
||||
|
||||
@@ -2775,6 +2845,17 @@ def _profile_operation_gate(op: str) -> list[str]:
|
||||
op, profile["allowed_operations"], profile["forbidden_operations"])
|
||||
if op_ok:
|
||||
return []
|
||||
|
||||
if _try_auto_switch_for_operation(op):
|
||||
try:
|
||||
profile = get_profile()
|
||||
op_ok, op_reason = gitea_config.check_operation(
|
||||
op, profile["allowed_operations"], profile["forbidden_operations"])
|
||||
if op_ok:
|
||||
return []
|
||||
except Exception as exc:
|
||||
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
|
||||
|
||||
if op_reason == "no-allowed-operations":
|
||||
return ["profile has no configured allowed operations (fail closed)"]
|
||||
if op_reason == "forbidden":
|
||||
@@ -4426,6 +4507,12 @@ def gitea_resolve_task_capability(
|
||||
required_permission, active_allowed, active_forbidden
|
||||
)
|
||||
|
||||
switching = gitea_config.is_runtime_switching_enabled()
|
||||
available_in_session = allowed_in_current_session
|
||||
configured = False
|
||||
restart_required = False
|
||||
reason_msg = None
|
||||
|
||||
# Find matching configured profiles
|
||||
matching_profiles = []
|
||||
if config and "profiles" in config:
|
||||
@@ -4448,7 +4535,24 @@ def gitea_resolve_task_capability(
|
||||
if ok:
|
||||
matching_profiles.append(p_name)
|
||||
|
||||
switching = gitea_config.is_runtime_switching_enabled()
|
||||
if not allowed_in_current_session and switching:
|
||||
# Try to auto switch
|
||||
if _try_auto_switch_for_operation(required_permission, h):
|
||||
# Successfully switched!
|
||||
profile = get_profile()
|
||||
username = _authenticated_username(h) if h else None
|
||||
active_allowed = profile.get("allowed_operations") or []
|
||||
active_forbidden = profile.get("forbidden_operations") or []
|
||||
allowed_in_current_session = True
|
||||
available_in_session = True
|
||||
else:
|
||||
# Check if any profile in config allows it (meaning it is configured but unattached)
|
||||
if matching_profiles:
|
||||
configured = True
|
||||
restart_required = True
|
||||
available_in_session = False
|
||||
reason_msg = f"{required_role.capitalize()} profile exists but MCP server was added after session startup and is not attached"
|
||||
|
||||
different_namespace_required = False
|
||||
next_safe_action = "None; ready for operations."
|
||||
|
||||
@@ -4526,14 +4630,19 @@ def gitea_resolve_task_capability(
|
||||
"active_identity": username,
|
||||
"active_profile_allowed_operations": active_allowed,
|
||||
"allowed_in_current_session": allowed_in_current_session,
|
||||
"stop_required": stop_required,
|
||||
"available_in_session": available_in_session,
|
||||
"configured": configured,
|
||||
"restart_required": restart_required,
|
||||
"stop_required": stop_required or restart_required,
|
||||
"task_role_guidance": task_role_guidance,
|
||||
"matching_configured_profile": matching_profiles,
|
||||
"runtime_switching_supported": switching,
|
||||
"different_mcp_namespace_required": different_namespace_required,
|
||||
"exact_safe_next_action": next_safe_action,
|
||||
}
|
||||
if stop_required:
|
||||
if reason_msg:
|
||||
result["reason"] = reason_msg
|
||||
if stop_required or restart_required:
|
||||
terminal = capability_stop_terminal.enter_from_capability_result(result)
|
||||
if terminal:
|
||||
result["terminal_mode"] = True
|
||||
|
||||
Reference in New Issue
Block a user