admin/openhuman
1
0
Fork 0
mirror of https://github.com/tinyhumansai/openhuman.git synced 2026-07-30 06:32:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(core): cut managed backend traffic over to Rust SDK (#5233)

Browse Source
This commit is contained in:
Steven Enamakel
2026-07-28 11:13:37 +03:00
committed by GitHub
parent 2d32656618
commit fe33726702
8 changed files with 470 additions and 616 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+12
View File
@@ -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`, …).
Cargo.lock
Generated
+15
View File
@@ -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"
Cargo.toml
+1
View File
@@ -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"
src/api/rest.rs
+175 -331
View File
@@ -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<Client> {
.map_err(|e| anyhow::anyhow!("failed to build HTTP client: {e}"))
}
fn parse_api_response_json(text: &str) -> Result<Value> {
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<String, Value>) -> Option<String> {
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<String>,
#[serde(default)]
state: Option<String>,
}
#[derive(Debug, Clone, Deserialize)]
struct IntegrationsEnvelope {
success: bool,
data: IntegrationsData,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
struct IntegrationsData {
integrations: Vec<IntegrationSummary>,
}
/// 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<ConnectResponse> {
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<Value> {
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<Value>,
) -> Result<Value> {
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<Vec<IntegrationSummary>> {
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(())
}
}
src/api/rest_tests.rs
+61
View File
@@ -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]
src/openhuman/integrations/client.rs
+140 -285
View File
@@ -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<Arc<crate::openhuman::config::Config>>,
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<IntegrationPricing>,
}
@@ -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<T> {
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<T> = 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<T: serde::de::DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
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<T> = 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<T>`
/// envelope parsing with `success:false` classification.
async fn parse_json_response<T: serde::de::DeserializeOwned>(
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<T: serde::de::DeserializeOwned>(
method: &str,
path: &str,
url: &str,
resp: reqwest::Response,
value: serde_json::Value,
) -> anyhow::Result<T> {
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<T> = resp.json().await?;
let envelope: BackendResponse<T> = serde_json::from_value(value)?;
if !envelope.success {
let msg = envelope
.error
@@ -620,6 +479,28 @@ impl IntegrationClient {
})
}
async fn request_json<T: serde::de::DeserializeOwned>(
&self,
method: reqwest::Method,
path: &str,
body: Option<&serde_json::Value>,
) -> anyhow::Result<T> {
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<T> {
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<T: serde::de::DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
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<T> {
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::<Vec<_>>();
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,
src/openhuman/integrations/client_tests.rs
+60
View File
@@ -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::<serde_json::Value>("/coupons/admin")
.await
.unwrap_err(),
);
assert_blocked(
"post",
client
.post::<serde_json::Value>("/admin/announcements", &body)
.await
.unwrap_err(),
);
assert_blocked(
"patch",
client
.patch::<serde_json::Value>("/webhooks/core/hook-1", &body)
.await
.unwrap_err(),
);
assert_blocked(
"delete",
client
.delete::<serde_json::Value>("/webhooks/core/hook-1")
.await
.unwrap_err(),
);
assert_blocked(
"upload_multipart",
client
.upload_multipart::<serde_json::Value>(
"/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 {
src/openhuman/webhooks/ops_tests.rs
+6
View File
@@ -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",