diff --git a/crates/Cargo.lock b/crates/Cargo.lock index 79b4a20..18bb97f 100644 --- a/crates/Cargo.lock +++ b/crates/Cargo.lock @@ -88,6 +88,17 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -326,6 +337,22 @@ version = "1.0.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" +[[package]] +name = "email-encoding" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "420b9da095f052ea597503e39073b5b3c522f7db933fbac202d91d24492693fd" +dependencies = [ + "base64 0.23.1", + "memchr", +] + +[[package]] +name = "email_address" +version = "0.2.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e079f19b08ca6239f47f8ba8509c11cf3ea30095831f7fed61441475edd8c449" + [[package]] name = "equivalent" version = "1.0.2" @@ -529,13 +556,21 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hub-auth" version = "0.1.0" dependencies = [ + "reqwest", "serde", "serde_json", "thiserror", + "tokio", "tracing", ] @@ -558,6 +593,7 @@ dependencies = [ name = "hub-net" version = "0.1.0" dependencies = [ + "lettre", "reqwest", "serde", "serde_json", @@ -827,6 +863,33 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "lettre" +version = "0.11.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2c646bd5cc763b1087b15493e29a64be6147ba8f19342004fa52048ee596eae" +dependencies = [ + "async-trait", + "base64 0.23.1", + "email-encoding", + "email_address", + "fastrand", + "futures-io", + "futures-util", + "httpdate", + "idna", + "mime", + "nom", + "percent-encoding", + "quoted_printable", + "rustls", + "socket2", + "tokio", + "tokio-rustls", + "url", + "webpki-roots", +] + [[package]] name = "libc" version = "0.2.189" @@ -881,6 +944,12 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "mio" version = "1.2.2" @@ -892,6 +961,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nom" +version = "8.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df9761775871bdef83bee530e60050f7e54b1105350d6884eb0fb4f46c2f9405" +dependencies = [ + "memchr", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1046,6 +1124,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "quoted_printable" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "478e0585659a122aa407eb7e3c0e1fa51b1d8a870038bd29f0cf4a8551eea972" + [[package]] name = "r-efi" version = "6.0.0" @@ -1236,6 +1320,7 @@ version = "0.23.43" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ + "log", "once_cell", "ring", "rustls-pki-types", diff --git a/crates/Cargo.toml b/crates/Cargo.toml index 94bf66f..f2b4a9f 100644 --- a/crates/Cargo.toml +++ b/crates/Cargo.toml @@ -36,3 +36,4 @@ url = "2" anyhow = "1" tempfile = "3" reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +lettre = { version = "0.11", default-features = false, features = ["tokio1-rustls-tls", "smtp-transport", "builder"] } diff --git a/crates/connector-hub/src/main.rs b/crates/connector-hub/src/main.rs index 0a7eb58..12a2d28 100644 --- a/crates/connector-hub/src/main.rs +++ b/crates/connector-hub/src/main.rs @@ -257,6 +257,7 @@ fn build_catalogue() -> anyhow::Result { json_schema: raw.parameters, }, tags: raw.tags, + transport: hub_core::Transport::Http, }); } @@ -269,5 +270,66 @@ fn build_catalogue() -> anyhow::Result { } } + register_email_operations(&mut catalogue); + Ok(catalogue) } + +fn register_email_operations(catalogue: &mut hub_core::Catalogue) { + catalogue.register(hub_core::Operation { + id: hub_core::OperationId("email.send".into()), + provider: "email".into(), + summary: "Send an email via SMTP".into(), + description: "Send an email message using the configured SMTP transport. Supports plain text, HTML, and multipart bodies. Requires SMTP credentials configured via EMAIL_SMTP_* environment variables.".into(), + mutation_class: hub_policy::MutationClass::Mutating, + http_method: "POST".into(), + path_template: String::new(), + parameters: hub_core::ParameterSchema { + json_schema: serde_json::json!({ + "type": "object", + "required": ["to", "subject"], + "properties": { + "to": { + "description": "Recipient email address(es). String or array of strings.", + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}} + ] + }, + "cc": { + "description": "CC recipients", + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}} + ] + }, + "bcc": { + "description": "BCC recipients", + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}} + ] + }, + "subject": { + "type": "string", + "description": "Email subject" + }, + "body": { + "type": "string", + "description": "Plain text body" + }, + "body_html": { + "type": "string", + "description": "HTML body (sent as multipart/alternative with plain text)" + }, + "from": { + "type": "string", + "description": "Override sender address (defaults to configured from_address)" + } + } + }), + }, + tags: vec!["email".into(), "smtp".into()], + transport: hub_core::Transport::Smtp, + }); +} diff --git a/crates/connector-hub/src/mcp.rs b/crates/connector-hub/src/mcp.rs index 82a9604..d30260a 100644 --- a/crates/connector-hub/src/mcp.rs +++ b/crates/connector-hub/src/mcp.rs @@ -1,10 +1,17 @@ +use std::sync::Arc; + use rmcp::handler::server::wrapper::Parameters; use rmcp::model::{CallToolResult, ContentBlock, ServerCapabilities, ServerInfo}; use rmcp::{ErrorData, ServiceExt, schemars, tool, tool_handler, tool_router}; use serde::Deserialize; #[derive(Clone)] -pub struct HubMcpServer; +pub struct HubMcpServer { + dispatcher: Arc, + auth: Arc, + policy: Arc, + net: Arc, +} #[derive(Deserialize, schemars::JsonSchema)] struct SearchParams { @@ -41,8 +48,7 @@ impl HubMcpServer { description = "List all available providers and their operation counts" )] async fn list_providers(&self) -> Result { - let catalogue = - super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?; + let catalogue = self.dispatcher.catalogue(); let mut providers = Vec::new(); for name in catalogue.providers() { @@ -68,8 +74,7 @@ impl HubMcpServer { &self, Parameters(params): Parameters, ) -> Result { - let catalogue = - super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?; + let catalogue = self.dispatcher.catalogue(); let results = catalogue.search(¶ms.query, params.provider.as_deref()); let items: Vec<_> = results @@ -96,8 +101,7 @@ impl HubMcpServer { &self, Parameters(params): Parameters, ) -> Result { - let catalogue = - super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?; + let catalogue = self.dispatcher.catalogue(); let op_id = hub_core::OperationId(params.id.clone()); match catalogue.get(&op_id) { @@ -119,39 +123,19 @@ impl HubMcpServer { &self, Parameters(params): Parameters, ) -> Result { - let catalogue = - super::build_catalogue().map_err(|e| ErrorData::internal_error(e.to_string(), None))?; - let op_id = hub_core::OperationId(params.id.clone()); - let op = catalogue.get(&op_id).ok_or_else(|| { - ErrorData::internal_error(format!("Unknown operation: {}", params.id), None) - })?; - - if params.dry_run.unwrap_or(false) { - let outcome = hub_core::ExecutionOutcome::DryRun { - would_execute: op.summary.clone(), - mutation_class: format!("{:?}", op.mutation_class), - }; - return Ok(CallToolResult::success(vec![ContentBlock::text( - serde_json::to_string_pretty(&outcome).unwrap(), - )])); - } - let policy = hub_policy::Policy::deny_all(); - let auth = hub_auth::AuthStore::from_env(); - let net = hub_net::NetClient::new(hub_net::SsrfPolicy::default()); - - let dispatcher = hub_core::Dispatcher::new(catalogue); - let result = dispatcher + let result = self + .dispatcher .call( &op_id, params.args, params.account.as_deref(), - false, + params.dry_run.unwrap_or(false), params.confirmation_token.as_deref(), - &policy, - &auth, - &net, + &self.policy, + &self.auth, + &self.net, ) .await; @@ -167,7 +151,14 @@ impl HubMcpServer { #[tool(name = "health", description = "Check the health of the connector hub")] async fn health(&self) -> String { - serde_json::json!({"ok": true, "service": "connector-hub"}).to_string() + let catalogue = self.dispatcher.catalogue(); + serde_json::json!({ + "ok": true, + "service": "connector-hub", + "providers": catalogue.providers().len(), + "operations": catalogue.len(), + }) + .to_string() } } @@ -183,9 +174,25 @@ impl rmcp::ServerHandler for HubMcpServer { pub async fn serve() -> anyhow::Result<()> { tracing::info!("starting MCP stdio server"); - let server = HubMcpServer; - let transport = rmcp::transport::io::stdio(); + let catalogue = super::build_catalogue()?; + let dispatcher = hub_core::Dispatcher::new(catalogue); + + let policy_path = std::path::Path::new("permissions.toml"); + let policy = if policy_path.exists() { + let content = std::fs::read_to_string(policy_path)?; + hub_policy::Policy::from_toml(&content)? + } else { + hub_policy::Policy::deny_all() + }; + let server = HubMcpServer { + dispatcher: Arc::new(dispatcher), + auth: Arc::new(hub_auth::AuthStore::from_env()), + policy: Arc::new(policy), + net: Arc::new(hub_net::NetClient::new(hub_net::SsrfPolicy::default())), + }; + + let transport = rmcp::transport::io::stdio(); let running = server.serve(transport).await?; tracing::info!("MCP server running on stdio"); diff --git a/crates/hub-auth/Cargo.toml b/crates/hub-auth/Cargo.toml index 724cf0e..070f7eb 100644 --- a/crates/hub-auth/Cargo.toml +++ b/crates/hub-auth/Cargo.toml @@ -10,3 +10,5 @@ serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } +tokio = { workspace = true } +reqwest = { workspace = true } diff --git a/crates/hub-auth/src/credential.rs b/crates/hub-auth/src/credential.rs index 67d443b..ab35d35 100644 --- a/crates/hub-auth/src/credential.rs +++ b/crates/hub-auth/src/credential.rs @@ -45,4 +45,13 @@ pub enum AuthMethod { password: String, token_url: String, }, + #[serde(rename = "smtp")] + Smtp { + host: String, + port: u16, + username: String, + password: String, + tls: bool, + from_address: String, + }, } diff --git a/crates/hub-auth/src/lib.rs b/crates/hub-auth/src/lib.rs index 8cb036a..d2225a3 100644 --- a/crates/hub-auth/src/lib.rs +++ b/crates/hub-auth/src/lib.rs @@ -2,7 +2,7 @@ pub mod credential; mod store; pub use credential::Credential; -pub use store::AuthStore; +pub use store::{AuthStore, ResolvedAuth}; #[derive(Debug, thiserror::Error)] pub enum AuthError { diff --git a/crates/hub-auth/src/store.rs b/crates/hub-auth/src/store.rs index 3a02de7..fcfc7d9 100644 --- a/crates/hub-auth/src/store.rs +++ b/crates/hub-auth/src/store.rs @@ -1,16 +1,48 @@ use std::collections::HashMap; +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use tokio::sync::RwLock; use crate::AuthError; -use crate::credential::Credential; +use crate::credential::{AuthMethod, Credential}; + +#[derive(Clone)] +struct CachedToken { + access_token: String, + expires_at: u64, +} + +impl CachedToken { + fn is_expired(&self) -> bool { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + now >= self.expires_at.saturating_sub(60) + } +} + +pub struct ResolvedAuth { + pub headers: HashMap, + pub base_url: String, +} pub struct AuthStore { credentials: HashMap<(String, String), Credential>, + token_cache: Arc>>, + http: reqwest::Client, } impl AuthStore { pub fn new() -> Self { Self { credentials: HashMap::new(), + token_cache: Arc::new(RwLock::new(HashMap::new())), + http: reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .expect("failed to build HTTP client for auth"), } } @@ -32,6 +64,225 @@ impl AuthStore { }) } + pub async fn resolve( + &self, + provider: &str, + account: Option<&str>, + ) -> Result { + let cred = self.get(provider, account)?; + let base_url = cred.base_url.clone().unwrap_or_default(); + let mut headers = HashMap::new(); + + match &cred.auth { + AuthMethod::Bearer { token } => { + headers.insert("Authorization".into(), format!("Bearer {token}")); + } + AuthMethod::Header { name, value } => { + headers.insert(name.clone(), value.clone()); + } + AuthMethod::Basic { username, password } => { + let encoded = base64_encode(format!("{username}:{password}").as_bytes()); + headers.insert("Authorization".into(), format!("Basic {encoded}")); + } + AuthMethod::OAuth2 { + client_id, + client_secret, + refresh_token, + token_url, + .. + } => { + let key = ( + provider.to_string(), + account.unwrap_or("default").to_string(), + ); + let access_token = self + .get_or_refresh_oauth2( + &key, + client_id, + client_secret, + refresh_token, + token_url, + provider, + ) + .await?; + headers.insert("Authorization".into(), format!("Bearer {access_token}")); + } + AuthMethod::PasswordGrant { + client_id, + client_secret, + username, + password, + token_url, + } => { + let key = ( + provider.to_string(), + account.unwrap_or("default").to_string(), + ); + let access_token = self + .get_or_refresh_password_grant( + &key, + client_id, + client_secret, + username, + password, + token_url, + provider, + ) + .await?; + headers.insert("Authorization".into(), format!("Bearer {access_token}")); + } + AuthMethod::Hmac { + app_key, + consumer_key, + .. + } => { + headers.insert("X-Ovh-Application".into(), app_key.clone()); + headers.insert("X-Ovh-Consumer".into(), consumer_key.clone()); + } + AuthMethod::Smtp { .. } => { + // SMTP auth is handled by the mail transport, not via HTTP headers + } + } + + Ok(ResolvedAuth { headers, base_url }) + } + + async fn get_or_refresh_oauth2( + &self, + key: &(String, String), + client_id: &str, + client_secret: &str, + refresh_token: &str, + token_url: &str, + provider: &str, + ) -> Result { + { + let cache = self.token_cache.read().await; + if let Some(cached) = cache.get(key) + && !cached.is_expired() + { + return Ok(cached.access_token.clone()); + } + } + + tracing::info!(provider, "refreshing OAuth2 access token"); + + let token = self + .request_token( + token_url, + &[ + ("grant_type", "refresh_token"), + ("client_id", client_id), + ("client_secret", client_secret), + ("refresh_token", refresh_token), + ], + provider, + ) + .await?; + + let mut cache = self.token_cache.write().await; + cache.insert(key.clone(), token.clone()); + Ok(token.access_token) + } + + #[allow(clippy::too_many_arguments)] + async fn get_or_refresh_password_grant( + &self, + key: &(String, String), + client_id: &str, + client_secret: &str, + username: &str, + password: &str, + token_url: &str, + provider: &str, + ) -> Result { + { + let cache = self.token_cache.read().await; + if let Some(cached) = cache.get(key) + && !cached.is_expired() + { + return Ok(cached.access_token.clone()); + } + } + + tracing::info!(provider, "refreshing password grant access token"); + + let token = self + .request_token( + token_url, + &[ + ("grant_type", "password"), + ("client_id", client_id), + ("client_secret", client_secret), + ("username", username), + ("password", password), + ], + provider, + ) + .await?; + + let mut cache = self.token_cache.write().await; + cache.insert(key.clone(), token.clone()); + Ok(token.access_token) + } + + async fn request_token( + &self, + token_url: &str, + params: &[(&str, &str)], + provider: &str, + ) -> Result { + let resp = self + .http + .post(token_url) + .form(params) + .send() + .await + .map_err(|e| AuthError::RefreshFailed { + provider: provider.into(), + reason: e.to_string(), + })?; + + if !resp.status().is_success() { + let status = resp.status().as_u16(); + let body = resp.text().await.unwrap_or_default(); + return Err(AuthError::RefreshFailed { + provider: provider.into(), + reason: format!("token endpoint returned {status}: {body}"), + }); + } + + let body: serde_json::Value = resp.json().await.map_err(|e| AuthError::RefreshFailed { + provider: provider.into(), + reason: format!("invalid token response: {e}"), + })?; + + let access_token = body + .get("access_token") + .and_then(|v| v.as_str()) + .ok_or_else(|| AuthError::RefreshFailed { + provider: provider.into(), + reason: "no access_token in response".into(), + })? + .to_string(); + + let expires_in = body + .get("expires_in") + .and_then(|v| v.as_u64()) + .unwrap_or(3600); + + let expires_at = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + + expires_in; + + Ok(CachedToken { + access_token, + expires_at, + }) + } + pub fn list_accounts(&self, provider: &str) -> Vec<&str> { self.credentials .keys() @@ -60,7 +311,7 @@ impl AuthStore { provider: "hetzner".into(), account_id: "default".into(), base_url: Some("https://api.hetzner.cloud/v1".into()), - auth: crate::credential::AuthMethod::Bearer { token }, + auth: AuthMethod::Bearer { token }, }); } @@ -69,7 +320,7 @@ impl AuthStore { provider: "github".into(), account_id: "default".into(), base_url: Some("https://api.github.com".into()), - auth: crate::credential::AuthMethod::Bearer { token }, + auth: AuthMethod::Bearer { token }, }); } @@ -78,7 +329,7 @@ impl AuthStore { provider: "openai".into(), account_id: "default".into(), base_url: Some("https://api.openai.com/v1".into()), - auth: crate::credential::AuthMethod::Bearer { token: key }, + auth: AuthMethod::Bearer { token: key }, }); } @@ -87,16 +338,13 @@ impl AuthStore { provider: "anthropic".into(), account_id: "default".into(), base_url: Some("https://api.anthropic.com/v1".into()), - auth: crate::credential::AuthMethod::Header { + auth: AuthMethod::Header { name: "x-api-key".into(), value: key, }, }); } - // Gmail OAuth2 accounts: GMAIL_CLIENT_ID, GMAIL_CLIENT_SECRET, plus - // GMAIL_REFRESH_TOKEN (default account) or GMAIL_REFRESH_TOKEN_