diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 20bd535..a4705a0 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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