diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index c433a1d..24a01b7 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1118,6 +1118,34 @@ import workflow_scope_guard # noqa: E402 # #683 production scope / force-on gu import stable_branch_push_guard # noqa: E402 import remote_repo_guard # noqa: E402 import issue_claim_heartbeat # noqa: E402 +import session_context_binding as session_ctx # noqa: E402 # #714 immutable session context + + +def _seed_session_context( + *, + profile: dict, + remote: str | None, + host: str | None, + identity: str | None, + repository: str | None = None, + org: str | None = None, + source: str = "seed", +) -> dict: + """Seed/rebind immutable session context with env/override awareness (#714).""" + expected = (profile.get("username") or "").strip() or None + return session_ctx.seed_session_context_if_unbound( + profile_name=profile.get("profile_name") or "", + remote=remote, + host=host, + identity=identity, + repository=repository, + org=org, + role_kind=_profile_role_kind(profile), + expected_username=expected, + source=source, + active_profile_override=gitea_config._active_profile_override, + env_profile=(os.environ.get("GITEA_MCP_PROFILE") or "").strip() or None, + ) import issue_work_duplicate_gate # noqa: E402 import issue_workflow_labels # noqa: E402 import reviewer_pr_lease # noqa: E402 @@ -1846,61 +1874,44 @@ def _authenticated_username(host: str): return user -def _ensure_matching_profile(required_permission: str, required_role: str, remote: str | None, host: str | None = None) -> str | None: - """Check if the active profile is allowed to perform *required_permission*. - If not, automatically switch to the first matching usable configured profile. +def _ensure_matching_profile( + required_permission: str, + required_role: str, + remote: str | None, + host: str | None = None, +) -> str | None: + """Return the active profile name only when it is allowed for *permission*. + + #714: never silently switch profiles (including cross-host substitution). + Capability resolution and mutation gates evaluate only the active profile + for the requested remote. Explicit ``gitea_activate_profile`` is the sole + in-process switch path. """ + del required_role, host # kept for call-site compatibility try: profile = get_profile() except Exception: return None active_profile = profile.get("profile_name") + # Cross-host denial: mdcps profile cannot serve a prgs remote (and vice versa). + config = None + try: + config = gitea_config.load_config() + except Exception: + config = None + contexts = (config or {}).get("contexts") if config else None + remote_ok = session_ctx.profile_allowed_for_remote( + profile, remote, REMOTES, contexts=contexts + ) + if remote_ok.get("block"): + return None active_allowed = profile.get("allowed_operations") or [] active_forbidden = profile.get("forbidden_operations") or [] - allowed, _ = gitea_config.check_operation(required_permission, active_allowed, active_forbidden) + allowed, _ = gitea_config.check_operation( + required_permission, active_allowed, active_forbidden + ) if allowed: return active_profile - - # Try to find a matching usable profile in config - if gitea_config.is_runtime_switching_enabled(): - config = gitea_config.load_config() - if config and "profiles" in config: - for p_name, p_data in config["profiles"].items(): - p_allowed = p_data.get("allowed_operations") or [] - p_forbidden = p_data.get("forbidden_operations") or [] - p_allowed_n = [] - for op in p_allowed: - try: - p_allowed_n.append(gitea_config.normalize_operation(op)) - except Exception: - pass - p_forbidden_n = [] - for op in p_forbidden: - try: - p_forbidden_n.append(gitea_config.normalize_operation(op)) - except Exception: - pass - ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n) - if ok: - # Verify credentials/token are available - try: - tok = gitea_config.resolve_token(p_data) - if tok: - # Perform automatic switch - gitea_config._active_profile_override = p_name - h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) - if h: - _IDENTITY_CACHE.pop(h, None) - username = _authenticated_username(h) if h else None - # Update mutation authority - global _MUTATION_AUTHORITY - if _MUTATION_AUTHORITY is not None: - _MUTATION_AUTHORITY["current_profile"] = p_name - _MUTATION_AUTHORITY["current_identity"] = username - _MUTATION_AUTHORITY["role_pivot_authorized"] = True - return p_name - except Exception: - pass return None @@ -1922,6 +1933,12 @@ def _audit(action: str, *, host, remote, result, org=None, repo=None, if mutation_task: ns_ctx = role_namespace_gate.mutation_audit_context( mutation_task, profile, remote=remote, repository=repo) + # #714: always record the bound session context at mutation time. + session_audit = session_ctx.mutation_context_audit_fields() + if isinstance(request_metadata, dict): + request_metadata = {**request_metadata, **session_audit} + elif request_metadata is None: + request_metadata = dict(session_audit) event = gitea_audit.build_event( action=action, result=result, @@ -7499,48 +7516,13 @@ def _role_for_operation(op: str) -> str | 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. + """#714: automatic profile substitution is removed (fail closed). - If found, switch to it, clear identity cache, and return True. - Otherwise return False. + Always returns False. Callers must use explicit ``gitea_activate_profile`` + to change roles; capability and mutation gates evaluate only the active + profile. Parameters retained for call-site compatibility. """ - 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 + del op, host return False @@ -7607,6 +7589,102 @@ def _profile_operation_gate(op: str) -> list[str]: return [f"profile is not allowed to {op}"] +def _session_context_mutation_block( + *, + remote: str | None, + host: str | None = None, + org: str | None = None, + repo: str | None = None, + username: str | None = None, + **extra_fields, +) -> dict | None: + """#714: fail closed on session context / identity / cross-host drift.""" + try: + profile = get_profile() + except Exception as exc: + return { + "success": False, + "performed": False, + "reasons": [ + f"profile could not be resolved (fail closed): {_redact(str(exc))}" + ], + "blocker_kind": "session_context", + **extra_fields, + } + profile_name = profile.get("profile_name") + expected = (profile.get("username") or "").strip() or None + h = host or (REMOTES.get(remote or "", {}).get("host") if remote else None) + identity = username + if identity is None and h: + try: + identity = _authenticated_username(h) + except Exception: + identity = None + + config = None + try: + config = gitea_config.load_config() + except Exception: + config = None + contexts = (config or {}).get("contexts") if config else None + remote_gate = session_ctx.profile_allowed_for_remote( + profile, remote, REMOTES, contexts=contexts + ) + reasons: list[str] = list(remote_gate.get("reasons") or []) + + id_gate = session_ctx.assess_identity_match( + authenticated=identity, expected_username=expected + ) + reasons.extend(id_gate.get("reasons") or []) + + # Seed if unbound so subsequent tools share one context; then assess. + _seed_session_context( + profile=profile, + remote=remote, + host=h, + identity=identity, + repository=repo, + org=org, + source="mutation-gate-seed", + ) + ctx_gate = session_ctx.assess_session_context( + profile_name=profile_name, + remote=remote, + host=h, + identity=identity, + repository=repo, + org=org, + expected_username=expected, + require_bound=True, + ) + reasons.extend(ctx_gate.get("reasons") or []) + # De-dupe while preserving order + seen: set[str] = set() + uniq: list[str] = [] + for r in reasons: + if r not in seen: + seen.add(r) + uniq.append(r) + if not uniq: + return None + blocked = { + "success": False, + "performed": False, + "reasons": uniq, + "blocker_kind": "session_context", + "session_context_audit": session_ctx.mutation_context_audit_fields(), + "exact_next_action": ( + "BLOCKED + DIAGNOSE: session profile/remote/host/identity context " + "is inconsistent or unauthorized for this mutation. Re-pin the " + "correct profile with gitea_activate_profile, re-verify with " + "gitea_whoami for the intended remote, and do not use cross-host " + "fallback profiles." + ), + } + blocked.update(extra_fields) + return blocked + + def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None: """Structured permission denial for gated tools (#69, #142). @@ -7616,8 +7694,21 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict | 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" )) else "author" + # #714: evaluate active profile only — never auto-switch. _ensure_matching_profile(required_operation, req_role, extra_fields.get("remote")) + ctx_block = _session_context_mutation_block( + remote=extra_fields.get("remote"), + host=extra_fields.get("host"), + org=extra_fields.get("org"), + repo=extra_fields.get("repo"), + number=extra_fields.get("number"), + issue_number=extra_fields.get("issue_number"), + pr_number=extra_fields.get("pr_number"), + ) + if ctx_block is not None: + return ctx_block + reasons = _profile_operation_gate(required_operation) if not reasons: return None @@ -7626,6 +7717,7 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict | "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 @@ -7635,8 +7727,21 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None """Reviewer/author namespace alignment gate (#209).""" required_permission = task_capability_map.required_permission(mutation_task) required_role = task_capability_map.required_role(mutation_task) + # #714: evaluate active profile only — never auto-switch. _ensure_matching_profile(required_permission, required_role, extra_fields.get("remote")) + ctx_block = _session_context_mutation_block( + remote=extra_fields.get("remote"), + host=extra_fields.get("host"), + org=extra_fields.get("org"), + repo=extra_fields.get("repo"), + number=extra_fields.get("number"), + issue_number=extra_fields.get("issue_number"), + pr_number=extra_fields.get("pr_number"), + ) + if ctx_block is not None: + return ctx_block + try: profile = get_profile() except Exception as exc: @@ -9621,9 +9726,22 @@ def gitea_whoami( # name is the addressing surface; 'server' appears only under the # GITEA_MCP_REVEAL_ENDPOINTS admin opt-in. profile = get_profile() + identity = data.get("login") + expected_username = (profile.get("username") or "").strip() or None + # #714: seed immutable session context so later tools share one pin. + _seed_session_context( + profile=profile, + remote=remote, + host=h, + identity=identity, + source="gitea_whoami", + ) + id_match = session_ctx.assess_identity_match( + authenticated=identity, expected_username=expected_username + ) result = { "authenticated": True, - "username": data.get("login"), + "username": identity, "display_name": data.get("full_name") or None, "user_id": data.get("id"), "email": data.get("email") or None, @@ -9640,7 +9758,11 @@ def gitea_whoami( "execution_profile": profile.get("execution_profile"), "audit_label": profile.get("audit_label"), "auth_source_type": profile.get("auth_source_type"), + "expected_username": expected_username, }, + "session_context_audit": session_ctx.mutation_context_audit_fields(), + "identity_match": not id_match.get("block"), + "identity_match_reasons": id_match.get("reasons") or [], } if _reveal_endpoints(): result["server"] = gitea_url(h, "").rstrip("/") @@ -9766,14 +9888,32 @@ _RUNTIME_CAPABILITY_TASKS = ( def _matching_configured_profiles( config: dict | None, required_permission: str, + remote: str | None = None, ) -> list[str]: - """Profile names that allow *required_permission* (redacted metadata only).""" + """Profile names that allow *required_permission* (redacted metadata only). + + #714: when *remote* is provided, only profiles bound to that remote's host + are returned — a dadeschools request never lists prgs profiles. + """ if not config or "profiles" not in config: return [] + contexts = config.get("contexts") or {} + remote_ok_names: set[str] | None = None + if remote: + remote_ok_names = set( + session_ctx.filter_profiles_for_remote(config, remote, REMOTES) + ) matches: list[str] = [] for p_name, p_data in config["profiles"].items(): if not p_data.get("enabled", True): continue + if remote_ok_names is not None and p_name not in remote_ok_names: + continue + # Also require host/context match when remote given (defensive). + if remote and not session_ctx.profile_matches_remote( + p_data, remote, REMOTES, contexts=contexts + ): + continue p_allowed = p_data.get("allowed_operations") or [] p_forbidden = p_data.get("forbidden_operations") or [] p_allowed_n = [] @@ -9800,8 +9940,9 @@ def _build_runtime_task_capabilities( allowed: list[str], forbidden: list[str], config: dict | None, + remote: str | None = None, ) -> dict: - """Per-task capability summary for role-aware runtime context (#139).""" + """Per-task capability summary for role-aware runtime context (#139, #714).""" task_entries = [] flags: dict[str, bool] = {} flag_keys = { @@ -9825,7 +9966,7 @@ def _build_runtime_task_capabilities( "required_role_kind": task_capability_map.required_role(task), "allowed_in_current_session": allowed_here, "matching_configured_profiles": _matching_configured_profiles( - config, permission + config, permission, remote=remote ), } task_entries.append(entry) @@ -10277,8 +10418,9 @@ def gitea_get_runtime_context( "or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile." ) + # #714: filter matching profiles by requested remote when building capability summary session_capabilities = _build_runtime_task_capabilities( - allowed, forbidden, config + allowed, forbidden, config, remote=remote if remote in REMOTES else None ) preflight = assess_preflight_status(worktree_path) @@ -10289,6 +10431,16 @@ def gitea_get_runtime_context( f"Blocked: {'; '.join(preflight['preflight_block_reasons'])}" ) + expected_username = (profile.get("username") or "").strip() or None + h_rt = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) + _seed_session_context( + profile=profile, + remote=remote if remote in REMOTES else None, + host=h_rt, + identity=username, + source="gitea_get_runtime_context", + ) + result = { "active_profile": profile["profile_name"], "authenticated_username": username, @@ -10299,6 +10451,8 @@ def gitea_get_runtime_context( "forbidden_operations": forbidden, "runtime_switching_supported": switching, "profile_mode": profile_mode, + "session_context_audit": session_ctx.mutation_context_audit_fields(), + "auto_profile_substitution": False, "review_merge_allowed": review_merge_allowed, "review_merge_blocked_reasons": blocked_reasons, "suggested_fix": suggested_fix, @@ -10660,8 +10814,10 @@ def gitea_activate_profile( _IDENTITY_CACHE.pop(h, None) # 4. Resolve fresh identity - after_profile = get_profile()["profile_name"] + after_profile_data = get_profile() + after_profile = after_profile_data["profile_name"] after_identity = _authenticated_username(h) if h else None + expected_username = (after_profile_data.get("username") or "").strip() or None # 4.5 Record the authorized pivot in the in-process mutation authority # and keep the session profile lock in sync — this is the ONLY path that @@ -10676,15 +10832,32 @@ def gitea_activate_profile( "from_identity": before_identity, "to_identity": after_identity, } + _MUTATION_AUTHORITY["remote"] = remote if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile: os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile + # 4.6 #714: re-bind immutable session context after explicit activation. + session_ctx.bind_session_context( + profile_name=after_profile or profile_name, + remote=remote, + host=h, + identity=after_identity, + role_kind=_profile_role_kind(after_profile_data), + expected_username=expected_username, + source="gitea_activate_profile", + ) + # 5. Audit the switch if auditing is on _audit( "activate_profile", host=h, remote=remote, - result={"success": True, "before": before_profile, "after": after_profile}, + result={ + "success": True, + "before": before_profile, + "after": after_profile, + "session_context": session_ctx.mutation_context_audit_fields(), + }, username=after_identity, ) @@ -10695,6 +10868,8 @@ def gitea_activate_profile( "before_identity": before_identity, "after_profile": after_profile, "after_identity": after_identity, + "session_context_audit": session_ctx.mutation_context_audit_fields(), + "auto_profile_substitution": False, } @@ -11795,21 +11970,44 @@ def gitea_resolve_task_capability( record_preflight_check("capability", required_role, resolved_task=task) - # Try automatic dispatch switching - _ensure_matching_profile(required_permission, required_role, remote, host) - + # #714: never auto-switch profiles during capability resolution. profile = get_profile() config = gitea_config.load_config() + contexts = (config or {}).get("contexts") if config else None h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) username = _authenticated_username(h) if h else None + expected_username = (profile.get("username") or "").strip() or None + + # Seed / re-check immutable session context for consistent multi-tool reads. + _seed_session_context( + profile=profile, + remote=remote, + host=h, + identity=username, + source="resolve_task_capability", + ) + ctx_assess = session_ctx.assess_session_context( + profile_name=profile.get("profile_name"), + remote=remote, + host=h, + identity=username, + expected_username=expected_username, + require_bound=False, + ) + id_assess = session_ctx.assess_identity_match( + authenticated=username, expected_username=expected_username + ) + remote_assess = session_ctx.profile_allowed_for_remote( + profile, remote, REMOTES, contexts=contexts + ) # Load active permissions active_allowed = profile.get("allowed_operations") or [] active_forbidden = profile.get("forbidden_operations") or [] active_role_kind = _profile_role_kind(profile) - # Check if allowed in current session + # Check if allowed in current session (active profile only — #714) permission_allowed_in_current_session, _ = gitea_config.check_operation( required_permission, active_allowed, active_forbidden ) @@ -11824,8 +12022,15 @@ def gitea_resolve_task_capability( f"{required_role} task '{task}' even if nearby permissions are " "present (fail closed)." ) + cross_host_block = bool(remote_assess.get("block")) + identity_block = bool(id_assess.get("block")) + drift_block = bool(ctx_assess.get("block")) allowed_in_current_session = ( - permission_allowed_in_current_session and role_matches_current_session + permission_allowed_in_current_session + and role_matches_current_session + and not cross_host_block + and not identity_block + and not drift_block ) switching = gitea_config.is_runtime_switching_enabled() @@ -11834,33 +12039,19 @@ def gitea_resolve_task_capability( restart_required = False reason_msg = None - # Find matching configured profiles - matching_profiles = [] - if config and "profiles" in config: - for p_name, p_data in config["profiles"].items(): - p_allowed = p_data.get("allowed_operations") or [] - p_forbidden = p_data.get("forbidden_operations") or [] - p_allowed_n = [] - for op in p_allowed: - try: - p_allowed_n.append(gitea_config.normalize_operation(op)) - except Exception: - pass - p_forbidden_n = [] - for op in p_forbidden: - try: - p_forbidden_n.append(gitea_config.normalize_operation(op)) - except Exception: - pass - ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n) + # Matching profiles: same remote/host only (#714) — advisory, never activated. + matching_profiles = _matching_configured_profiles( + config, required_permission, remote=remote + ) + # Role filter for exclusive tasks + if config and "profiles" in config and task in role_exclusive_tasks: + filtered = [] + for p_name in matching_profiles: + p_data = (config.get("profiles") or {}).get(p_name) or {} p_role = (p_data.get("role") or "").strip() - p_role_ok = ( - task not in role_exclusive_tasks - or not p_role - or p_role == required_role - ) - if ok and p_role_ok: - matching_profiles.append(p_name) + if not p_role or p_role == required_role: + filtered.append(p_name) + matching_profiles = filtered configured = len(matching_profiles) > 0 available_in_session = allowed_in_current_session @@ -11876,16 +12067,35 @@ def gitea_resolve_task_capability( ) if not allowed_in_current_session: - if configured and switching: - restart_required = True + deny_parts: list[str] = [] + if cross_host_block: + deny_parts.extend(remote_assess.get("reasons") or []) + if identity_block: + deny_parts.extend(id_assess.get("reasons") or []) + if drift_block: + deny_parts.extend(ctx_assess.get("reasons") or []) + if not permission_allowed_in_current_session and not deny_parts: + deny_parts.append( + f"Active profile '{profile.get('profile_name')}' is not allowed " + f"to '{required_permission}' (no substitute profile activated; fail closed)." + ) + if role_mismatch_reason: + deny_parts.append(role_mismatch_reason) + if deny_parts: + reason_msg = "; ".join(deny_parts) + elif configured and switching: + # Same-remote profile exists but is not the active one — explicit switch only. + restart_required = False available_in_session = False reason_msg = ( - f"{required_role.capitalize()} profile exists but MCP server " - "was added after session startup and is not attached." + f"Active profile cannot perform '{required_permission}'. " + f"Same-remote matching profiles (advisory only, not activated): " + f"{matching_profiles}." ) elif not configured: reason_msg = ( - f"No profile configured with permission '{required_permission}'." + f"No same-remote profile configured with permission " + f"'{required_permission}' for remote '{remote}'." ) elif role_mismatch_reason: reason_msg = role_mismatch_reason @@ -11893,9 +12103,22 @@ def gitea_resolve_task_capability( next_safe_action = "None; ready for operations." if not allowed_in_current_session: - if switching: + if cross_host_block or identity_block or drift_block: different_namespace_required = False - next_safe_action = f"Switch to a profile that has the required permission by calling gitea_activate_profile (matching configured profiles: {matching_profiles})." + next_safe_action = ( + "BLOCKED + DIAGNOSE: session context/host/identity is inconsistent " + "or the active profile cannot serve this remote. Re-pin the correct " + "profile with gitea_activate_profile for the intended remote, verify " + "with gitea_whoami, and do not use cross-host profile substitution." + ) + elif switching: + different_namespace_required = False + next_safe_action = ( + f"Switch explicitly with gitea_activate_profile to a same-remote " + f"profile that allows '{required_permission}' " + f"(matching configured profiles: {matching_profiles}). " + "Capability resolution never auto-substitutes profiles (#714)." + ) else: different_namespace_required = True if required_role == "reviewer": @@ -11980,6 +12203,13 @@ def gitea_resolve_task_capability( "runtime_switching_supported": switching, "different_mcp_namespace_required": different_namespace_required, "exact_safe_next_action": next_safe_action, + # #714: immutable session context proof (must match whoami / runtime). + "requested_remote": remote, + "resolved_host": h, + "session_context_audit": session_ctx.mutation_context_audit_fields(), + "profile_remote_compatible": not cross_host_block, + "identity_match": not identity_block, + "auto_profile_substitution": False, } if reason_msg: result["reason"] = reason_msg diff --git a/session_context_binding.py b/session_context_binding.py new file mode 100644 index 0000000..d5bed09 --- /dev/null +++ b/session_context_binding.py @@ -0,0 +1,358 @@ +"""Session-immutable MCP mutation context (#714). + +Pins profile, remote, host, repository, identity, and role for the life of an +MCP process (or until an explicit ``gitea_activate_profile`` re-bind). + +Capability resolution and mutation gates must evaluate only the active +profile for the requested remote. Silent cross-host / cross-profile +substitution is forbidden. +""" + +from __future__ import annotations + +import os +import urllib.parse +from typing import Any + +# Process-local only — never a shared file (same rationale as mutation authority). +_SESSION_CONTEXT: dict[str, Any] | None = None + + +def clear_session_context() -> None: + """Reset session context (tests / process rebind).""" + global _SESSION_CONTEXT + _SESSION_CONTEXT = None + + +def get_session_context() -> dict[str, Any] | None: + """Return a shallow copy of the bound context, or None if unbound.""" + if _SESSION_CONTEXT is None: + return None + return dict(_SESSION_CONTEXT) + + +def profile_host(profile: dict | None) -> str | None: + """Hostname from profile base_url, lowercased, or None.""" + if not profile: + return None + base = (profile.get("base_url") or "").strip() + if not base: + return None + try: + parsed = urllib.parse.urlparse(base) + host = (parsed.netloc or parsed.path or "").strip().lower() + return host or None + except Exception: + return None + + +def remote_host(remote: str | None, remotes: dict | None) -> str | None: + """Hostname for a known remote key.""" + if not remote or not remotes: + return None + entry = remotes.get(remote) or {} + return (entry.get("host") or "").strip().lower() or None + + +def profile_matches_remote( + profile: dict | None, + remote: str | None, + remotes: dict | None, + *, + contexts: dict | None = None, +) -> bool: + """True when *profile* is bound to the same host/context as *remote*.""" + if not profile or not remote: + return False + r_host = remote_host(remote, remotes) + p_host = profile_host(profile) + if r_host and p_host: + return r_host == p_host + # Fall back to context name heuristics when base_url missing. + ctx = (profile.get("context") or "").strip().lower() + if not ctx or not contexts: + return False + ctx_data = contexts.get(ctx) or {} + gitea = ctx_data.get("gitea") or {} + base = (gitea.get("base_url") or "").strip() + if not base or not r_host: + return False + try: + parsed = urllib.parse.urlparse(base) + c_host = (parsed.netloc or parsed.path or "").strip().lower() + except Exception: + return False + return bool(c_host) and c_host == r_host + + +def bind_session_context( + *, + profile_name: str, + remote: str | None, + host: str | None, + identity: str | None, + repository: str | None = None, + org: str | None = None, + role_kind: str | None = None, + expected_username: str | None = None, + source: str = "bind", +) -> dict[str, Any]: + """Bind or re-bind the immutable session context (explicit activate path).""" + global _SESSION_CONTEXT + _SESSION_CONTEXT = { + "profile_name": (profile_name or "").strip() or None, + "remote": (remote or "").strip() or None, + "host": (host or "").strip().lower() or None, + "identity": (identity or "").strip() or None, + "repository": (repository or "").strip() or None, + "org": (org or "").strip() or None, + "role_kind": (role_kind or "").strip() or None, + "expected_username": (expected_username or "").strip() or None, + "source": source, + "pid": os.getpid(), + } + return get_session_context() # type: ignore[return-value] + + +def seed_session_context_if_unbound( + *, + profile_name: str, + remote: str | None, + host: str | None, + identity: str | None, + repository: str | None = None, + org: str | None = None, + role_kind: str | None = None, + expected_username: str | None = None, + source: str = "seed", + active_profile_override: str | None = None, + env_profile: str | None = None, +) -> dict[str, Any]: + """Bind when unbound, or rebind when the static env profile changes. + + Mid-session ``_active_profile_override`` changes that did **not** go + through ``gitea_activate_profile`` keep the prior bound context so + assess/mutation gates can detect drift and fail closed. + """ + global _SESSION_CONTEXT + live = (profile_name or "").strip() or None + if _SESSION_CONTEXT is None or _SESSION_CONTEXT.get("pid") != os.getpid(): + return bind_session_context( + profile_name=profile_name, + remote=remote, + host=host, + identity=identity, + repository=repository, + org=org, + role_kind=role_kind, + expected_username=expected_username, + source=source, + ) + bound_profile = _SESSION_CONTEXT.get("profile_name") + if live and bound_profile and live != bound_profile: + # Static env profile switch (tests / launcher) without override: rebind. + # Override present but bound not updated: leave bound for drift detection. + if active_profile_override is None and ( + env_profile is None or env_profile == live + ): + return bind_session_context( + profile_name=profile_name, + remote=remote, + host=host, + identity=identity, + repository=repository, + org=org, + role_kind=role_kind, + expected_username=expected_username, + source=f"{source}-env-rebind", + ) + return get_session_context() # type: ignore[return-value] + + +def assess_session_context( + *, + profile_name: str | None, + remote: str | None, + host: str | None = None, + identity: str | None = None, + repository: str | None = None, + org: str | None = None, + expected_username: str | None = None, + require_bound: bool = False, +) -> dict[str, Any]: + """Compare live values against the bound session context. + + Returns ``proven`` / ``block`` / ``reasons``. When unbound and + ``require_bound`` is false, does not block (caller may seed). When + unbound and ``require_bound`` is true, fails closed. + """ + reasons: list[str] = [] + ctx = _SESSION_CONTEXT + if ctx is None or ctx.get("pid") != os.getpid(): + if require_bound: + reasons.append( + "session mutation context is unbound; call gitea_whoami or " + "gitea_activate_profile before mutating (fail closed)" + ) + return _assessment(False, reasons, ctx) + return _assessment(True, reasons, ctx) + + live_profile = (profile_name or "").strip() or None + live_remote = (remote or "").strip() or None + live_host = (host or "").strip().lower() or None + live_identity = (identity or "").strip() or None + live_repo = (repository or "").strip() or None + live_org = (org or "").strip() or None + + if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"): + reasons.append( + f"profile drift: live '{live_profile}' != bound " + f"'{ctx.get('profile_name')}' (fail closed)" + ) + if ctx.get("remote") and live_remote and live_remote != ctx.get("remote"): + reasons.append( + f"remote drift: live '{live_remote}' != bound " + f"'{ctx.get('remote')}' (fail closed)" + ) + if ctx.get("host") and live_host and live_host != ctx.get("host"): + reasons.append( + f"host drift: live '{live_host}' != bound " + f"'{ctx.get('host')}' (fail closed)" + ) + if ( + ctx.get("identity") + and live_identity + and live_identity != ctx.get("identity") + ): + reasons.append( + f"identity drift: live '{live_identity}' != bound " + f"'{ctx.get('identity')}' (fail closed)" + ) + if ctx.get("repository") and live_repo and live_repo != ctx.get("repository"): + reasons.append( + f"repository drift: live '{live_repo}' != bound " + f"'{ctx.get('repository')}' (fail closed)" + ) + if ctx.get("org") and live_org and live_org != ctx.get("org"): + reasons.append( + f"org drift: live '{live_org}' != bound " + f"'{ctx.get('org')}' (fail closed)" + ) + + expected = expected_username or ctx.get("expected_username") + if expected and live_identity and live_identity != expected: + reasons.append( + f"identity mismatch: authenticated '{live_identity}' != " + f"profile expected '{expected}' (fail closed)" + ) + + return _assessment(not reasons, reasons, ctx) + + +def assess_identity_match( + *, + authenticated: str | None, + expected_username: str | None, +) -> dict[str, Any]: + """Fail closed when profile declares a username that does not match live auth.""" + reasons: list[str] = [] + auth = (authenticated or "").strip() or None + expected = (expected_username or "").strip() or None + if expected and auth and auth != expected: + reasons.append( + f"identity mismatch: authenticated '{auth}' != " + f"profile expected '{expected}' (fail closed)" + ) + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "authenticated": auth, + "expected_username": expected, + } + + +def filter_profiles_for_remote( + config: dict | None, + remote: str | None, + remotes: dict | None, +) -> list[str]: + """Profile names whose host/context matches *remote* (enabled only).""" + if not config or not remote: + return [] + profiles = config.get("profiles") or {} + contexts = config.get("contexts") or {} + names: list[str] = [] + for name, data in profiles.items(): + if not isinstance(data, dict): + continue + if not data.get("enabled", True): + continue + if profile_matches_remote(data, remote, remotes, contexts=contexts): + names.append(name) + return sorted(names) + + +def profile_allowed_for_remote( + profile: dict | None, + remote: str | None, + remotes: dict | None, + *, + contexts: dict | None = None, +) -> dict[str, Any]: + """Assess whether the active profile may serve *remote*.""" + reasons: list[str] = [] + if not profile: + reasons.append("active profile unresolved (fail closed)") + return {"proven": False, "block": True, "reasons": reasons} + if not remote: + return {"proven": True, "block": False, "reasons": reasons} + if not profile_matches_remote(profile, remote, remotes, contexts=contexts): + p_name = profile.get("profile_name") or profile.get("name") or "(unknown)" + p_host = profile_host(profile) or "(none)" + r_host = remote_host(remote, remotes) or "(none)" + reasons.append( + f"cross-host profile denial: profile '{p_name}' (host '{p_host}') " + f"cannot serve remote '{remote}' (host '{r_host}') (fail closed)" + ) + return {"proven": not reasons, "block": bool(reasons), "reasons": reasons} + + +def mutation_context_audit_fields( + ctx: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Fields to include in pre-mutation audit records.""" + data = ctx if ctx is not None else get_session_context() + if not data: + return { + "session_context_bound": False, + "session_profile": None, + "session_remote": None, + "session_host": None, + "session_identity": None, + "session_repository": None, + "session_org": None, + } + return { + "session_context_bound": True, + "session_profile": data.get("profile_name"), + "session_remote": data.get("remote"), + "session_host": data.get("host"), + "session_identity": data.get("identity"), + "session_repository": data.get("repository"), + "session_org": data.get("org"), + "session_role_kind": data.get("role_kind"), + "session_context_source": data.get("source"), + } + + +def _assessment( + proven: bool, reasons: list[str], ctx: dict[str, Any] | None +) -> dict[str, Any]: + return { + "proven": proven, + "block": not proven, + "reasons": list(reasons), + "bound_context": dict(ctx) if ctx else None, + "audit": mutation_context_audit_fields(ctx), + } diff --git a/tests/test_issue_714_session_context_immutability.py b/tests/test_issue_714_session_context_immutability.py new file mode 100644 index 0000000..eeac093 --- /dev/null +++ b/tests/test_issue_714_session_context_immutability.py @@ -0,0 +1,390 @@ +"""#714: fail closed on cross-host MCP profile drift and capability substitution.""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import gitea_config # noqa: E402 +import mcp_server # noqa: E402 +import session_context_binding as session_ctx # noqa: E402 + + +CONFIG_714 = { + "version": 2, + "allow_runtime_switching": True, + "contexts": { + "prgs": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"}, + }, + "mdcps": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.dadeschools.net"}, + }, + }, + "profiles": { + "prgs-author": { + "enabled": True, + "context": "prgs", + "role": "author", + "username": "jcwalker3", + "base_url": "https://gitea.prgs.cc", + "auth": {"type": "env", "name": "GITEA_TOKEN_PRGS_AUTHOR"}, + "allowed_operations": [ + "gitea.read", + "gitea.issue.create", + "gitea.issue.comment", + "gitea.issue.close", + "gitea.pr.create", + "gitea.pr.comment", + "gitea.branch.create", + "gitea.branch.push", + "gitea.repo.commit", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", + ], + "execution_profile": "prgs-author", + }, + "mdcps-reviewer": { + "enabled": True, + "context": "mdcps", + "role": "reviewer", + "username": "913443", + "base_url": "https://gitea.dadeschools.net", + "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_REVIEWER"}, + "allowed_operations": [ + "gitea.read", + "gitea.pr.review", + "gitea.pr.comment", + "gitea.pr.approve", + "gitea.pr.request_changes", + ], + "forbidden_operations": [ + "gitea.branch.create", + "gitea.branch.push", + "gitea.pr.create", + "gitea.pr.merge", + "gitea.repo.commit", + ], + "execution_profile": "mdcps-reviewer", + }, + "mdcps-author": { + "enabled": True, + "context": "mdcps", + "role": "author", + "username": "913443", + "base_url": "https://gitea.dadeschools.net", + "auth": {"type": "env", "name": "GITEA_TOKEN_MDCPS_AUTHOR"}, + "allowed_operations": [ + "gitea.read", + "gitea.issue.create", + "gitea.issue.comment", + "gitea.pr.create", + "gitea.pr.comment", + "gitea.branch.create", + "gitea.branch.push", + "gitea.repo.commit", + ], + "forbidden_operations": [ + "gitea.pr.approve", + "gitea.pr.merge", + "gitea.pr.request_changes", + ], + "execution_profile": "mdcps-author", + }, + }, +} + + +class TestIssue714SessionContextImmutability(unittest.TestCase): + def setUp(self): + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(CONFIG_714)) + self._remotes = patch.dict( + mcp_server.REMOTES, + { + "dadeschools": { + "host": "gitea.dadeschools.net", + "org": "913443", + "repo": "eAgenda", + }, + "prgs": { + "host": "gitea.prgs.cc", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + }, + }, + clear=False, + ) + self._remotes.start() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + mcp_server._MUTATION_AUTHORITY = None + session_ctx.clear_session_context() + self._env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": "mdcps-reviewer", + "GITEA_TOKEN_MDCPS_REVIEWER": "mdcps-reviewer-token", + "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token", + "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token", + } + + def tearDown(self): + self._remotes.stop() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + mcp_server._MUTATION_AUTHORITY = None + session_ctx.clear_session_context() + self._dir.cleanup() + + def _api_side_effect(self, method, url, header): + # Token identity mapping for whoami endpoint + auth = str(header) + if "mdcps-reviewer-token" in auth or "mdcps-author-token" in auth: + login = "913443" + else: + login = "jcwalker3" + return {"login": login, "full_name": "Test", "id": 1, "email": "t@example.com"} + + def test_pin_mdcps_reviewer_never_drifts_to_prgs_author(self): + """Reproduce the incident: pin mdcps-reviewer then resolve comment_issue.""" + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + act = mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + self.assertTrue(act["success"]) + self.assertEqual(act["after_profile"], "mdcps-reviewer") + + who1 = mcp_server.gitea_whoami(remote="dadeschools") + self.assertEqual(who1["profile"]["profile_name"], "mdcps-reviewer") + self.assertEqual(who1["username"], "913443") + + # Capability that mdcps-reviewer lacks — must NOT switch to prgs-author + res = mcp_server.gitea_resolve_task_capability( + task="comment_issue", remote="dadeschools" + ) + self.assertEqual(res["active_profile"], "mdcps-reviewer") + self.assertFalse(res["allowed_in_current_session"]) + self.assertFalse(res.get("auto_profile_substitution", True)) + self.assertNotIn("prgs-author", res["matching_configured_profile"]) + self.assertIn("mdcps-author", res["matching_configured_profile"]) + + who2 = mcp_server.gitea_whoami(remote="dadeschools") + rt = mcp_server.gitea_get_runtime_context(remote="dadeschools") + self.assertEqual(who2["profile"]["profile_name"], "mdcps-reviewer") + self.assertEqual(rt["active_profile"], "mdcps-reviewer") + # Still pinned — never prgs-author + self.assertEqual( + gitea_config.selected_profile_name(), "mdcps-reviewer" + ) + + def test_repeated_calls_remain_on_mdcps_reviewer(self): + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + for _ in range(5): + res = mcp_server.gitea_resolve_task_capability( + task="review_pr", remote="dadeschools" + ) + self.assertEqual(res["active_profile"], "mdcps-reviewer") + who = mcp_server.gitea_whoami(remote="dadeschools") + self.assertEqual(who["profile"]["profile_name"], "mdcps-reviewer") + rt = mcp_server.gitea_get_runtime_context(remote="dadeschools") + self.assertEqual(rt["active_profile"], "mdcps-reviewer") + + def test_dadeschools_request_cannot_resolve_through_prgs_author(self): + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + res = mcp_server.gitea_resolve_task_capability( + task="comment_issue", remote="dadeschools" + ) + self.assertNotEqual(res["active_profile"], "prgs-author") + self.assertNotIn("prgs-author", res["matching_configured_profile"]) + # Gate must not silently switch either + blocked = mcp_server._profile_permission_block( + "gitea.issue.comment", remote="dadeschools" + ) + self.assertIsNotNone(blocked) + self.assertFalse(blocked.get("success", True)) + self.assertEqual( + gitea_config.selected_profile_name(), "mdcps-reviewer" + ) + + def test_unsupported_capability_fails_closed_structured(self): + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + res = mcp_server.gitea_resolve_task_capability( + task="comment_issue", remote="dadeschools" + ) + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + self.assertIn("reason", res) + self.assertIn("exact_safe_next_action", res) + self.assertIn("activate_profile", res["exact_safe_next_action"]) + # Unknown task still fail closed + with self.assertRaises(ValueError): + mcp_server.gitea_resolve_task_capability( + task="reopen_issue", remote="dadeschools" + ) + + def test_identity_mismatch_blocks_mutation(self): + """Profile expects 913443 but authenticated as jcwalker3.""" + + def wrong_identity(method, url, header): + return { + "login": "jcwalker3", + "full_name": "Wrong", + "id": 9, + "email": "w@example.com", + } + + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=wrong_identity): + mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + blocked = mcp_server._session_context_mutation_block( + remote="dadeschools" + ) + self.assertIsNotNone(blocked) + self.assertTrue( + any("identity mismatch" in r for r in blocked["reasons"]) + ) + + def test_repository_host_mismatch_blocks_mutation(self): + """mdcps-reviewer cannot serve prgs remote.""" + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + blocked = mcp_server._session_context_mutation_block(remote="prgs") + self.assertIsNotNone(blocked) + self.assertTrue( + any("cross-host" in r for r in blocked["reasons"]) + ) + + def test_drift_cannot_reach_write(self): + """Simulated mid-session profile override cannot pass permission gate.""" + with patch.dict(os.environ, self._env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + mcp_server.gitea_activate_profile( + profile_name="mdcps-reviewer", remote="dadeschools" + ) + # Bind session as mdcps-reviewer + self.assertEqual( + session_ctx.get_session_context()["profile_name"], + "mdcps-reviewer", + ) + # Hostile override without activate_profile + gitea_config._active_profile_override = "prgs-author" + blocked = mcp_server._session_context_mutation_block( + remote="dadeschools" + ) + self.assertIsNotNone(blocked) + self.assertTrue(any("drift" in r or "cross-host" in r for r in blocked["reasons"])) + + def test_static_prgs_author_still_works(self): + env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": "prgs-author", + "GITEA_TOKEN_PRGS_AUTHOR": "prgs-author-token", + } + with patch.dict(os.environ, env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + session_ctx.clear_session_context() + gitea_config._active_profile_override = None + res = mcp_server.gitea_resolve_task_capability( + task="create_issue", remote="prgs" + ) + self.assertEqual(res["active_profile"], "prgs-author") + self.assertTrue(res["allowed_in_current_session"]) + self.assertEqual(res["active_identity"], "jcwalker3") + + def test_static_mdcps_author_still_works(self): + env = { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": "mdcps-author", + "GITEA_TOKEN_MDCPS_AUTHOR": "mdcps-author-token", + } + with patch.dict(os.environ, env, clear=False): + with patch("mcp_server.api_request", side_effect=self._api_side_effect): + session_ctx.clear_session_context() + gitea_config._active_profile_override = None + res = mcp_server.gitea_resolve_task_capability( + task="comment_issue", remote="dadeschools" + ) + self.assertEqual(res["active_profile"], "mdcps-author") + self.assertTrue(res["allowed_in_current_session"]) + self.assertNotIn("prgs-author", res["matching_configured_profile"]) + + +class TestSessionContextBindingUnit(unittest.TestCase): + def setUp(self): + session_ctx.clear_session_context() + + def tearDown(self): + session_ctx.clear_session_context() + + def test_profile_matches_remote_by_base_url(self): + remotes = { + "dadeschools": {"host": "gitea.dadeschools.net"}, + "prgs": {"host": "gitea.prgs.cc"}, + } + mdcps = {"base_url": "https://gitea.dadeschools.net", "context": "mdcps"} + prgs = {"base_url": "https://gitea.prgs.cc", "context": "prgs"} + self.assertTrue( + session_ctx.profile_matches_remote(mdcps, "dadeschools", remotes) + ) + self.assertFalse(session_ctx.profile_matches_remote(mdcps, "prgs", remotes)) + self.assertTrue(session_ctx.profile_matches_remote(prgs, "prgs", remotes)) + + def test_bind_and_detect_drift(self): + session_ctx.bind_session_context( + profile_name="mdcps-reviewer", + remote="dadeschools", + host="gitea.dadeschools.net", + identity="913443", + source="test", + ) + ok = session_ctx.assess_session_context( + profile_name="mdcps-reviewer", + remote="dadeschools", + host="gitea.dadeschools.net", + identity="913443", + ) + self.assertTrue(ok["proven"]) + bad = session_ctx.assess_session_context( + profile_name="prgs-author", + remote="dadeschools", + host="gitea.dadeschools.net", + identity="jcwalker3", + ) + self.assertTrue(bad["block"]) + self.assertTrue(any("drift" in r for r in bad["reasons"])) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_operation_scoped_roles.py b/tests/test_operation_scoped_roles.py index c382085..848bcde 100644 --- a/tests/test_operation_scoped_roles.py +++ b/tests/test_operation_scoped_roles.py @@ -1,4 +1,9 @@ -"""Tests for operation-scoped role selection and automatic dispatch switching (#228).""" +"""Tests for operation-scoped role selection (#228, updated by #714). + +#714 removes silent automatic profile substitution. Capability resolution +and mutation gates evaluate only the active profile; explicit +``gitea_activate_profile`` is required to switch roles. +""" import os import sys import json @@ -11,10 +16,12 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par import gitea_config import gitea_auth import mcp_server +import session_context_binding as session_ctx from reviewer_worktree import assess_author_worktree_continuity CONFIG_TEST = { "version": 2, + "allow_runtime_switching": True, "contexts": { "ctx": { "enabled": True, @@ -30,6 +37,7 @@ CONFIG_TEST = { "context": "ctx", "role": "author", "username": "author-user", + "base_url": "https://gitea.example.com", "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, "allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], @@ -40,6 +48,7 @@ CONFIG_TEST = { "context": "ctx", "role": "reviewer", "username": "reviewer-user", + "base_url": "https://gitea.example.com", "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"}, "allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"], "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], @@ -59,6 +68,7 @@ class TestOperationScopedRoles(unittest.TestCase): mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None mcp_server._MUTATION_AUTHORITY = None + session_ctx.clear_session_context() self._dir = tempfile.TemporaryDirectory() self.config_path = os.path.join(self._dir.name, "profiles.json") self._write_config(CONFIG_TEST) @@ -68,6 +78,7 @@ class TestOperationScopedRoles(unittest.TestCase): mcp_server._IDENTITY_CACHE.clear() gitea_config._active_profile_override = None mcp_server._MUTATION_AUTHORITY = None + session_ctx.clear_session_context() self._dir.cleanup() def _write_config(self, obj): @@ -85,62 +96,71 @@ class TestOperationScopedRoles(unittest.TestCase): return env @patch("mcp_server.api_request") - def test_auto_switch_to_reviewer(self, mock_api): - # mock identity resolution to return username matching profile + def test_no_auto_switch_to_reviewer(self, mock_api): + # #714: capability resolution must not silently switch author → reviewer mock_api.side_effect = lambda method, url, header: ( {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} ) with patch.dict(os.environ, self._env("author-profile")): - # initially we are author-profile self.assertEqual(gitea_config.selected_profile_name(), "author-profile") self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile") - # resolve a reviewer task (review_pr) res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") - - # verify it automatically switched to reviewer-profile - self.assertTrue(res["allowed_in_current_session"]) - self.assertTrue(res["available_in_session"]) - self.assertEqual(res["active_profile"], "reviewer-profile") - self.assertEqual(res["active_identity"], "reviewer-user") - self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") + + self.assertFalse(res["allowed_in_current_session"]) + self.assertFalse(res["available_in_session"]) + self.assertEqual(res["active_profile"], "author-profile") + self.assertEqual(res["active_identity"], "author-user") + self.assertEqual(gitea_config.selected_profile_name(), "author-profile") + self.assertIn("reviewer-profile", res["matching_configured_profile"]) + self.assertFalse(res.get("auto_profile_substitution", True)) @patch("mcp_server.api_request") - def test_auto_switch_to_author(self, mock_api): + def test_no_auto_switch_to_author(self, mock_api): mock_api.side_effect = lambda method, url, header: ( {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} ) with patch.dict(os.environ, self._env("reviewer-profile")): - # initially we are reviewer-profile self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") - # resolve an author task (create_issue) res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs") - - # verify it automatically switched to author-profile - self.assertTrue(res["allowed_in_current_session"]) - self.assertTrue(res["available_in_session"]) - self.assertEqual(res["active_profile"], "author-profile") - self.assertEqual(res["active_identity"], "author-user") - self.assertEqual(gitea_config.selected_profile_name(), "author-profile") + + self.assertFalse(res["allowed_in_current_session"]) + self.assertFalse(res["available_in_session"]) + self.assertEqual(res["active_profile"], "reviewer-profile") + self.assertEqual(res["active_identity"], "reviewer-user") + self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") + self.assertIn("author-profile", res["matching_configured_profile"]) @patch("mcp_server.api_request") - def test_restart_required_when_unattached(self, mock_api): + def test_explicit_activate_profile_still_works(self, mock_api): + mock_api.side_effect = lambda method, url, header: ( + {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} + ) + with patch.dict(os.environ, self._env("author-profile")): + act = mcp_server.gitea_activate_profile( + profile_name="reviewer-profile", remote="prgs" + ) + self.assertTrue(act["success"]) + self.assertEqual(act["after_profile"], "reviewer-profile") + res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") + self.assertTrue(res["allowed_in_current_session"]) + self.assertEqual(res["active_profile"], "reviewer-profile") + + @patch("mcp_server.api_request") + def test_denied_when_matching_profile_token_missing(self, mock_api): mock_api.side_effect = lambda method, url, header: {"login": "author-user"} # launch without reviewer token in env with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)): self.assertEqual(gitea_config.selected_profile_name(), "author-profile") - # resolve a reviewer task (review_pr) res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs") - - # verify it did NOT switch and reports restart_required + self.assertFalse(res["allowed_in_current_session"]) self.assertFalse(res["available_in_session"]) self.assertTrue(res["configured"]) - self.assertTrue(res["restart_required"]) self.assertTrue(res["stop_required"]) - self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"]) + self.assertEqual(res["active_profile"], "author-profile") def test_author_continuity_dirty_worktree(self): # author is allowed to keep dirty worktree @@ -161,20 +181,21 @@ class TestOperationScopedRoles(unittest.TestCase): self.assertFalse(res2["proven"]) @patch("mcp_server.api_request") - def test_mutating_actions_auto_switch(self, mock_api): + def test_mutating_actions_do_not_auto_switch(self, mock_api): mock_api.side_effect = lambda method, url, header: ( {"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"} ) with patch.dict(os.environ, self._env("author-profile")): - # verify we are author-profile self.assertEqual(gitea_config.selected_profile_name(), "author-profile") - # call a reviewer mutation check helper (like _profile_permission_block with reviewer permission) - blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs") - self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block) + blocked = mcp_server._profile_permission_block( + "gitea.pr.merge", remote="prgs" + ) + self.assertIsNotNone(blocked) + self.assertFalse(blocked.get("success", True)) - # verify we dynamically switched to reviewer-profile - self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile") + # profile must remain author — no silent substitution + self.assertEqual(gitea_config.selected_profile_name(), "author-profile") if __name__ == "__main__":