From fe337267022a5be5026c29d2be37b0db9b3dc7f0 Mon Sep 17 00:00:00 2001 From: Steven Enamakel <31011319+senamakel@users.noreply.github.com> Date: Tue, 28 Jul 2026 11:13:37 +0300 Subject: [PATCH] refactor(core): cut managed backend traffic over to Rust SDK (#5233) --- AGENTS.md | 12 + Cargo.lock | 15 + Cargo.toml | 1 + src/api/rest.rs | 506 +++++++-------------- src/api/rest_tests.rs | 61 +++ src/openhuman/integrations/client.rs | 425 ++++++----------- src/openhuman/integrations/client_tests.rs | 60 +++ src/openhuman/webhooks/ops_tests.rs | 6 + 8 files changed, 470 insertions(+), 616 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 503409361..20997f2f2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -175,6 +175,18 @@ Embedded provider webviews **must not** grow new JS injection. No new `.js` unde ## Rust core (`src/`) +### TinyHumans backend SDK boundary + +- All Rust-core calls to the TinyHumans managed backend must go through + `tinyhumans-sdk`, with OpenHuman adapters retaining local egress, budget, + session-expiry, TLS, and observability policy. +- Do not add direct `reqwest` calls for TinyHumans JSON APIs or duplicate SDK + wire types in OpenHuman. Extend the SDK and update its pinned revision when a + public route or type is missing. +- Never expose or call TinyHumans admin or webhook APIs from the SDK boundary. + Local inbound webhook routing is an OpenHuman runtime feature and is not an + exception for backend `/webhooks/*` calls. + ### Domain layout (`src/openhuman/`) ~130 domain directories — authoritative list: `ls -d src/openhuman/*/`. Major families: agent (`agent`, `agent_experience`, `agent_meetings`, `agent_memory`, `agent_orchestration`, `agent_registry`, `agent_tool_policy`, `agentbox`, `orchestration`), memory (`memory`, `memory_archivist`, `memory_conversations`, `memory_diff`, `memory_goals`, `memory_queue`, `memory_search`, `memory_sources`, `memory_store`, `memory_sync`, `memory_tools`, `memory_tree`, `tinycortex`), skills/flows (`skills`, `skill_registry`, `skill_runtime`, `flows`, `tinyflows`, `tinyagents`, `rhai_workflows`), inference/AI (`inference`, `embeddings`, `routing`), MCP (`mcp_audit`, `mcp_client`, `mcp_registry`, `mcp_server`), runtimes (`runtime_node`, `runtime_python`, `runtime_python_server`, `javascript`, `sandbox`, `cwd_jail`), channels/webviews (`channels`, `whatsapp_data`), meet (`meet`, `meet_agent`), web3 (`wallet`, `web3`, `x402`, `tokenjuice`), plus platform domains (`about_app`, `approval`, `config`, `cron`, `credentials`, `keyring`, `security`, `threads`, `tools`, `update`, `voice`, …). diff --git a/Cargo.lock b/Cargo.lock index 665db11e3..47f586cb9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4750,6 +4750,7 @@ dependencies = [ "tinychannels", "tinycortex", "tinyflows", + "tinyhumans-sdk", "tinyjuice", "tinyplace", "tokio", @@ -7439,6 +7440,20 @@ dependencies = [ "tracing", ] +[[package]] +name = "tinyhumans-sdk" +version = "0.1.0" +source = "git+https://github.com/tinyhumansai/sdk.git?rev=3ee4123ba3b7a76c5f167d7bc2c72fca86671292#3ee4123ba3b7a76c5f167d7bc2c72fca86671292" +dependencies = [ + "base64 0.22.1", + "percent-encoding", + "reqwest 0.12.28", + "serde", + "serde_json", + "thiserror 2.0.18", + "url", +] + [[package]] name = "tinyjuice" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index e1be97602..0c2f503d6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -144,6 +144,7 @@ dhat = { version = "0.3", optional = true } # AV root CAs. So the two TLS backends coexist only on Windows, never on # the RAM-sensitive platforms. reqwest = { version = "0.12", default-features = false, features = ["json", "blocking", "rustls-tls", "stream", "http2", "multipart", "socks"] } +tinyhumans-sdk = { git = "https://github.com/tinyhumansai/sdk.git", rev = "3ee4123ba3b7a76c5f167d7bc2c72fca86671292" } # Already in-tree via reqwest/hyper; named directly so `IntegrationClient::get_bytes` # can return `bytes::Bytes` without a copy. bytes = "1" diff --git a/src/api/rest.rs b/src/api/rest.rs index ba8165a7c..66ce09ab1 100644 --- a/src/api/rest.rs +++ b/src/api/rest.rs @@ -2,13 +2,12 @@ use anyhow::{Context, Result}; use base64::Engine; -use reqwest::header::{HeaderMap, HeaderName, HeaderValue, AUTHORIZATION}; +use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use reqwest::{Client, Method, Url}; use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use std::time::Duration; - -use super::jwt::bearer_authorization_value; +use tinyhumans_sdk::{Error as SdkError, TinyHumansClient}; /// Typed errors surfaced by `authed_json` for expected backend states that /// callers should recover from in-flow rather than funnel into Sentry. @@ -225,37 +224,6 @@ fn build_backend_reqwest_client() -> Result { .map_err(|e| anyhow::anyhow!("failed to build HTTP client: {e}")) } -fn parse_api_response_json(text: &str) -> Result { - let v: Value = serde_json::from_str(text).with_context(|| format!("parse API JSON: {text}"))?; - let Some(obj) = v.as_object() else { - return Ok(v); - }; - if let Some(success) = obj.get("success").and_then(|x| x.as_bool()) { - if !success { - let msg = obj - .get("message") - .or_else(|| obj.get("error")) - .and_then(|x| x.as_str()) - .unwrap_or("request unsuccessful"); - anyhow::bail!("API request failed: {msg}"); - } - if let Some(data) = obj.get("data") { - if !data.is_null() { - return Ok(data.clone()); - } - } - if let Some(user) = obj.get("user") { - if !user.is_null() { - return Ok(user.clone()); - } - } - let mut m = obj.clone(); - m.remove("success"); - return Ok(Value::Object(m)); - } - Ok(v) -} - fn user_id_from_object(obj: &serde_json::Map) -> Option { for key in ["id", "_id", "userId"] { if let Some(s) = obj.get(key).and_then(|x| x.as_str()) { @@ -303,27 +271,6 @@ pub struct ConnectResponse { pub state: String, } -#[derive(Debug, Clone, Deserialize)] -struct ConnectEnvelope { - success: bool, - #[serde(default, alias = "oauthUrl")] - oauth_url: Option, - #[serde(default)] - state: Option, -} - -#[derive(Debug, Clone, Deserialize)] -struct IntegrationsEnvelope { - success: bool, - data: IntegrationsData, -} - -#[derive(Debug, Clone, Deserialize)] -#[serde(rename_all = "camelCase")] -struct IntegrationsData { - integrations: Vec, -} - /// A summary of an active integration, as returned by the backend. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -336,28 +283,6 @@ pub struct IntegrationSummary { pub created_at: String, } -#[derive(Debug, Clone, Deserialize)] -struct TokensEnvelope { - success: bool, - data: TokensData, -} - -#[derive(Debug, Clone, Deserialize)] -struct TokensData { - encrypted: String, -} - -#[derive(Debug, Clone, Deserialize)] -struct LoginTokenConsumeEnvelope { - success: bool, - data: LoginTokenConsumeData, -} - -#[derive(Debug, Clone, Deserialize)] -struct LoginTokenConsumeData { - jwt: String, -} - /// Decrypted OAuth token payload for handing off tokens to a local service or skill. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -376,6 +301,7 @@ pub struct IntegrationTokensHandoff { pub struct BackendOAuthClient { client: Client, base: Url, + sdk: TinyHumansClient, } impl BackendOAuthClient { @@ -397,7 +323,8 @@ impl BackendOAuthClient { base.set_query(None); base.set_fragment(None); let client = build_backend_reqwest_client()?; - Ok(Self { client, base }) + let sdk = TinyHumansClient::new(base.as_str()).with_http_client(client.clone()); + Ok(Self { client, base, sdk }) } /// Borrow the underlying `reqwest::Client` for callers that need to @@ -436,67 +363,48 @@ impl BackendOAuthClient { ) -> Result { let p = provider.trim().trim_matches('/'); anyhow::ensure!(!p.is_empty(), "provider is required"); - let mut url = self - .base - .join(&format!("auth/{p}/connect")) - .context("build connect URL")?; - if let Some(s) = skill_id.filter(|s| !s.is_empty()) { - url.query_pairs_mut().append_pair("skillId", s); - } - if let Some(r) = response_type.filter(|r| !r.is_empty()) { - url.query_pairs_mut().append_pair("responseType", r); - } - if let Some(e) = encryption_mode.filter(|e| !e.is_empty()) { - url.query_pairs_mut().append_pair("encryptionMode", e); - } - - let resp = self - .client - .get(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .send() + let query = { + let mut serializer = url::form_urlencoded::Serializer::new(String::new()); + if let Some(s) = skill_id.filter(|s| !s.is_empty()) { + serializer.append_pair("skillId", s); + } + if let Some(r) = response_type.filter(|r| !r.is_empty()) { + serializer.append_pair("responseType", r); + } + if let Some(e) = encryption_mode.filter(|e| !e.is_empty()) { + serializer.append_pair("encryptionMode", e); + } + serializer.finish() + }; + let path = if query.is_empty() { + format!("auth/{p}/connect") + } else { + format!("auth/{p}/connect?{query}") + }; + let value = self + .authed_json(bearer_jwt, Method::GET, &path, None) .await .context("auth connect request")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("auth connect failed ({status}): {text}"); - } - - let env: ConnectEnvelope = - serde_json::from_str(&text).with_context(|| format!("parse connect JSON: {text}"))?; - if !env.success { - anyhow::bail!("auth connect unsuccessful: {text}"); - } - let oauth_url = env - .oauth_url - .filter(|u| !u.is_empty()) + let oauth_url = value + .get("oauthUrl") + .or_else(|| value.get("oauth_url")) + .and_then(Value::as_str) + .filter(|url| !url.is_empty()) + .map(str::to_owned) .context("missing oauthUrl in response")?; - let state = env - .state - .filter(|s| !s.is_empty()) + let state = value + .get("state") + .and_then(Value::as_str) + .filter(|state| !state.is_empty()) + .map(str::to_owned) .context("missing state")?; Ok(ConnectResponse { oauth_url, state }) } /// Fetches the current authenticated user profile using the provided JWT. pub async fn fetch_current_user(&self, bearer_jwt: &str) -> Result { - let url = self.base.join("auth/me").context("build /auth/me URL")?; - let resp = self - .client - .get(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .send() + self.authed_json(bearer_jwt, Method::GET, "auth/me", None) .await - .context("GET /auth/me")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("GET /auth/me failed ({status}): {text}"); - } - parse_api_response_json(&text) } /// Exchanges a one-time login token (e.g. from Telegram) for a long-lived JWT. @@ -510,32 +418,20 @@ impl BackendOAuthClient { // `telegram/login-tokens/{token}/consume` path-param route was removed, so // the old call 404'd and Telegram/OAuth-token login could never complete // (WIRING_GAPS_AUDIT C1/C2). - let url = self - .base - .join("auth/login-token/consume") - .context("build login-token consume URL")?; - - let resp = self - .client - .post(url) - .json(&serde_json::json!({ "token": token })) - .send() + let response = self + .sdk + .auth() + .consume_login_token(&tinyhumans_sdk::api::types::LoginTokenRequest { + token: token.to_string(), + }) .await - .context("consume login token")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("consume login token failed ({status}): {text}"); - } - - let env: LoginTokenConsumeEnvelope = serde_json::from_str(&text) - .with_context(|| format!("parse consume-login-token JSON: {text}"))?; - if !env.success { - anyhow::bail!("consume login token unsuccessful: {text}"); - } - - let jwt = env.data.jwt.trim().to_string(); + .context("consume login token through TinyHumans SDK")?; + let jwt = response + .get("jwt") + .and_then(Value::as_str) + .unwrap_or_default() + .trim() + .to_string(); anyhow::ensure!(!jwt.is_empty(), "consume login token response missing jwt"); Ok(jwt) } @@ -556,26 +452,13 @@ impl BackendOAuthClient { anyhow::ensure!(!channel.is_empty(), "channel is required"); let encoded_channel = urlencoding::encode(channel); - let url = self - .base - .join(&format!("auth/channels/{encoded_channel}/link-token")) - .context("build channel link-token URL")?; - - let resp = self - .client - .post(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .send() - .await - .context("create channel link token")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("create channel link token failed ({status}): {text}"); - } - - parse_api_response_json(&text) + self.authed_json( + bearer_jwt, + Method::POST, + &format!("auth/channels/{encoded_channel}/link-token"), + None, + ) + .await } /// Generic authenticated JSON request helper for backend API routes. @@ -586,68 +469,89 @@ impl BackendOAuthClient { path: &str, body: Option, ) -> Result { - let url = self - .base - .join(path.trim_start_matches('/')) - .with_context(|| format!("build URL for {path}"))?; - - let mut request = self - .client - .request(method.clone(), url.clone()) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)); - - if let Some(body) = body { - request = request.json(&body); - } - - let response = request.send().await.map_err(|e| { - // Walk the error source chain so transient markers hidden in nested - // causes (reqwest -> hyper -> rustls TLS EOF, etc.) still classify - // correctly. The top-level `e.to_string()` often only carries the - // outermost wrapper, e.g. "error sending request for url (...)". - let mut error_message = e.to_string(); - let mut src: Option<&(dyn std::error::Error + 'static)> = std::error::Error::source(&e); - while let Some(s) = src { - error_message.push_str(" → "); - error_message.push_str(&s.to_string()); - src = s.source(); - } - if crate::core::observability::contains_transient_transport_phrase(&error_message) { - tracing::warn!( - domain = "backend_api", - operation = "authed_json", - method = method.as_str(), - path = url.path(), - failure = "transport", - error = %error_message, - "[backend_api] transient transport failure on {} {}: {}", + // OpenHuman does not consume backend webhook APIs. Keep that boundary + // local even though the shared SDK intentionally exposes the + // user-owned `/webhooks/core` tunnel CRUD surface for other clients. + // Platform-admin operations are independently rejected by the SDK's + // generated route gate below. + anyhow::ensure!( + !path + .split(['/', '?']) + .any(|segment| segment.eq_ignore_ascii_case("webhooks")), + "backend webhook routes are not exposed through OpenHuman" + ); + let url = self.url_for(path)?; + let sdk = self + .sdk + .clone() + .with_token(Some(bearer_jwt.trim().to_string())); + let response = sdk + .raw() + .send(method.clone(), path, &[], body.as_ref(), true) + .await; + let value = match response { + Ok(value) => return Ok(value), + Err(SdkError::Http(e)) => { + // Walk the error source chain so transient markers hidden in nested + // causes (reqwest -> hyper -> rustls TLS EOF, etc.) still classify + // correctly. The top-level `e.to_string()` often only carries the + // outermost wrapper, e.g. "error sending request for url (...)". + let mut error_message = e.to_string(); + let mut src: Option<&(dyn std::error::Error + 'static)> = + std::error::Error::source(&e); + while let Some(s) = src { + error_message.push_str(" → "); + error_message.push_str(&s.to_string()); + src = s.source(); + } + if crate::core::observability::contains_transient_transport_phrase(&error_message) { + tracing::warn!( + domain = "backend_api", + operation = "authed_json", + method = method.as_str(), + path = url.path(), + failure = "transport", + error = %error_message, + "[backend_api] transient transport failure on {} {}: {}", + method.as_str(), + url.path(), + error_message, + ); + } else { + crate::core::observability::report_error( + error_message.as_str(), + "backend_api", + "authed_json", + &[ + ("method", method.as_str()), + ("path", url.path()), + ("failure", "transport"), + ], + ); + } + return Err(anyhow::Error::new(e).context(format!( + "backend request {} {}", method.as_str(), - url.path(), - error_message, - ); - } else { - crate::core::observability::report_error( - error_message.as_str(), - "backend_api", - "authed_json", - &[ - ("method", method.as_str()), - ("path", url.path()), - ("failure", "transport"), - ], - ); + url.path() + ))); } - anyhow::Error::new(e).context(format!( - "backend request {} {}", - method.as_str(), - url.path() - )) - })?; - - let status = response.status(); - let text = response.text().await.unwrap_or_default(); - if !status.is_success() { - let status_code = status.as_u16(); + Err(SdkError::Status { status, body }) => (status, body), + Err(error) => { + return Err(anyhow::Error::new(error).context(format!( + "backend request {} {}", + method.as_str(), + url.path() + ))); + } + }; + { + let (status_code, response_body) = value; + let status = reqwest::StatusCode::from_u16(status_code) + .context("SDK returned an invalid HTTP status")?; + let text = match response_body { + Value::String(text) => text, + other => serde_json::to_string(&other).unwrap_or_default(), + }; let status_str = status_code.to_string(); // 401 on any authed backend endpoint is an expected user-session @@ -810,35 +714,18 @@ impl BackendOAuthClient { url.path() ); } - - parse_api_response_json(&text) } /// Lists all active integrations for the current user. pub async fn list_integrations(&self, bearer_jwt: &str) -> Result> { - let url = self - .base - .join("auth/integrations") - .context("build integrations URL")?; - let resp = self - .client - .get(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .send() - .await - .context("list integrations")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("list integrations failed ({status}): {text}"); - } - let env: IntegrationsEnvelope = serde_json::from_str(&text) - .with_context(|| format!("parse integrations JSON: {text}"))?; - if !env.success { - anyhow::bail!("list integrations unsuccessful: {text}"); - } - Ok(env.data.integrations) + let value = self + .authed_json(bearer_jwt, Method::GET, "auth/integrations", None) + .await?; + let integrations = value + .get("integrations") + .cloned() + .unwrap_or_else(|| value.clone()); + serde_json::from_value(integrations).context("parse integrations response") } /// Fetches the decrypted OAuth tokens for a specific integration. @@ -856,31 +743,21 @@ impl BackendOAuthClient { !id.is_empty() && id.len() == 24, "integrationId must be a 24-char hex id" ); - let url = self - .base - .join(&format!("auth/integrations/{id}/tokens")) - .context("build tokens URL")?; let body = serde_json::json!({ "key": encryption_key.trim() }); - let resp = self - .client - .post(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .json(&body) - .send() + let value = self + .authed_json( + bearer_jwt, + Method::POST, + &format!("auth/integrations/{id}/tokens"), + Some(body), + ) .await .context("integration tokens handoff")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("integration tokens failed ({status}): {text}"); - } - let env: TokensEnvelope = - serde_json::from_str(&text).with_context(|| format!("parse tokens JSON: {text}"))?; - if !env.success { - anyhow::bail!("integration tokens unsuccessful: {text}"); - } - let plaintext = decrypt_handoff_blob(&env.data.encrypted, encryption_key.trim())?; + let encrypted = value + .get("encrypted") + .and_then(Value::as_str) + .context("integration tokens response missing encrypted payload")?; + let plaintext = decrypt_handoff_blob(encrypted, encryption_key.trim())?; serde_json::from_str(&plaintext).context("parse decrypted token JSON") } @@ -894,42 +771,19 @@ impl BackendOAuthClient { !id.is_empty() && id.len() == 24, "integrationId must be a 24-char hex id" ); - let url = self - .base - .join(&format!("auth/integrations/{id}/client-key")) - .context("build client-key URL")?; - let resp = self - .client - .post(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .send() + let value = self + .authed_json( + bearer_jwt, + Method::POST, + &format!("auth/integrations/{id}/client-key"), + None, + ) .await .context("fetch client key")?; - - let status = resp.status(); - let text = resp.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("fetch client key failed ({status}): {text}"); - } - let v: Value = serde_json::from_str(&text) - .with_context(|| format!("parse client-key JSON: {text}"))?; - let obj = v.as_object().context("expected JSON object")?; - let success = obj - .get("success") - .and_then(|s| s.as_bool()) - .unwrap_or(false); - if !success { - let msg = obj - .get("error") - .and_then(|e| e.as_str()) - .unwrap_or("client key retrieval unsuccessful"); - anyhow::bail!("fetch client key failed: {msg}"); - } - let client_key = obj - .get("data") - .and_then(|d| d.get("clientKey")) + let client_key = value + .get("clientKey") .and_then(|k| k.as_str()) - .context("missing data.clientKey in response")?; + .context("missing clientKey in response")?; Ok(client_key.to_string()) } @@ -1120,23 +974,13 @@ impl BackendOAuthClient { pub async fn revoke_integration(&self, integration_id: &str, bearer_jwt: &str) -> Result<()> { let id = integration_id.trim(); anyhow::ensure!(!id.is_empty(), "integration id is required"); - let url = self - .base - .join(&format!("auth/integrations/{id}")) - .context("build revoke URL")?; - let resp = self - .client - .delete(url) - .header(AUTHORIZATION, bearer_authorization_value(bearer_jwt)) - .send() - .await - .context("revoke integration")?; - - let status = resp.status(); - if !status.is_success() { - let text = resp.text().await.unwrap_or_default(); - anyhow::bail!("revoke integration failed ({status}): {text}"); - } + self.authed_json( + bearer_jwt, + Method::DELETE, + &format!("auth/integrations/{id}"), + None, + ) + .await?; Ok(()) } } diff --git a/src/api/rest_tests.rs b/src/api/rest_tests.rs index 3a968afce..1338344d6 100644 --- a/src/api/rest_tests.rs +++ b/src/api/rest_tests.rs @@ -215,6 +215,67 @@ async fn backend_client_sends_x_core_version_on_auth_requests() { version, sanitize_client_version(env!("CARGO_PKG_VERSION")).unwrap() ); + assert_eq!( + request_headers + .get("x-sdk-client") + .and_then(|value| value.to_str().ok()), + Some("tinyhumans-rust"), + "typed auth requests must be sent by the TinyHumans SDK transport" + ); +} + +#[tokio::test] +async fn authed_json_uses_sdk_transport_with_bearer_and_host_headers() { + let (base_url, captured) = spawn_header_capture_server().await; + let client = BackendOAuthClient::new(&base_url).unwrap(); + + let response = client + .authed_json("sdk-cutover-token", Method::GET, "/probe", None) + .await + .unwrap(); + assert_eq!(response, json!({ "ok": true })); + + let headers = captured.take(); + let request_headers = headers.last().unwrap(); + assert_eq!( + request_headers + .get("authorization") + .and_then(|value| value.to_str().ok()), + Some("Bearer sdk-cutover-token") + ); + assert_eq!( + request_headers + .get("x-sdk-client") + .and_then(|value| value.to_str().ok()), + Some("tinyhumans-rust") + ); + assert!( + request_headers.get("x-core-version").is_some(), + "OpenHuman host metadata must survive the SDK cutover" + ); +} + +#[tokio::test] +async fn authed_json_cannot_bypass_sdk_admin_or_webhook_exclusions() { + let client = BackendOAuthClient::new("http://127.0.0.1:9").unwrap(); + + for (method, path) in [ + (Method::POST, "/admin/announcements"), + (Method::GET, "/webhooks/core"), + ] { + let err = client + .authed_json("token", method, path, None) + .await + .unwrap_err(); + assert!( + err.chain().any(|source| { + let message = source.to_string(); + message.contains("intentionally not exposed") + || message.contains("webhook routes are not exposed") + }), + "{path} must be rejected locally by the SDK: {err:#}" + ); + } } #[tokio::test] diff --git a/src/openhuman/integrations/client.rs b/src/openhuman/integrations/client.rs index b3dd0a0a1..8818e2a11 100644 --- a/src/openhuman/integrations/client.rs +++ b/src/openhuman/integrations/client.rs @@ -4,6 +4,7 @@ use super::types::{BackendResponse, IntegrationPricing}; use std::error::Error as _; use std::sync::Arc; use std::time::Duration; +use tinyhumans_sdk::{Error as SdkError, TinyHumansClient}; /// Maximum length (in bytes) of backend error body included in propagated /// errors. Keep this bounded — error messages flow through tracing/Sentry and @@ -41,6 +42,21 @@ fn managed_budget_applies_to_path(path: &str) -> bool { path != "/agent-integrations/pricing" && path.starts_with("/agent-integrations/") } +fn reject_backend_webhook_path(method: &str, path: &str) -> anyhow::Result<()> { + let route = path.split('?').next().unwrap_or(path); + if route + .split('/') + .any(|segment| segment.eq_ignore_ascii_case("webhooks")) + { + anyhow::bail!( + "route is intentionally not exposed by the SDK: {} {}", + method, + route + ); + } + Ok(()) +} + /// Handle a `401 Unauthorized` from the OpenHuman backend's /// `/agent-integrations/*` routes. /// @@ -272,7 +288,11 @@ pub struct IntegrationClient { pub backend_url: String, pub auth_token: String, budget_config: Option>, - http_client: reqwest::Client, + sdk: TinyHumansClient, + // Temporary compatibility exception: the SDK's binary primitive returns + // bytes only, while file storage also consumes Content-Type and + // Content-Disposition. Remove when the SDK exposes response metadata. + download_client: reqwest::Client, pricing: tokio::sync::OnceCell, } @@ -316,12 +336,22 @@ impl IntegrationClient { .connect_timeout(Duration::from_secs(15)) .build() .expect("failed to build integration HTTP client"); + let sdk = TinyHumansClient::new(&backend_url) + .with_token(Some(auth_token.clone())) + .with_http_client(http_client.clone()); + let download_client = crate::openhuman::tls::tls_client_builder() + .http1_only() + .timeout(Duration::from_secs(15 * 60)) + .connect_timeout(Duration::from_secs(15)) + .build() + .expect("failed to build integration download HTTP client"); Self { backend_url, auth_token, budget_config, - http_client, + sdk, + download_client, pricing: tokio::sync::OnceCell::new(), } } @@ -346,194 +376,13 @@ impl IntegrationClient { path: &str, body: &serde_json::Value, ) -> anyhow::Result { - enforce_backend_egress(path)?; - emit_backend_egress(path); - self.ensure_budget_available(path).await?; - let url = crate::api::config::api_url(&self.backend_url, path); - tracing::debug!("[integrations] POST {}", url); - - let resp = self - .http_client - .post(&url) - .header("Authorization", format!("Bearer {}", self.auth_token)) - .header("Content-Type", "application/json") - .json(body) - .send() + self.request_json(reqwest::Method::POST, path, Some(body)) .await - .map_err(|e| { - // Log the full error source chain so the caller gets - // something useful instead of reqwest's top-level - // "error sending request for url (…)" which hides the - // real cause (DNS / TLS / connect / timeout). - let mut chain = format!("{e}"); - let mut src: Option<&(dyn std::error::Error + 'static)> = e.source(); - while let Some(s) = src { - chain.push_str(" → "); - chain.push_str(&s.to_string()); - src = s.source(); - } - // Use `report_error_or_expected` so transport-level shapes - // ("error sending request for url", "tls handshake eof", - // "connection refused/reset", …) are classified as - // `NetworkUnreachable` and skip Sentry — user-environment - // problems (VPN drop, captive portal, ISP block, TLS MITM) - // that no retry on our side can resolve (OPENHUMAN-TAURI-2G). - crate::core::observability::report_error_or_expected( - chain.as_str(), - "integrations", - "post", - &[("path", path), ("failure", "transport")], - ); - anyhow::anyhow!("POST {} failed: {}", url, chain) - })?; - - let status = resp.status(); - if !status.is_success() { - let body_text = resp.text().await.unwrap_or_default(); - let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); - let status_str = status.as_u16().to_string(); - // A 401 is the backend's auth middleware rejecting our app-session - // JWT (not a third-party provider 401 — see - // `handle_session_jwt_unauthorized` for the full - // session-JWT-vs-provider argument). Route it into the - // session-expiry recovery flow: demote from Sentry AND publish - // `SessionExpired` so re-login fires even though the autonomous - // agent loop swallows the propagated error (TAURI-RUST-84E). This - // must come BEFORE the generic non-2xx branch so the 401 doesn't - // fall through to a plain `BackendUserError` that only demotes. - if status == reqwest::StatusCode::UNAUTHORIZED { - let message = handle_session_jwt_unauthorized("POST", path, &url, &detail); - anyhow::bail!("{message}"); - } - // Route through `report_error_or_expected` so 4xx user-input / - // auth-state failures (e.g. OPENHUMAN-TAURI-BC: SharePoint - // authorize 400 because the user didn't fill in the required - // Tenant Name field) demote to a warn breadcrumb instead of - // firing a Sentry event. 5xx and non-transient 4xx still - // surface — see `is_backend_user_error_message` for the exact - // status set classified as expected. - crate::core::observability::report_error_or_expected( - format!("Backend returned {status} for POST {url}: {detail}").as_str(), - "integrations", - "post", - &[ - ("path", path), - ("status", status_str.as_str()), - ("failure", "non_2xx"), - ], - ); - anyhow::bail!("Backend returned {status} for POST {url}: {detail}"); - } - - let envelope: BackendResponse = resp.json().await?; - if !envelope.success { - let msg = envelope - .error - .unwrap_or_else(|| "unknown backend error".into()); - // Route through `report_error_or_expected` so user-state envelope - // failures the backend wraps as 2xx + `success: false` (composio - // "Toolkit X is not enabled", "Trigger type … not found", - // "Missing required fields: …" — OPENHUMAN-TAURI-3R / -3S / -34 / - // -97) demote to an info breadcrumb instead of firing a Sentry - // event. Genuine backend bugs (unknown envelope shapes, internal - // panics) still surface. - crate::core::observability::report_error_or_expected( - msg.as_str(), - "integrations", - "post", - &[("path", path), ("failure", "envelope_error")], - ); - anyhow::bail!("Backend error for POST {}: {}", url, msg); - } - envelope - .data - .ok_or_else(|| anyhow::anyhow!("Backend returned success but no data for POST {}", url)) } /// GET from a backend endpoint and parse the response `data` field. pub async fn get(&self, path: &str) -> anyhow::Result { - enforce_backend_egress(path)?; - emit_backend_egress(path); - self.ensure_budget_available(path).await?; - let url = crate::api::config::api_url(&self.backend_url, path); - tracing::debug!("[integrations] GET {}", url); - - let resp = self - .http_client - .get(&url) - .header("Authorization", format!("Bearer {}", self.auth_token)) - .send() - .await - .map_err(|e| { - let mut chain = format!("{e}"); - let mut src: Option<&(dyn std::error::Error + 'static)> = e.source(); - while let Some(s) = src { - chain.push_str(" → "); - chain.push_str(&s.to_string()); - src = s.source(); - } - // Mirrors the post() transport site — classify reqwest - // transport-level failures as NetworkUnreachable so they - // skip Sentry. OPENHUMAN-TAURI-2G: TLS handshake EOF - // against api.tinyhumans.ai from a SG user. - crate::core::observability::report_error_or_expected( - chain.as_str(), - "integrations", - "get", - &[("path", path), ("failure", "transport")], - ); - anyhow::anyhow!("GET {} failed: {}", url, chain) - })?; - - let status = resp.status(); - if !status.is_success() { - let body_text = resp.text().await.unwrap_or_default(); - let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); - let status_str = status.as_u16().to_string(); - // Mirrors the post() session-JWT 401 arm — a 401 here is the - // backend rejecting our session JWT, so drive re-login (publish - // SessionExpired) and demote from Sentry, before the generic - // non-2xx branch. See `handle_session_jwt_unauthorized`. - if status == reqwest::StatusCode::UNAUTHORIZED { - let message = handle_session_jwt_unauthorized("GET", path, &url, &detail); - anyhow::bail!("{message}"); - } - // Mirrors the post() site — see OPENHUMAN-TAURI-BC. 4xx - // user-input / auth-state shapes demote to a warn breadcrumb - // via the observability classifier; 5xx and non-transient 4xx - // still surface. - crate::core::observability::report_error_or_expected( - format!("Backend returned {status} for GET {url}: {detail}").as_str(), - "integrations", - "get", - &[ - ("path", path), - ("status", status_str.as_str()), - ("failure", "non_2xx"), - ], - ); - anyhow::bail!("Backend returned {status} for GET {url}: {detail}"); - } - - let envelope: BackendResponse = resp.json().await?; - if !envelope.success { - let msg = envelope - .error - .unwrap_or_else(|| "unknown backend error".into()); - // Mirrors the post() envelope-error site — see the comment there - // for OPENHUMAN-TAURI-3R/-3S/-34/-97 rationale. User-state - // envelope failures demote; genuine backend bugs still surface. - crate::core::observability::report_error_or_expected( - msg.as_str(), - "integrations", - "get", - &[("path", path), ("failure", "envelope_error")], - ); - anyhow::bail!("Backend error for GET {}: {}", url, msg); - } - envelope - .data - .ok_or_else(|| anyhow::anyhow!("Backend returned success but no data for GET {}", url)) + self.request_json(reqwest::Method::GET, path, None).await } /// Render a reqwest transport error with its full source chain (mirrors @@ -562,43 +411,53 @@ impl IntegrationClient { anyhow::anyhow!("{} {} failed: {}", method.to_uppercase(), url, chain) } - /// Shared non-2xx / 401 / envelope handling for the newer HTTP verbs - /// (`patch`, `delete`, `upload_multipart`). Mirrors the [`Self::post`] - /// error classification: 401 → session-expiry recovery flow, other - /// non-2xx → demoted/classified generic error, then `BackendResponse` - /// envelope parsing with `success:false` classification. - async fn parse_json_response( + fn map_sdk_error(error: SdkError, method: &str, path: &str, url: &str) -> anyhow::Error { + let method_upper = method.to_uppercase(); + match error { + SdkError::Http(error) => Self::report_transport_error(error, method, path, url), + SdkError::Status { status, body } => { + let body_text = match body { + serde_json::Value::String(text) => text, + value => value.to_string(), + }; + let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); + if status == reqwest::StatusCode::UNAUTHORIZED.as_u16() { + return anyhow::anyhow!(handle_session_jwt_unauthorized( + &method_upper, + path, + url, + &detail + )); + } + let status_text = reqwest::StatusCode::from_u16(status) + .map(|status| status.to_string()) + .unwrap_or_else(|_| status.to_string()); + let status_code = status.to_string(); + crate::core::observability::report_error_or_expected( + format!("Backend returned {status_text} for {method_upper} {url}: {detail}") + .as_str(), + "integrations", + method, + &[ + ("path", path), + ("status", status_code.as_str()), + ("failure", "non_2xx"), + ], + ); + anyhow::anyhow!("Backend returned {status_text} for {method_upper} {url}: {detail}") + } + other => anyhow::anyhow!("{method_upper} {url} failed: {other}"), + } + } + + fn parse_envelope( method: &str, path: &str, url: &str, - resp: reqwest::Response, + value: serde_json::Value, ) -> anyhow::Result { - let status = resp.status(); let method_upper = method.to_uppercase(); - if !status.is_success() { - let body_text = resp.text().await.unwrap_or_default(); - let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); - let status_str = status.as_u16().to_string(); - // Session-JWT rejection — same argument as in post()/get(): the - // backend auth middleware is the only 401 source on these routes. - if status == reqwest::StatusCode::UNAUTHORIZED { - let message = handle_session_jwt_unauthorized(&method_upper, path, url, &detail); - anyhow::bail!("{message}"); - } - crate::core::observability::report_error_or_expected( - format!("Backend returned {status} for {method_upper} {url}: {detail}").as_str(), - "integrations", - method, - &[ - ("path", path), - ("status", status_str.as_str()), - ("failure", "non_2xx"), - ], - ); - anyhow::bail!("Backend returned {status} for {method_upper} {url}: {detail}"); - } - - let envelope: BackendResponse = resp.json().await?; + let envelope: BackendResponse = serde_json::from_value(value)?; if !envelope.success { let msg = envelope .error @@ -620,6 +479,28 @@ impl IntegrationClient { }) } + async fn request_json( + &self, + method: reqwest::Method, + path: &str, + body: Option<&serde_json::Value>, + ) -> anyhow::Result { + reject_backend_webhook_path(method.as_str(), path)?; + enforce_backend_egress(path)?; + emit_backend_egress(path); + self.ensure_budget_available(path).await?; + let url = crate::api::config::api_url(&self.backend_url, path); + let method_name = method.as_str().to_ascii_lowercase(); + tracing::debug!("[integrations] {} {}", method.as_str(), url); + let value = self + .sdk + .raw() + .send(method, path, &[], body, false) + .await + .map_err(|error| Self::map_sdk_error(error, &method_name, path, &url))?; + Self::parse_envelope(&method_name, path, &url, value) + } + /// PATCH JSON to a backend endpoint and parse the response `data` field. /// Mirrors [`Self::post`] (auth header, 401 → session-expiry, error /// classification). @@ -628,44 +509,15 @@ impl IntegrationClient { path: &str, body: &serde_json::Value, ) -> anyhow::Result { - enforce_backend_egress(path)?; - emit_backend_egress(path); - self.ensure_budget_available(path).await?; - let url = crate::api::config::api_url(&self.backend_url, path); - tracing::debug!("[integrations] PATCH {}", url); - - let resp = self - .http_client - .patch(&url) - .header("Authorization", format!("Bearer {}", self.auth_token)) - .header("Content-Type", "application/json") - .json(body) - .send() + self.request_json(reqwest::Method::PATCH, path, Some(body)) .await - .map_err(|e| Self::report_transport_error(e, "patch", path, &url))?; - - Self::parse_json_response("patch", path, &url, resp).await } /// DELETE a backend resource and parse the response `data` field. /// Mirrors [`Self::post`] (auth header, 401 → session-expiry, error /// classification). pub async fn delete(&self, path: &str) -> anyhow::Result { - enforce_backend_egress(path)?; - emit_backend_egress(path); - self.ensure_budget_available(path).await?; - let url = crate::api::config::api_url(&self.backend_url, path); - tracing::debug!("[integrations] DELETE {}", url); - - let resp = self - .http_client - .delete(&url) - .header("Authorization", format!("Bearer {}", self.auth_token)) - .send() - .await - .map_err(|e| Self::report_transport_error(e, "delete", path, &url))?; - - Self::parse_json_response("delete", path, &url, resp).await + self.request_json(reqwest::Method::DELETE, path, None).await } /// POST a `multipart/form-data` body to a backend endpoint and parse the @@ -677,22 +529,26 @@ impl IntegrationClient { path: &str, form: reqwest::multipart::Form, ) -> anyhow::Result { + reject_backend_webhook_path("POST", path)?; enforce_backend_egress(path)?; emit_backend_egress(path); self.ensure_budget_available(path).await?; let url = crate::api::config::api_url(&self.backend_url, path); tracing::debug!("[integrations] POST(multipart) {}", url); - let resp = self - .http_client - .post(&url) - .header("Authorization", format!("Bearer {}", self.auth_token)) - .multipart(form) - .send() + let value = self + .sdk + .raw() + .post_multipart(path, form) .await - .map_err(|e| Self::report_transport_error(e, "post_multipart", path, &url))?; - - Self::parse_json_response("post_multipart", path, &url, resp).await + .map_err(|error| Self::map_sdk_error(error, "post_multipart", path, &url))?; + // The SDK unwraps successful `{success,data}` responses. Preserve + // compatibility with endpoints that return their payload directly, + // while still recognizing a `success:false` envelope. + if value.get("success") == Some(&serde_json::Value::Bool(false)) { + return Self::parse_envelope("post_multipart", path, &url, value); + } + Ok(serde_json::from_value(value)?) } /// Authenticated GET returning the raw response body plus content-type and @@ -708,54 +564,53 @@ impl IntegrationClient { enforce_backend_egress(path)?; emit_backend_egress(path); self.ensure_budget_available(path).await?; + let route = path.split('?').next().unwrap_or(path); + let segments = route.trim_matches('/').split('/').collect::>(); + let is_file_download = matches!( + segments.as_slice(), + ["agent-integrations", "file-storage", "files", _, "download"] + ); + if !is_file_download { + anyhow::bail!("route is intentionally not exposed by the SDK: GET {route}"); + } let url = crate::api::config::api_url(&self.backend_url, path); tracing::debug!("[integrations] GET(bytes) {}", url); let resp = self - .http_client + .download_client .get(&url) .header("Authorization", format!("Bearer {}", self.auth_token)) .send() .await - .map_err(|e| Self::report_transport_error(e, "get_bytes", path, &url))?; - + .map_err(|error| Self::report_transport_error(error, "get_bytes", path, &url))?; let status = resp.status(); if !status.is_success() { let body_text = resp.text().await.unwrap_or_default(); - let detail = extract_error_detail(&body_text, MAX_ERROR_BODY_LEN); - let status_str = status.as_u16().to_string(); - if status == reqwest::StatusCode::UNAUTHORIZED { - let message = handle_session_jwt_unauthorized("GET", path, &url, &detail); - anyhow::bail!("{message}"); - } - crate::core::observability::report_error_or_expected( - format!("Backend returned {status} for GET {url}: {detail}").as_str(), - "integrations", + let body = + serde_json::from_str(&body_text).unwrap_or(serde_json::Value::String(body_text)); + return Err(Self::map_sdk_error( + SdkError::Status { + status: status.as_u16(), + body, + }, "get_bytes", - &[ - ("path", path), - ("status", status_str.as_str()), - ("failure", "non_2xx"), - ], - ); - anyhow::bail!("Backend returned {status} for GET {url}: {detail}"); + path, + &url, + )); } - let content_type = resp .headers() .get(reqwest::header::CONTENT_TYPE) - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_string()); + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); let filename = resp .headers() .get(reqwest::header::CONTENT_DISPOSITION) - .and_then(|v| v.to_str().ok()) + .and_then(|value| value.to_str().ok()) .and_then(parse_content_disposition_filename); - - let body = resp - .bytes() - .await - .map_err(|e| anyhow::anyhow!("failed to read response body for GET {}: {}", url, e))?; + let body = resp.bytes().await.map_err(|error| { + anyhow::anyhow!("failed to read response body for GET {url}: {error}") + })?; tracing::debug!( "[integrations] GET(bytes) {} → {} bytes (content_type={:?})", url, diff --git a/src/openhuman/integrations/client_tests.rs b/src/openhuman/integrations/client_tests.rs index 83321cc5f..5b52afd1a 100644 --- a/src/openhuman/integrations/client_tests.rs +++ b/src/openhuman/integrations/client_tests.rs @@ -212,6 +212,66 @@ async fn verb_methods_block_user_data_egress_under_local_only() { ); } +/// The OpenHuman compatibility wrapper must retain the SDK's privileged-route +/// denylist on every public verb, including the temporary reqwest download +/// exception. An unreachable base URL makes a transport error the failure +/// mode if any request gets past the local gate. +#[tokio::test] +async fn verb_methods_never_expose_admin_or_webhook_routes() { + let client = client_for("http://127.0.0.1:0".into()); + let body = json!({}); + let assert_blocked = |verb: &str, error: anyhow::Error| { + let message = error.to_string(); + assert!( + message.contains("route is intentionally not exposed by the SDK"), + "{verb}: expected SDK route denylist, got: {message}" + ); + }; + + assert_blocked( + "get", + client + .get::("/coupons/admin") + .await + .unwrap_err(), + ); + assert_blocked( + "post", + client + .post::("/admin/announcements", &body) + .await + .unwrap_err(), + ); + assert_blocked( + "patch", + client + .patch::("/webhooks/core/hook-1", &body) + .await + .unwrap_err(), + ); + assert_blocked( + "delete", + client + .delete::("/webhooks/core/hook-1") + .await + .unwrap_err(), + ); + assert_blocked( + "upload_multipart", + client + .upload_multipart::( + "/webhooks/composio", + reqwest::multipart::Form::new(), + ) + .await + .unwrap_err(), + ); + assert_blocked( + "get_bytes", + client.get_bytes("/webhooks/core").await.unwrap_err(), + ); +} + // ── Integration: HTTP error propagation through `post`/`get` ────── async fn start_mock_backend(app: Router) -> String { diff --git a/src/openhuman/webhooks/ops_tests.rs b/src/openhuman/webhooks/ops_tests.rs index 08e4ca16c..d139d3ed0 100644 --- a/src/openhuman/webhooks/ops_tests.rs +++ b/src/openhuman/webhooks/ops_tests.rs @@ -232,6 +232,7 @@ async fn id_bearing_tunnel_ops_reject_empty_or_whitespace_id() { // ── Authed HTTP round-trips via a mock backend ─────────────── #[tokio::test] +#[ignore = "backend webhook routes are intentionally excluded"] async fn list_tunnels_hits_webhooks_core_endpoint_and_returns_payload() { // Inspect the inbound Authorization header so we catch regressions // where the JWT stops being forwarded (or is sent with the wrong @@ -264,6 +265,7 @@ async fn list_tunnels_hits_webhooks_core_endpoint_and_returns_payload() { } #[tokio::test] +#[ignore = "backend webhook routes are intentionally excluded"] async fn create_tunnel_posts_name_and_optional_description() { let app = Router::new().route( "/webhooks/core", @@ -296,6 +298,7 @@ async fn create_tunnel_posts_name_and_optional_description() { } #[tokio::test] +#[ignore = "backend webhook routes are intentionally excluded"] async fn get_tunnel_encodes_id_in_path() { // Use an id full of reserved URL characters so we actually verify // percent-encoding on the outbound path. axum's `Path` extractor @@ -320,6 +323,7 @@ async fn get_tunnel_encodes_id_in_path() { } #[tokio::test] +#[ignore = "backend webhook routes are intentionally excluded"] async fn update_tunnel_patches_id_with_body() { let app = Router::new().route( "/webhooks/core/{id}", @@ -341,6 +345,7 @@ async fn update_tunnel_patches_id_with_body() { } #[tokio::test] +#[ignore = "backend webhook routes are intentionally excluded"] async fn delete_tunnel_deletes_by_id() { let app = Router::new().route( "/webhooks/core/{id}", @@ -354,6 +359,7 @@ async fn delete_tunnel_deletes_by_id() { } #[tokio::test] +#[ignore = "backend webhook routes are intentionally excluded"] async fn get_bandwidth_fetches_the_bandwidth_endpoint() { let app = Router::new().route( "/webhooks/core/bandwidth",