Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions crates/animus-mcp-oauth/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ rmcp = { version = "1.7", features = [

[dev-dependencies]
protocol = { workspace = true, features = ["test-utils"] }
orchestrator-config = { workspace = true, features = ["test-utils"] }
tempfile = "3"
axum = "0.8"
http = "1"
Expand Down
40 changes: 31 additions & 9 deletions crates/animus-mcp-oauth/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -59,33 +59,55 @@ impl McpSession {
timeout: Duration,
) -> Result<Self> {
crate::ensure_crypto_provider();
// The ONE `config_source` resolution per `mcp call`. The resolved URL +
// flow are threaded to the spawned proxy (`--url` + `--auth-code`) so the
// proxy TRUSTS them and does NOT re-resolve — collapsing the historical
// 3× config_source spawn amplification (CLI + proxy-bin + proxy-connect)
// down to this single touch.
let resolution = resolve_server_url(project_root, server, url_override)?;
// OAuth servers (every flow) are served through the local proxy, which
// resolves + injects the live bearer itself; plain-HTTP servers are
// connected directly.
let uses_oauth = resolution.is_authorization_code || resolution.broker_oauth.is_some();
if uses_oauth {
Self::connect_via_proxy(project_root, server, url_override, timeout).await
Self::connect_via_proxy(project_root, server, &resolution, timeout).await
} else {
Self::connect_http(&resolution.url, timeout).await
}
}

/// Spawn `animus-mcp-proxy --server <server> [--url <url>] --project-root
/// <root>` and connect an MCP client to its stdio. The proxy injects the
/// cached bearer; no secret is passed on argv or read here. stderr is
/// inherited so proxy diagnostics (e.g. "run `animus mcp auth`") reach the
/// user; stdout carries only JSON-RPC frames.
/// Spawn `animus-mcp-proxy --server <server> --url <resolved-url>
/// [--auth-code] --project-root <root>` and connect an MCP client to its
/// stdio. The proxy injects the cached bearer; no secret is passed on argv
/// or read here. stderr is inherited so proxy diagnostics (e.g. "run `animus
/// mcp auth`") reach the user; stdout carries only JSON-RPC frames.
///
/// Passing the already-resolved `--url` (and, for keychain servers,
/// `--auth-code`) lets the proxy skip its own `config_source` lookup — so a
/// bulk `mcp call` burst does one source resolution per call, here, instead
/// of one per proxy spawn.
async fn connect_via_proxy(
project_root: &Path,
server: &str,
url_override: Option<&str>,
resolution: &crate::config::ServerResolution,
timeout: Duration,
) -> Result<Self> {
let mut cmd = Command::new(mcp_proxy_command());
cmd.arg("--server").arg(server).arg("--project-root").arg(project_root);
if let Some(url) = url_override.filter(|u| !u.trim().is_empty()) {
cmd.arg("--url").arg(url);
// Pass the resolved `--url` for BOTH flows so the proxy binds the exact
// upstream the parent selected (honoring a `--url` override) instead of
// re-resolving the server name to a possibly-different same-named entry.
if !resolution.url.trim().is_empty() {
cmd.arg("--url").arg(&resolution.url);
}
// `--auth-code` is added ONLY for keychain (`authorization_code`) servers:
// it tells the proxy to trust `--url` and skip its own config_source
// lookup entirely (the amplification cut). Broker flows omit it because
// they still need their full oauth block from config; the proxy re-resolves
// those — and if the source is down it errors cleanly (ConfigSourceUnavailable)
// rather than misrouting the broker server to the keychain path.
if resolution.is_authorization_code {
cmd.arg("--auth-code");
}
// `TokioChildProcess::new` pipes stdin/stdout and inherits stderr; the
// child is killed on drop of the returned transport (held by the
Expand Down
159 changes: 129 additions & 30 deletions crates/animus-mcp-oauth/src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use std::path::Path;
use std::sync::Arc;

use orchestrator_config::workflow_config::{
load_workflow_config_or_default, load_workflow_config_with_metadata, OauthConfig, OauthFlow,
try_load_workflow_config, OauthConfig, OauthFlow, WorkflowConfigAvailability,
};
use orchestrator_core::SecretStore;
use protocol::repository_scope::{repository_scope_for_path, scoped_state_root};
Expand All @@ -21,6 +21,13 @@ pub enum ServerResolutionError {
UnknownServer(String),
#[error("MCP server `{0}` has no `url`; an HTTP-transport URL is required for OAuth")]
MissingUrl(String),
/// The `config_source` failed to load (spawn / RPC / DB overload / validation),
/// so the server's config could NOT be determined. This is distinct from
/// [`UnknownServer`], which fires only when the config loaded and the server
/// is genuinely absent. Callers should retry / report a transient source
/// error rather than "server not configured".
#[error("MCP server `{0}` could not be resolved: the workflow config source failed to load ({1}). This is a config-source error, not a missing-server error — retry shortly or check the `config_source` plugin.")]
ConfigSourceUnavailable(String, String),
#[error("failed to load workflow config: {0}")]
WorkflowConfig(String),
#[error("failed to load project config: {0}")]
Expand Down Expand Up @@ -99,22 +106,30 @@ pub fn resolve_server_url(
) -> Result<ServerResolution, ServerResolutionError> {
// Workflow config mcp_servers first (authoritative for daemon runs).
//
// A malformed `.animus/workflows.yaml` must surface its YAML/validation
// error rather than silently falling back to the builtin default (which
// would mislead the user into "unknown server"). But an *absent* workflow
// config is normal (project-config-only setups), and
// `load_workflow_config_with_metadata` returns a "missing" error in that
// case. So: only propagate the load error when a workflow YAML source
// actually exists; otherwise fall back to the builtin default and continue
// to project config.
let loaded = match load_workflow_config_with_metadata(project_root, None) {
Ok(loaded) => loaded,
Err(err) if workflow_yaml_present(project_root) => {
return Err(ServerResolutionError::WorkflowConfig(err.to_string()));
// Use the NON-SWALLOWING loader so a transient `config_source` failure (DB
// overload under bulk `mcp call`) is NOT degraded into an empty config that
// then misreports the server as "not defined". `try_load_workflow_config`
// distinguishes three cases:
// * Loaded — use its `mcp_servers`.
// * NoSource — no config_source configured; benign, fall through to
// project config (project-config-only setups).
// * SourceUnavailable — the source failed; surface a DISTINCT retryable
// error. We do this EVEN when the caller supplied a
// `--url`: a `--url` overrides only the upstream
// endpoint, not the server's oauth FLOW/block, so
// synthesizing an `authorization_code` resolution from
// the URL alone would misroute a broker-flow server
// (`manual_bearer` / `client_credentials` /
// `refresh_token`) to the keychain path. Erroring lets
// the caller retry once the source recovers.
let loaded = match try_load_workflow_config(project_root, None) {
WorkflowConfigAvailability::Loaded(loaded) => Some(loaded),
WorkflowConfigAvailability::NoSource => None,
WorkflowConfigAvailability::SourceUnavailable(err) => {
return Err(ServerResolutionError::ConfigSourceUnavailable(server.to_string(), err.to_string()));
}
Err(_) => load_workflow_config_or_default(project_root),
};
if let Some(def) = loaded.config.mcp_servers.get(server) {
if let Some(def) = loaded.as_ref().and_then(|loaded| loaded.config.mcp_servers.get(server)) {
return finalize(server, url_override, def.url.clone(), def.oauth.clone());
}

Expand All @@ -139,21 +154,6 @@ pub fn resolve_server_url(
}
}

/// True when a workflow YAML source exists at `.animus/workflows.yaml` or
/// any `.animus/workflows/*.yaml`. Used to decide whether a workflow-config
/// load failure is a real (malformed-config) error worth propagating vs the
/// benign "no workflow config" case.
fn workflow_yaml_present(project_root: &Path) -> bool {
let animus = project_root.join(".animus");
if animus.join("workflows.yaml").exists() {
return true;
}
let dir = animus.join("workflows");
std::fs::read_dir(&dir)
.map(|entries| entries.flatten().any(|e| e.path().extension().is_some_and(|ext| ext == "yaml" || ext == "yml")))
.unwrap_or(false)
}

fn finalize(
server: &str,
url_override: Option<&str>,
Expand Down Expand Up @@ -188,3 +188,102 @@ fn finalize(
}),
}
}

#[cfg(test)]
mod tests {
use super::*;
use orchestrator_config::workflow_config::builtin_workflow_config;
use orchestrator_config::workflow_config::config_source_client::test_seam;
use std::sync::{Mutex, MutexGuard, OnceLock};

// These tests mutate the process-global config_source test seam + HOME env,
// so serialize them under one lock held for the whole test body.
fn serial() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}

/// RAII guard that points `HOME` at a temp dir for the test and RESTORES the
/// previous value on drop, so the mutation never leaks to sibling tests.
struct HomeGuard {
prev: Option<std::ffi::OsString>,
}

impl HomeGuard {
fn set(path: &Path) -> Self {
let prev = std::env::var_os("HOME");
std::env::set_var("HOME", path);
Self { prev }
}
}

impl Drop for HomeGuard {
fn drop(&mut self) {
match self.prev.take() {
Some(v) => std::env::set_var("HOME", v),
None => std::env::remove_var("HOME"),
}
}
}

/// Common fixture: hold the serial lock, isolate `HOME` to a fresh tempdir
/// (restored on drop). Bound in the caller in declaration order so the
/// tempdir drops LAST — after `HomeGuard` has restored `HOME`.
fn fixture() -> (MutexGuard<'static, ()>, tempfile::TempDir, HomeGuard) {
let guard = serial().lock().unwrap_or_else(|p| p.into_inner());
let temp = tempfile::tempdir().expect("tempdir");
let home = HomeGuard::set(temp.path());
(guard, temp, home)
}

/// TASK-326 (fix d): a transient config_source failure must surface as the
/// DISTINCT `ConfigSourceUnavailable`, NOT `UnknownServer` — otherwise the
/// caller concludes "krisp isn't configured" when the source is merely down.
#[test]
fn source_failure_yields_config_source_unavailable_not_unknown_server() {
let (_guard, temp, _home) = fixture();
let root = temp.path();

let _fail = test_seam::install_failure(root, "config_source RPC timed out");
let err = resolve_server_url(root, "krisp", None).expect_err("must fail when source is down");
assert!(
matches!(err, ServerResolutionError::ConfigSourceUnavailable(ref s, _) if s == "krisp"),
"expected ConfigSourceUnavailable, got {err:?}"
);
}

/// Even with an explicit `--url`, a source outage yields
/// `ConfigSourceUnavailable` (never a synthesized `authorization_code`
/// resolution): a `--url` overrides only the upstream endpoint, not the
/// server's oauth FLOW, so guessing the flow could misroute a broker server
/// to the keychain path. The caller retries once the source recovers.
#[test]
fn explicit_url_still_errors_when_source_unavailable() {
let (_guard, temp, _home) = fixture();
let root = temp.path();

let _fail = test_seam::install_failure(root, "config_source RPC timed out");
let err = resolve_server_url(root, "krisp", Some("https://example.test/mcp"))
.expect_err("a source outage must error even with --url, to avoid misrouting a broker server");
assert!(
matches!(err, ServerResolutionError::ConfigSourceUnavailable(ref s, _) if s == "krisp"),
"expected ConfigSourceUnavailable, got {err:?}"
);
}

/// When the config LOADS but the server is genuinely absent, the error is
/// `UnknownServer` (the pre-existing behavior) — never `ConfigSourceUnavailable`.
#[test]
fn loaded_config_missing_server_yields_unknown_server() {
let (_guard, temp, _home) = fixture();
let root = temp.path();

// A base with no mcp_servers loads cleanly; the server is simply absent.
let _base = test_seam::install(root, builtin_workflow_config());
let err = resolve_server_url(root, "krisp", None).expect_err("absent server must error");
assert!(
matches!(err, ServerResolutionError::UnknownServer(ref s) if s == "krisp"),
"expected UnknownServer, got {err:?}"
);
}
}
48 changes: 35 additions & 13 deletions crates/animus-mcp-oauth/src/flow.rs
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ enum Exchange {
}

use crate::callback::CallbackListener;
use crate::config::{build_secret_store, resolve_principal_id, resolve_server_url};
use crate::config::{build_secret_store, resolve_principal_id, resolve_server_url, ServerResolutionError};
use crate::keychain_store::KeychainCredentialStore;
use crate::pending::{PendingAuth, PendingStore};
use crate::state_store::PersistentStateStore;
Expand Down Expand Up @@ -745,17 +745,26 @@ pub async fn auth_status(project_root: &Path, server: Option<&str>, url_override

let servers: Vec<String> = match server {
Some(s) => vec![s.to_string()],
None => authorization_code_servers(project_root),
// Non-swallowing enumeration: a transient config_source failure surfaces
// as an Err here rather than an empty `servers:[]` that would read as
// "no OAuth servers configured".
None => authorization_code_servers(project_root)?,
};

let mut out = Vec::with_capacity(servers.len());
for name in servers {
// Tokens are keyed by the upstream URL too, so resolve it. A server
// that can't be resolved (e.g. dropped from config and no --url) is
// reported as unauthenticated rather than failing the whole report.
let Some(url) = resolve_server_url(project_root, &name, url_override).ok().map(|r| r.url) else {
out.push(server_state_from_creds(&name, &principal, None));
continue;
// reported as unauthenticated rather than failing the whole report —
// EXCEPT a transient config_source failure, which must surface (not be
// silently reported as an unauthenticated server) so the user retries.
let url = match resolve_server_url(project_root, &name, url_override) {
Ok(resolution) => resolution.url,
Err(err @ ServerResolutionError::ConfigSourceUnavailable(..)) => return Err(err.into()),
Err(_) => {
out.push(server_state_from_creds(&name, &principal, None));
continue;
}
};
let store = KeychainCredentialStore::new(secrets.clone(), &name, &principal, &url);
let creds =
Expand Down Expand Up @@ -814,14 +823,27 @@ fn server_state_from_creds(server: &str, principal: &str, creds: Option<StoredCr

/// Servers in workflow/project config carrying an `authorization_code`
/// oauth flow.
fn authorization_code_servers(project_root: &Path) -> Vec<String> {
use orchestrator_config::workflow_config::{load_workflow_config_or_default, OauthFlow};
///
/// Uses the NON-SWALLOWING loader: a transient `config_source` failure returns
/// `Err` so `auth_status` reports a transient error rather than an empty
/// `servers:[]` (which would read as "no OAuth servers configured"). A genuinely
/// absent config source (`NoSource`) is benign — enumeration proceeds from
/// project config only.
fn authorization_code_servers(project_root: &Path) -> Result<Vec<String>> {
use orchestrator_config::workflow_config::{try_load_workflow_config, OauthFlow, WorkflowConfigAvailability};
let mut names = std::collections::BTreeSet::new();

let loaded = load_workflow_config_or_default(project_root);
for (name, def) in &loaded.config.mcp_servers {
if def.oauth.as_ref().is_some_and(|o| o.flow == OauthFlow::AuthorizationCode) {
names.insert(name.clone());
match try_load_workflow_config(project_root, None) {
WorkflowConfigAvailability::Loaded(loaded) => {
for (name, def) in &loaded.config.mcp_servers {
if def.oauth.as_ref().is_some_and(|o| o.flow == OauthFlow::AuthorizationCode) {
names.insert(name.clone());
}
}
}
WorkflowConfigAvailability::NoSource => {}
WorkflowConfigAvailability::SourceUnavailable(err) => {
return Err(anyhow!("workflow config source unavailable while listing OAuth servers: {err}"));
}
}

Expand All @@ -838,7 +860,7 @@ fn authorization_code_servers(project_root: &Path) -> Vec<String> {
}
}

names.into_iter().collect()
Ok(names.into_iter().collect())
}

/// Extract the `state` query parameter from an authorization URL.
Expand Down
Loading
Loading