From b286b2e2267073545418da1ea5a6402f9136eed7 Mon Sep 17 00:00:00 2001 From: YOMXXX Date: Tue, 30 Jun 2026 22:50:24 +0800 Subject: [PATCH] fix(composio): stop direct-mode invalid key polling (#4318) --- app/scripts/e2e-run-session.sh | 2 + app/scripts/e2e-web-session.sh | 2 + scripts/mock-api/routes/integrations.mjs | 9 ++ src/openhuman/composio/client.rs | 128 ++++++++++----- src/openhuman/composio/client_tests.rs | 78 +++++++++- src/openhuman/composio/direct_auth/mod.rs | 125 +++++++++++++++ src/openhuman/composio/mod.rs | 1 + src/openhuman/composio/ops/direct_mode.rs | 38 +++++ src/openhuman/composio/ops_tests.rs | 147 +++++++++++++++++- src/openhuman/composio/tools/direct.rs | 4 + ...osio_credentials_state_raw_coverage_e2e.rs | 13 +- tests/composio_ops_raw_coverage_e2e.rs | 48 +++++- tests/worker_c_modules_e2e.rs | 43 ++++- 13 files changed, 592 insertions(+), 46 deletions(-) create mode 100644 src/openhuman/composio/direct_auth/mod.rs diff --git a/app/scripts/e2e-run-session.sh b/app/scripts/e2e-run-session.sh index abeb1976c..f0b03d421 100755 --- a/app/scripts/e2e-run-session.sh +++ b/app/scripts/e2e-run-session.sh @@ -174,6 +174,8 @@ export CEF_CDP_PORT # rest of the mock backend. The core reads this at TelegramChannel::new() time, # which runs after the config is fully loaded. export OPENHUMAN_TELEGRAM_BOT_API_BASE="http://127.0.0.1:${E2E_MOCK_PORT}" +export OPENHUMAN_COMPOSIO_DIRECT_BASE_V2="http://127.0.0.1:${E2E_MOCK_PORT}" +export OPENHUMAN_COMPOSIO_DIRECT_BASE_V3="http://127.0.0.1:${E2E_MOCK_PORT}" echo "[runner] Killing any running OpenHuman instances..." case "$OS" in diff --git a/app/scripts/e2e-web-session.sh b/app/scripts/e2e-web-session.sh index 51aacae03..9976f0d50 100755 --- a/app/scripts/e2e-web-session.sh +++ b/app/scripts/e2e-web-session.sh @@ -129,6 +129,8 @@ fi export OPENHUMAN_CORE_TOKEN="$PW_CORE_RPC_TOKEN" export OPENHUMAN_TELEGRAM_BOT_API_BASE="http://127.0.0.1:${E2E_MOCK_PORT}" +export OPENHUMAN_COMPOSIO_DIRECT_BASE_V2="http://127.0.0.1:${E2E_MOCK_PORT}" +export OPENHUMAN_COMPOSIO_DIRECT_BASE_V3="http://127.0.0.1:${E2E_MOCK_PORT}" # Keep the standalone core aligned with the Rust mock runner: sub-agent # orchestration builds large async futures and can overflow the default stack. export RUST_MIN_STACK="${RUST_MIN_STACK:-16777216}" diff --git a/scripts/mock-api/routes/integrations.mjs b/scripts/mock-api/routes/integrations.mjs index a1e5df08b..93d39ef83 100644 --- a/scripts/mock-api/routes/integrations.mjs +++ b/scripts/mock-api/routes/integrations.mjs @@ -197,6 +197,15 @@ export function handleIntegrations(ctx) { // (chat/completions is handled by routes/llm.mjs ahead of this route) // ── Composio ─────────────────────────────────────────────── + if ( + method === "GET" && + /^\/(?:api\/v3\/)?connected_accounts\/?(\?.*)?$/.test(url) + ) { + const items = parseBehaviorJson("composioDirectConnectedAccounts", []); + json(res, 200, { items }); + return true; + } + if ( method === "GET" && /^\/agent-integrations\/composio\/toolkits\/?(\?.*)?$/.test(url) diff --git a/src/openhuman/composio/client.rs b/src/openhuman/composio/client.rs index 1e9326b61..563ebf3c3 100644 --- a/src/openhuman/composio/client.rs +++ b/src/openhuman/composio/client.rs @@ -705,6 +705,54 @@ pub enum ComposioClientKind { Direct(Arc), } +pub(crate) fn create_direct_composio_tool_for_api_key( + config: &crate::openhuman::config::Config, + api_key: &str, +) -> anyhow::Result> { + let api_key = api_key.trim(); + if api_key.is_empty() { + anyhow::bail!("composio direct api key must not be empty"); + } + + // The direct client takes a `SecurityPolicy` for `Tool::execute` + // gating, but the factory's job is only to materialize a *client* + // — it does not actually invoke `execute()` itself, so the + // default policy is sufficient here. Callers that go through + // the `Tool` surface re-acquire the live policy from their own + // context. + let security = Arc::new(crate::openhuman::security::SecurityPolicy::default()); + #[cfg(debug_assertions)] + let tool = match ( + std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2").ok(), + std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3").ok(), + ) { + (Some(base_v2), Some(base_v3)) => { + crate::openhuman::tools::ComposioTool::new_with_base_urls_for_loopback( + api_key, + Some(config.composio.entity_id.as_str()), + security, + base_v2, + base_v3, + ) + .map_err(|e| { + anyhow::anyhow!("invalid debug composio direct loopback base override: {e}") + })? + } + _ => crate::openhuman::tools::ComposioTool::new( + api_key, + Some(config.composio.entity_id.as_str()), + security, + ), + }; + #[cfg(not(debug_assertions))] + let tool = crate::openhuman::tools::ComposioTool::new( + api_key, + Some(config.composio.entity_id.as_str()), + security, + ); + Ok(Arc::new(tool)) +} + impl ComposioClientKind { /// Returns `"backend"` or `"direct"` — handy for logging and tests. pub fn mode(&self) -> &'static str { @@ -772,47 +820,12 @@ pub fn create_composio_client( ) })?; - // The direct client takes a `SecurityPolicy` for `Tool::execute` - // gating, but the factory's job is only to materialize a *client* - // — it does not actually invoke `execute()` itself, so the - // default policy is sufficient here. Callers that go through - // the `Tool` surface re-acquire the live policy from their own - // context. - let security = Arc::new(crate::openhuman::security::SecurityPolicy::default()); - #[cfg(debug_assertions)] - let tool = match ( - std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2").ok(), - std::env::var("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3").ok(), - ) { - (Some(base_v2), Some(base_v3)) => { - crate::openhuman::tools::ComposioTool::new_with_base_urls_for_loopback( - &api_key, - Some(config.composio.entity_id.as_str()), - security, - base_v2, - base_v3, - ) - .map_err(|e| { - anyhow::anyhow!("invalid debug composio direct loopback base override: {e}") - })? - } - _ => crate::openhuman::tools::ComposioTool::new( - &api_key, - Some(config.composio.entity_id.as_str()), - security, - ), - }; - #[cfg(not(debug_assertions))] - let tool = crate::openhuman::tools::ComposioTool::new( - &api_key, - Some(config.composio.entity_id.as_str()), - security, - ); + let tool = create_direct_composio_tool_for_api_key(config, &api_key)?; tracing::debug!( key_len = api_key.len(), "[composio-factory] resolved direct variant (key redacted)" ); - Ok(ComposioClientKind::Direct(Arc::new(tool))) + Ok(ComposioClientKind::Direct(tool)) } unknown => { tracing::warn!(mode = %unknown, "[composio-factory] unknown composio mode"); @@ -836,7 +849,7 @@ pub fn create_composio_client( // direct-mode plumbing can see the full envelope-translation surface // in one place. -use super::types::ComposioConnection; +use super::{direct_auth, types::ComposioConnection}; /// Direct-mode counterpart to [`ComposioClient::authorize`]. Calls /// Composio v3 `/connected_accounts/link` via @@ -963,7 +976,44 @@ pub async fn direct_list_connections( direct: &Arc, ) -> anyhow::Result { tracing::debug!("[composio-direct] list_connections: GET v3 /connected_accounts"); - let items = direct.list_connected_accounts().await?; + let key_id = direct.auth_key_fingerprint(); + if let Some(error) = direct_auth::direct_auth_backoff_error(key_id) { + tracing::warn!( + "[composio-direct] list_connections: direct API key backoff gate open; \ + skipping v3 /connected_accounts" + ); + anyhow::bail!("{error}"); + } + + let items = match direct.list_connected_accounts().await { + Ok(items) => { + direct_auth::record_direct_auth_success(key_id); + items + } + Err(error) => { + let rendered = format!("{error:#}"); + match direct_auth::record_direct_auth_failure(key_id, &rendered) { + direct_auth::DirectAuthFailureDecision::NotAuthFailure => {} + direct_auth::DirectAuthFailureDecision::RetryAllowed { consecutive } => { + tracing::warn!( + consecutive, + threshold = direct_auth::DIRECT_INVALID_API_KEY_THRESHOLD, + "[composio-direct] list_connections: direct API key rejected" + ); + } + direct_auth::DirectAuthFailureDecision::CircuitOpened { consecutive } => { + let backoff = direct_auth::invalid_api_key_backoff_message(consecutive); + tracing::warn!( + consecutive, + threshold = direct_auth::DIRECT_INVALID_API_KEY_THRESHOLD, + "[composio-direct] list_connections: direct API key backoff gate opened" + ); + anyhow::bail!("{backoff}"); + } + } + return Err(error); + } + }; let connections: Vec = items .into_iter() .filter_map(|item| { diff --git a/src/openhuman/composio/client_tests.rs b/src/openhuman/composio/client_tests.rs index b98f3e1db..87cb6c2ac 100644 --- a/src/openhuman/composio/client_tests.rs +++ b/src/openhuman/composio/client_tests.rs @@ -1142,14 +1142,90 @@ fn direct_tool_for_test() -> std::sync::Arc std::sync::Arc { + direct_tool_for_mock_with_key(base_v3, "ck_test_direct") +} + +fn direct_tool_for_mock_with_key( + base_v3: String, + api_key: &str, +) -> std::sync::Arc { std::sync::Arc::new(crate::openhuman::tools::ComposioTool::new_with_v3_base( - "ck_test_direct", + api_key, Some("default"), std::sync::Arc::new(crate::openhuman::security::SecurityPolicy::default()), base_v3, )) } +struct DirectAuthFailureGuard { + key_id: u64, +} + +impl DirectAuthFailureGuard { + fn for_tool(tool: &std::sync::Arc) -> Self { + let key_id = tool.auth_key_fingerprint(); + crate::openhuman::composio::direct_auth::reset_direct_auth_failure(key_id); + Self { key_id } + } +} + +impl Drop for DirectAuthFailureGuard { + fn drop(&mut self) { + crate::openhuman::composio::direct_auth::reset_direct_auth_failure(self.key_id); + } +} + +#[tokio::test] +async fn direct_list_connections_stops_hitting_composio_after_repeated_invalid_api_key() { + let hits = Arc::new(AtomicUsize::new(0)); + let app = Router::new() + .route( + "/connected_accounts", + get(|State(hits): State>| async move { + hits.fetch_add(1, Ordering::SeqCst); + ( + StatusCode::UNAUTHORIZED, + Json(json!({ "error": { "message": "Invalid API key" } })), + ) + }), + ) + .with_state(hits.clone()); + let base = start_mock_backend(app).await; + let tool = direct_tool_for_mock_with_key(base, "ck_test_direct_invalid_backoff"); + let _auth_guard = DirectAuthFailureGuard::for_tool(&tool); + + for _ in 0..2 { + let err = direct_list_connections(&tool) + .await + .expect_err("invalid key should reject"); + assert!( + err.to_string().contains("Invalid API key"), + "unexpected error: {err:#}" + ); + } + + let opened = direct_list_connections(&tool) + .await + .expect_err("third invalid-key failure should open the backoff gate"); + assert!( + opened.to_string().contains("re-enter"), + "backoff error should be actionable, got: {opened:#}" + ); + + let short_circuit = direct_list_connections(&tool) + .await + .expect_err("open backoff gate should short-circuit before HTTP"); + assert!( + short_circuit.to_string().contains("re-enter"), + "short-circuit error should stay actionable, got: {short_circuit:#}" + ); + assert_eq!( + hits.load(Ordering::SeqCst), + 3, + "after three invalid-key failures, later polls must not hit Composio" + ); +} + #[tokio::test] async fn direct_authorize_rejects_empty_toolkit() { let tool = direct_tool_for_test(); diff --git a/src/openhuman/composio/direct_auth/mod.rs b/src/openhuman/composio/direct_auth/mod.rs new file mode 100644 index 000000000..c3d9230e9 --- /dev/null +++ b/src/openhuman/composio/direct_auth/mod.rs @@ -0,0 +1,125 @@ +//! Direct-mode Composio API-key health tracking. +//! +//! Invalid/revoked BYO keys used to make every 5s poll hit Composio v3 and +//! fail again. This module keeps a process-local consecutive-failure counter +//! keyed by a non-logged key fingerprint so repeated `401 Invalid API key` +//! responses open a short-circuit gate until the user re-enters the key. + +use std::collections::HashMap; +use std::hash::{Hash, Hasher}; +use std::sync::{LazyLock, Mutex, MutexGuard}; + +pub(crate) const DIRECT_INVALID_API_KEY_THRESHOLD: u32 = 3; + +static DIRECT_AUTH_FAILURES: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DirectAuthFailureDecision { + NotAuthFailure, + RetryAllowed { consecutive: u32 }, + CircuitOpened { consecutive: u32 }, +} + +fn failure_counts() -> MutexGuard<'static, HashMap> { + DIRECT_AUTH_FAILURES + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +pub(crate) fn fingerprint_api_key(api_key: &str) -> u64 { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + api_key.trim().hash(&mut hasher); + hasher.finish() +} + +pub(crate) fn is_invalid_api_key_error(message: &str) -> bool { + let lower = message.to_ascii_lowercase(); + lower.contains("invalid api key") + || (lower.contains("401") && lower.contains("api key") && lower.contains("invalid")) +} + +pub(crate) fn record_direct_auth_success(key_id: u64) { + failure_counts().remove(&key_id); +} + +pub(crate) fn reset_direct_auth_failure(key_id: u64) { + failure_counts().remove(&key_id); +} + +pub(crate) fn reset_all_direct_auth_failures() { + failure_counts().clear(); +} + +pub(crate) fn record_direct_auth_failure(key_id: u64, message: &str) -> DirectAuthFailureDecision { + if !is_invalid_api_key_error(message) { + reset_direct_auth_failure(key_id); + return DirectAuthFailureDecision::NotAuthFailure; + } + + let mut counts = failure_counts(); + let consecutive = counts.entry(key_id).or_insert(0); + *consecutive = consecutive.saturating_add(1); + if *consecutive >= DIRECT_INVALID_API_KEY_THRESHOLD { + DirectAuthFailureDecision::CircuitOpened { + consecutive: *consecutive, + } + } else { + DirectAuthFailureDecision::RetryAllowed { + consecutive: *consecutive, + } + } +} + +pub(crate) fn direct_auth_backoff_error(key_id: u64) -> Option { + let counts = failure_counts(); + let consecutive = counts.get(&key_id).copied().unwrap_or_default(); + (consecutive >= DIRECT_INVALID_API_KEY_THRESHOLD) + .then(|| invalid_api_key_backoff_message(consecutive)) +} + +pub(crate) fn invalid_api_key_backoff_message(consecutive: u32) -> String { + format!( + "Direct-mode Composio API key was rejected {consecutive} consecutive times with HTTP 401 Invalid API key; re-enter a valid key in Settings > Connections > Composio to resume polling." + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn generic_http_401_is_not_classified_as_invalid_api_key() { + assert!(!is_invalid_api_key_error("HTTP 401")); + assert!(is_invalid_api_key_error("HTTP 401: Invalid API key")); + } + + #[test] + fn non_auth_failure_resets_invalid_key_streak() { + let key_id = fingerprint_api_key("ck_test_streak_reset"); + reset_direct_auth_failure(key_id); + + assert_eq!( + record_direct_auth_failure(key_id, "HTTP 401: Invalid API key"), + DirectAuthFailureDecision::RetryAllowed { consecutive: 1 } + ); + assert_eq!( + record_direct_auth_failure(key_id, "HTTP 401: Invalid API key"), + DirectAuthFailureDecision::RetryAllowed { consecutive: 2 } + ); + assert_eq!( + record_direct_auth_failure(key_id, "HTTP 500: upstream unavailable"), + DirectAuthFailureDecision::NotAuthFailure + ); + assert!( + direct_auth_backoff_error(key_id).is_none(), + "non-auth failures must clear stale invalid-key counts" + ); + assert_eq!( + record_direct_auth_failure(key_id, "HTTP 401: Invalid API key"), + DirectAuthFailureDecision::RetryAllowed { consecutive: 1 } + ); + + reset_direct_auth_failure(key_id); + } +} diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index 4baee0e4f..746d75d7f 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -40,6 +40,7 @@ pub mod auth_retry; pub mod bus; pub mod client; mod connected_integrations; +pub(crate) mod direct_auth; pub mod error_mapping; pub mod execute_dispatch; pub mod execute_prepare; diff --git a/src/openhuman/composio/ops/direct_mode.rs b/src/openhuman/composio/ops/direct_mode.rs index d3f0eb8df..cd2c99325 100644 --- a/src/openhuman/composio/ops/direct_mode.rs +++ b/src/openhuman/composio/ops/direct_mode.rs @@ -3,8 +3,44 @@ use crate::openhuman::config::Config; use crate::rpc::RpcOutcome; +use super::super::{ + client::{create_direct_composio_tool_for_api_key, direct_list_connections}, + direct_auth, +}; use super::error_utils::OpResult; +async fn validate_direct_api_key_before_store(config: &Config, api_key: &str) -> OpResult<()> { + let key_id = direct_auth::fingerprint_api_key(api_key); + direct_auth::reset_direct_auth_failure(key_id); + + let direct = create_direct_composio_tool_for_api_key(config, api_key) + .map_err(|e| format!("[composio-direct] validate_api_key: {e}"))?; + + match direct_list_connections(&direct).await { + Ok(_) => { + direct_auth::record_direct_auth_success(key_id); + tracing::debug!("[composio-direct] validate_api_key: probe succeeded"); + Ok(()) + } + Err(error) => { + let rendered = format!("{error:#}"); + if direct_auth::is_invalid_api_key_error(&rendered) { + tracing::warn!( + "[composio-direct] validate_api_key: rejected invalid API key before storage" + ); + Err("Invalid Composio API key. Re-enter a valid key in Settings > Connections > Composio.".into()) + } else { + tracing::warn!( + error = %rendered, + "[composio-direct] validate_api_key: probe failed for a non-auth reason; \ + storing key so the user can retry when Composio is reachable" + ); + Ok(()) + } + } + } +} + /// Read the current Composio routing mode and whether a direct-mode API /// key is stored. **The key itself is never returned** — only a boolean /// flag so the UI can show a "Connected" / "Not set" status. @@ -49,6 +85,7 @@ pub async fn composio_set_api_key( activate_direct, "[composio-direct] set_api_key (redacted)" ); + validate_direct_api_key_before_store(config, trimmed).await?; crate::openhuman::credentials::store_composio_api_key(config, trimmed) .await @@ -110,6 +147,7 @@ pub async fn composio_clear_api_key(config: &Config) -> OpResult, +} + +impl EnvVarGuard { + fn set(key: &'static str, value: &str) -> Self { + let previous = std::env::var_os(key); + unsafe { + std::env::set_var(key, value); + } + Self { key, previous } + } +} + +impl Drop for EnvVarGuard { + fn drop(&mut self) { + match self.previous.take() { + Some(prev) => unsafe { + std::env::set_var(self.key, prev); + }, + None => unsafe { + std::env::remove_var(self.key); + }, + } + } +} + +struct DirectAuthFailureGuard { + key_id: u64, +} + +impl DirectAuthFailureGuard { + fn new(api_key: &str) -> Self { + let key_id = crate::openhuman::composio::direct_auth::fingerprint_api_key(api_key); + crate::openhuman::composio::direct_auth::reset_direct_auth_failure(key_id); + Self { key_id } + } +} + +impl Drop for DirectAuthFailureGuard { + fn drop(&mut self) { + crate::openhuman::composio::direct_auth::reset_direct_auth_failure(self.key_id); + } +} + async fn start_mock_backend(app: Router) -> String { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); @@ -1926,6 +1973,104 @@ async fn composio_list_connections_returns_empty_when_direct_mode_no_key() { ); } +#[tokio::test] +async fn composio_set_api_key_rejects_invalid_direct_key_before_persisting() { + use crate::openhuman::config::TEST_ENV_LOCK; + use crate::openhuman::credentials::get_composio_api_key; + + let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let app = Router::new().route( + "/connected_accounts", + get(|| async { + ( + axum::http::StatusCode::UNAUTHORIZED, + Json(json!({ "error": { "message": "Invalid API key" } })), + ) + }), + ); + let base = start_mock_backend(app).await; + let _base_v2 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &base); + let _base_v3 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &base); + + let tmp = tempfile::tempdir().unwrap(); + let config = direct_mode_no_key_config(&tmp); + let _auth_guard = DirectAuthFailureGuard::new("ck_invalid_direct"); + + let err = composio_set_api_key(&config, "ck_invalid_direct", false) + .await + .expect_err("invalid direct-mode key must be rejected before persistence"); + assert!( + err.contains("Invalid Composio API key"), + "unexpected error: {err}" + ); + assert_eq!( + get_composio_api_key(&config).expect("read composio key after failed save"), + None, + "invalid key must not be stored" + ); +} + +#[tokio::test] +async fn composio_set_api_key_validates_candidate_key_even_when_stored_key_exists() { + use crate::openhuman::config::TEST_ENV_LOCK; + use crate::openhuman::credentials::{get_composio_api_key, store_composio_api_key}; + use std::sync::{Arc, Mutex}; + + let _env_guard = TEST_ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let seen_keys = Arc::new(Mutex::new(Vec::::new())); + let app = Router::new() + .route( + "/connected_accounts", + get( + |State(seen_keys): State>>>, headers: HeaderMap| async move { + let key = headers + .get("x-api-key") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(); + seen_keys.lock().unwrap().push(key.clone()); + if key == "ck_old_valid" { + (axum::http::StatusCode::OK, Json(json!({ "items": [] }))) + } else { + ( + axum::http::StatusCode::UNAUTHORIZED, + Json(json!({ "error": { "message": "Invalid API key" } })), + ) + } + }, + ), + ) + .with_state(seen_keys.clone()); + let base = start_mock_backend(app).await; + let _base_v2 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &base); + let _base_v3 = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &base); + + let tmp = tempfile::tempdir().unwrap(); + let config = direct_mode_no_key_config(&tmp); + store_composio_api_key(&config, "ck_old_valid") + .await + .expect("seed old stored composio key"); + let _new_key_guard = DirectAuthFailureGuard::new("ck_new_invalid"); + + let err = composio_set_api_key(&config, "ck_new_invalid", false) + .await + .expect_err("candidate invalid key must be rejected even if old stored key is valid"); + assert!( + err.contains("Invalid Composio API key"), + "unexpected error: {err}" + ); + assert_eq!( + get_composio_api_key(&config).expect("read composio key after failed replacement"), + Some("ck_old_valid".to_string()), + "failed replacement must leave the old stored key intact" + ); + assert_eq!( + seen_keys.lock().unwrap().as_slice(), + ["ck_new_invalid"], + "validation must probe the candidate key, not the stored key" + ); +} + // ── Direct-mode authorize / list_tools / execute (commit 1, #1710) ─ /// Direct-mode `composio_list_tools` now hits Composio v3 with the diff --git a/src/openhuman/composio/tools/direct.rs b/src/openhuman/composio/tools/direct.rs index 4334deace..885ce09f3 100644 --- a/src/openhuman/composio/tools/direct.rs +++ b/src/openhuman/composio/tools/direct.rs @@ -90,6 +90,10 @@ impl ComposioTool { ) } + pub(crate) fn auth_key_fingerprint(&self) -> u64 { + crate::openhuman::composio::direct_auth::fingerprint_api_key(&self.api_key) + } + /// Debug-test seam for raw integration coverage: construct a direct /// Composio tool against explicit v2/v3 base URLs. Non-HTTPS URLs are /// accepted only for loopback hosts and only in debug builds. diff --git a/tests/composio_credentials_state_raw_coverage_e2e.rs b/tests/composio_credentials_state_raw_coverage_e2e.rs index fa8d43db5..a50e224c6 100644 --- a/tests/composio_credentials_state_raw_coverage_e2e.rs +++ b/tests/composio_credentials_state_raw_coverage_e2e.rs @@ -12,7 +12,7 @@ use axum::body::to_bytes; use axum::extract::{Request, State}; use axum::http::{Method, StatusCode}; use axum::response::{IntoResponse, Response}; -use axum::routing::any; +use axum::routing::{any, get}; use axum::{Json, Router}; use chrono::{Duration as ChronoDuration, Utc}; use serde_json::{json, Value}; @@ -388,6 +388,13 @@ async fn round15_composio_direct_key_mode_flips_without_network() { let _lock = env_lock(); let harness = setup("http://127.0.0.1:9"); let config = harness.config().await; + let direct_base = start_loopback_backend(Router::new().route( + "/connected_accounts", + get(composio_direct_connected_accounts), + )) + .await; + let _direct_v2_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &direct_base); + let _direct_v3_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &direct_base); let empty = composio_set_api_key(&config, " ", false) .await @@ -850,6 +857,10 @@ async fn composio_backend_handler(State(state): State, request: Reque } } +async fn composio_direct_connected_accounts() -> Json { + Json(json!({ "items": [] })) +} + async fn start_loopback_backend(app: Router) -> String { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await diff --git a/tests/composio_ops_raw_coverage_e2e.rs b/tests/composio_ops_raw_coverage_e2e.rs index 47be1513b..5ef7a2493 100644 --- a/tests/composio_ops_raw_coverage_e2e.rs +++ b/tests/composio_ops_raw_coverage_e2e.rs @@ -4,13 +4,13 @@ //! public ops layer instead of unit-test-only helpers so coverage lands on the //! RPC-facing paths used by controllers, tools, and prompt integration fetches. -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, OnceLock}; use axum::body::to_bytes; use axum::extract::{Request, State}; use axum::http::{Method, StatusCode}; use axum::response::{IntoResponse, Response}; -use axum::routing::any; +use axum::routing::{any, get}; use axum::{Json, Router}; use serde_json::Map; use serde_json::{json, Value}; @@ -50,6 +50,37 @@ struct RecordedRequest { body: Value, } +static COMPOSIO_OPS_ENV_LOCK: OnceLock> = OnceLock::new(); + +struct EnvGuard { + key: &'static str, + old: Option, +} + +impl EnvGuard { + fn set(key: &'static str, value: &str) -> Self { + let old = std::env::var(key).ok(); + std::env::set_var(key, value); + Self { key, old } + } +} + +impl Drop for EnvGuard { + fn drop(&mut self) { + match &self.old { + Some(value) => std::env::set_var(self.key, value), + None => std::env::remove_var(self.key), + } + } +} + +fn env_lock() -> std::sync::MutexGuard<'static, ()> { + COMPOSIO_OPS_ENV_LOCK + .get_or_init(|| Mutex::new(())) + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + #[tokio::test] async fn composio_ops_use_loopback_backend_for_happy_and_error_paths() { let state = MockState::default(); @@ -323,6 +354,15 @@ async fn composio_ops_use_loopback_backend_for_happy_and_error_paths() { #[tokio::test] async fn composio_direct_key_ops_and_agent_tools_take_local_validation_paths() { + let _lock = env_lock(); + let direct_base = start_loopback_backend(Router::new().route( + "/connected_accounts", + get(composio_direct_connected_accounts), + )) + .await; + let _direct_v2_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &direct_base); + let _direct_v3_guard = EnvGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &direct_base); + let dir = tempdir().expect("tempdir"); let mut config = Config { workspace_dir: dir.path().join("workspace"), @@ -727,6 +767,10 @@ async fn start_loopback_backend(app: Router) -> String { format!("http://127.0.0.1:{}", addr.port()) } +async fn composio_direct_connected_accounts() -> Json { + Json(json!({ "items": [] })) +} + fn store_app_session_token(config: &Config, token: &str) { AuthService::from_config(config) .store_provider_token( diff --git a/tests/worker_c_modules_e2e.rs b/tests/worker_c_modules_e2e.rs index 3e6d33d55..3e08c7069 100644 --- a/tests/worker_c_modules_e2e.rs +++ b/tests/worker_c_modules_e2e.rs @@ -5,11 +5,13 @@ //! an isolated workspace. It avoids live network calls. use std::path::Path; -use std::sync::{Mutex, OnceLock}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::Duration; +use axum::extract::State; use axum::http::header::{AUTHORIZATION, CONTENT_TYPE}; -use axum::{response::Html, routing::get, Router}; +use axum::{response::Html, routing::get, Json, Router}; use reqwest::StatusCode; use serde_json::{json, Value}; use tempfile::{tempdir, TempDir}; @@ -124,6 +126,8 @@ async fn setup() -> Harness { EnvVarGuard::unset("BACKEND_URL"), EnvVarGuard::unset("VITE_BACKEND_URL"), EnvVarGuard::unset("OPENHUMAN_API_URL"), + EnvVarGuard::unset("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2"), + EnvVarGuard::unset("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3"), EnvVarGuard::set("OPENHUMAN_KEYRING_BACKEND", "file"), EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_STRICT", "false"), EnvVarGuard::set("OPENHUMAN_MEMORY_EMBED_ENDPOINT", ""), @@ -362,6 +366,9 @@ async fn channels_remaining_controller_paths_validate_without_live_services() { async fn composio_direct_mode_api_key_and_static_catalogs_round_trip() { let _lock = env_lock(); let harness = setup().await; + let (composio_base, composio_hits, composio_join) = serve_composio_direct_fixtures().await; + let _composio_v2_guard = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V2", &composio_base); + let _composio_v3_guard = EnvVarGuard::set("OPENHUMAN_COMPOSIO_DIRECT_BASE_V3", &composio_base); let capabilities = rpc( &harness.rpc_base, @@ -430,6 +437,11 @@ async fn composio_direct_mode_api_key_and_static_catalogs_round_trip() { .and_then(Value::as_str), Some("direct") ); + assert_eq!( + composio_hits.load(Ordering::SeqCst), + 1, + "saving a direct-mode API key should validate the candidate key against the v3 mock" + ); let mode1 = rpc( &harness.rpc_base, @@ -480,6 +492,7 @@ async fn composio_direct_mode_api_key_and_static_catalogs_round_trip() { .and_then(Value::as_str), Some("backend") ); + composio_join.abort(); } #[tokio::test] @@ -1005,6 +1018,32 @@ async fn serve_source_fixtures() -> (String, tokio::task::JoinHandle ( + String, + Arc, + tokio::task::JoinHandle>, +) { + async fn connected_accounts(State(hits): State>) -> Json { + hits.fetch_add(1, Ordering::SeqCst); + Json(json!({ + "items": [] + })) + } + + let hits = Arc::new(AtomicUsize::new(0)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind composio fixture listener"); + let addr = listener + .local_addr() + .expect("composio fixture listener addr"); + let app = Router::new() + .route("/connected_accounts", get(connected_accounts)) + .with_state(hits.clone()); + let join = tokio::spawn(async move { axum::serve(listener, app).await }); + (format!("http://{addr}"), hits, join) +} + #[tokio::test] async fn memory_sources_folder_web_and_rss_readers_sync_through_rpc() { let _lock = env_lock();