From 1c766ca73e070b75adc31b0da84b6082dfce2e27 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aras=20=C3=96rs?= <83857485+arasors@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:33:18 +0300 Subject: [PATCH 01/10] feat(rust): establish native SDK transport and integration scaffold --- .github/workflows/rust-sdk.yml | 30 +++++++ Cargo.toml | 3 + rust/Cargo.toml | 34 ++++++++ rust/README.md | 23 +++++ rust/src/error.rs | 72 ++++++++++++++++ rust/src/http.rs | 150 +++++++++++++++++++++++++++++++++ rust/src/lib.rs | 68 +++++++++++++++ rust/src/resources/mod.rs | 3 + rust/src/types/mod.rs | 7 ++ 9 files changed, 390 insertions(+) create mode 100644 .github/workflows/rust-sdk.yml create mode 100644 Cargo.toml create mode 100644 rust/Cargo.toml create mode 100644 rust/README.md create mode 100644 rust/src/error.rs create mode 100644 rust/src/http.rs create mode 100644 rust/src/lib.rs create mode 100644 rust/src/resources/mod.rs create mode 100644 rust/src/types/mod.rs diff --git a/.github/workflows/rust-sdk.yml b/.github/workflows/rust-sdk.yml new file mode 100644 index 0000000..8fcfa01 --- /dev/null +++ b/.github/workflows/rust-sdk.yml @@ -0,0 +1,30 @@ +name: Rust SDK +on: + pull_request: + paths: ["rust/**", "Cargo.toml", "Cargo.lock", ".github/workflows/rust-sdk.yml"] + push: + branches: [main] + paths: ["rust/**", "Cargo.toml", "Cargo.lock", ".github/workflows/rust-sdk.yml"] + workflow_dispatch: +permissions: + contents: read +jobs: + test: + runs-on: ubuntu-22.04 + timeout-minutes: 25 + steps: + - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 + - name: Install Rust toolchain + run: rustup toolchain install stable --profile minimal --component rustfmt --component clippy + - name: Check formatting + run: cargo +stable fmt --all -- --check + - name: Check all targets + run: cargo +stable clippy --workspace --all-targets --all-features -- -D warnings + - name: Test SDK and examples + run: cargo +stable test --workspace --all-features + - name: Build documentation + env: + RUSTDOCFLAGS: -D warnings + run: cargo +stable doc --workspace --all-features --no-deps + - name: Check crate packaging + run: cargo +stable package --manifest-path rust/Cargo.toml --allow-dirty diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000..d50b2af --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,3 @@ +[workspace] +members = ["rust"] +resolver = "2" diff --git a/rust/Cargo.toml b/rust/Cargo.toml new file mode 100644 index 0000000..241151f --- /dev/null +++ b/rust/Cargo.toml @@ -0,0 +1,34 @@ +[package] +name = "sentroy-client-sdk" +version = "0.1.0" +edition = "2021" +rust-version = "1.88" +description = "Async Rust client for Sentroy Mail, Storage, WhatsApp, Auth, Vault and MCP" +license = "MIT" +repository = "https://github.com/Sentroy-Co/client-sdk" +homepage = "https://docs.sentroy.com" +readme = "README.md" +keywords = ["sentroy", "sdk", "email", "storage", "mcp"] +categories = ["api-bindings", "web-programming::http-client"] + +[dependencies] +base64 = "0.22" +bytes = "1" +chrono = { version = "0.4", default-features = false, features = ["std", "clock"] } +futures-util = "0.3" +hmac = "0.12" +jsonwebtoken = { version = "9.3.1", default-features = false } +percent-encoding = "2" +rand = "0.8" +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "multipart", "stream"] } +reqwest13 = { package = "reqwest", version = "0.13", default-features = false, features = ["rustls"] } +rmcp = { version = "3.8.0", default-features = false, features = ["client", "transport-streamable-http-client-reqwest", "reqwest"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +sha2 = "0.10" +tokio = { version = "1", features = ["sync", "time", "fs", "io-util", "rt"] } +url = "2" +uuid = { version = "1", features = ["v4"] } + +[dev-dependencies] +tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "io-util"] } diff --git a/rust/README.md b/rust/README.md new file mode 100644 index 0000000..7f827ae --- /dev/null +++ b/rust/README.md @@ -0,0 +1,23 @@ +# Sentroy Rust SDK + +Async native Rust client for Mail, Storage, WhatsApp, Auth, Env Vault and MCP. +This source package follows TypeScript SDK 2.18.3 at commit +`326f131dec57c8e8a4dd6b2f8eb18020a772e861`. Rust 1.88 or newer is required. + +The package is being prepared as version 0.1.0; no crates.io publication is +performed by this change. React components, browser UI and the Node CLI remain +TypeScript products. Rust exposes portable service APIs and native integration +helpers instead of pretending to run browser ceremonies. + +```rust,no_run +use sentroy_client_sdk::Sentroy; +# async fn example() -> Result<(), Box> { +let client = Sentroy::new("https://sentroy.com", "my-company", std::env::var("SENTROY_TOKEN")?)?; +let domains = client.domains.list().await?; +println!("{} domains", domains.len()); +# Ok(()) } +``` + +Transport does not automatically follow HTTP redirects or retry application +requests. BCC compose errors retain the request UUID for explicit recovery. +Authentication namespaces for REST, Vault and MCP are separate. diff --git a/rust/src/error.rs b/rust/src/error.rs new file mode 100644 index 0000000..5a89ff6 --- /dev/null +++ b/rust/src/error.rs @@ -0,0 +1,72 @@ +use serde_json::Value; +use std::fmt; + +/// An SDK failure. Compose errors retain the UUID needed for explicit recovery. +pub enum Error { + InvalidInput(String), + InvalidResponse(String), + Http { status: u16, message: String, body: Value }, + Transport(String), + Json(serde_json::Error), + Io(std::io::Error), + Auth { code: String, description: String, status: Option }, + Mcp(String), + ComposeRequest { request_id: String, source: Box }, +} + +impl Error { + pub fn status_code(&self) -> Option { + match self { + Self::Http { status, .. } => Some(*status), + Self::Auth { status, .. } => *status, + Self::ComposeRequest { source, .. } => source.status_code(), + _ => None, + } + } + pub fn body(&self) -> Option<&Value> { + match self { + Self::Http { body, .. } => Some(body), + Self::ComposeRequest { source, .. } => source.body(), + _ => None, + } + } + pub fn compose_request_id(&self) -> Option<&str> { + match self { Self::ComposeRequest { request_id, .. } => Some(request_id), _ => None } + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidInput(s) => write!(f, "Invalid input: {s}"), + Self::InvalidResponse(s) => write!(f, "Invalid response: {s}"), + Self::Http { status, message, .. } => write!(f, "Sentroy HTTP {status}: {message}"), + Self::Transport(s) => write!(f, "HTTP transport failed: {s}"), + Self::Json(e) => write!(f, "JSON conversion failed: {e}"), + Self::Io(e) => write!(f, "I/O failed: {e}"), + Self::Auth { code, description, .. } => write!(f, "Auth {code}: {description}"), + Self::Mcp(s) => write!(f, "MCP failed: {s}"), + Self::ComposeRequest { request_id, source } => write!(f, "Compose request {request_id} requires explicit outcome recovery: {source}"), + } + } +} + +// Deliberately do not include response bodies in Debug output. +impl fmt::Debug for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { fmt::Display::fmt(self, f) } +} +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Json(e) => Some(e), Self::Io(e) => Some(e), + Self::ComposeRequest { source, .. } => Some(source.as_ref()), + _ => None, + } + } +} +impl From for Error { fn from(e: serde_json::Error) -> Self { Self::Json(e) } } +impl From for Error { fn from(e: std::io::Error) -> Self { Self::Io(e) } } +impl From for Error { + fn from(e: reqwest::Error) -> Self { Self::Transport(e.without_url().to_string()) } +} +pub type Result = std::result::Result; diff --git a/rust/src/http.rs b/rust/src/http.rs new file mode 100644 index 0000000..47c218e --- /dev/null +++ b/rust/src/http.rs @@ -0,0 +1,150 @@ +use crate::{Error, Result}; +use percent_encoding::{utf8_percent_encode, AsciiSet, NON_ALPHANUMERIC}; +use reqwest::{header, Method, RequestBuilder, Response}; +use serde::{de::DeserializeOwned, Serialize}; +use serde_json::Value; +use std::{fmt, time::Duration}; +use url::Url; + +const SEGMENT: &AsciiSet = &NON_ALPHANUMERIC.remove(b'-').remove(b'_').remove(b'.') + .remove(b'~').remove(b'!').remove(b'\'').remove(b'(').remove(b')').remove(b'*'); +const JSON_LIMIT: usize = 16 * 1024 * 1024; +pub(crate) type Query = Vec<(String, String)>; + +/// Encode exactly one path segment. Dot segments are rejected before URL parsing. +pub(crate) fn segment(value: &str) -> Result { + if value.is_empty() || value == "." || value == ".." || value.chars().any(char::is_control) { + return Err(Error::InvalidInput("empty, control or dot path segment".into())); + } + Ok(utf8_percent_encode(value, SEGMENT).to_string()) +} +pub(crate) fn query(value: &T) -> Result { + let value = serde_json::to_value(value)?; + if value.is_null() { return Ok(Vec::new()); } + let fields = value.as_object().ok_or_else(|| Error::InvalidInput("query must be an object".into()))?; + fields.iter().filter(|(_, v)| !v.is_null()).map(|(k, v)| { + let text = match v { + Value::String(s) => s.clone(), + Value::Bool(_) | Value::Number(_) => v.to_string(), + Value::Array(items) => items.iter().map(|i| match i { + Value::String(s) => Ok(s.clone()), + Value::Bool(_) | Value::Number(_) => Ok(i.to_string()), + _ => Err(Error::InvalidInput("query array contains a non-scalar".into())), + }).collect::>>()?.join(","), + _ => return Err(Error::InvalidInput("query contains a non-scalar".into())), + }; + Ok((k.clone(), text)) + }).collect() +} + +#[derive(Clone)] +pub(crate) struct Http { base: String, token: Option, client: reqwest::Client } +impl fmt::Debug for Http { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Http").field("base", &self.base).field("authenticated", &self.token.is_some()).finish() + } +} +impl Http { + pub(crate) fn new(base: impl Into, token: Option, timeout: Duration) -> Result { + let base = base.into().trim_end_matches('/').to_string(); + let url = Url::parse(&base).map_err(|_| Error::InvalidInput("invalid base URL".into()))?; + if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() + || !url.username().is_empty() || url.password().is_some() || url.query().is_some() + || url.fragment().is_some() || timeout.is_zero() { + return Err(Error::InvalidInput("base URL must be HTTP(S), without credentials, query or fragment; timeout must be positive".into())); + } + let client = reqwest::Client::builder().redirect(reqwest::redirect::Policy::none()).timeout(timeout).build()?; + Self { base, token: None, client }.with_token(token) + } + pub(crate) fn with_token(&self, token: Option) -> Result { + let token = token.map(|t| { + if t.is_empty() { return Err(Error::InvalidInput("empty access token".into())); } + let mut v = header::HeaderValue::from_str(&format!("Bearer {t}")) + .map_err(|_| Error::InvalidInput("invalid access token header".into()))?; + v.set_sensitive(true); + Ok(v) + }).transpose()?; + Ok(Self { base: self.base.clone(), token, client: self.client.clone() }) + } + fn request(&self, method: Method, path: &str, query: &[(String, String)]) -> Result { + if !path.starts_with('/') || path.starts_with("//") || path.contains(['?', '#']) + || path.chars().any(char::is_control) { + return Err(Error::InvalidInput("invalid relative API path".into())); + } + let url = Url::parse(&format!("{}{path}", self.base)) + .map_err(|_| Error::InvalidInput("invalid API URL".into()))?; + let mut request = self.client.request(method, url).query(query).header(header::ACCEPT, "application/json"); + if let Some(token) = &self.token { request = request.header(header::AUTHORIZATION, token.clone()); } + Ok(request) + } + async fn read_limited(mut response: Response, limit: usize) -> Result<(u16, Vec)> { + let status = response.status().as_u16(); + if response.content_length().is_some_and(|n| n > limit as u64) { + return Err(Error::InvalidResponse("response exceeds byte limit".into())); + } + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await? { + if chunk.len() > limit.saturating_sub(bytes.len()) { + return Err(Error::InvalidResponse("response exceeds byte limit".into())); + } + bytes.extend_from_slice(&chunk); + } + Ok((status, bytes)) + } + async fn decode(response: Response, once: bool) -> Result { + if once && (response.status().is_redirection() || !response.headers().get(header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()).is_some_and(|v| v.to_ascii_lowercase().starts_with("application/json"))) { + return Err(Error::InvalidResponse("invalid compose receipt".into())); + } + let (status, bytes) = Self::read_limited(response, if once { 4096 } else { JSON_LIMIT }).await?; + let parsed = if bytes.is_empty() { Ok(Value::Null) } else { serde_json::from_slice(&bytes) }; + if !(200..300).contains(&status) { + let body = parsed.unwrap_or(Value::Null); + let message = body.get("error").and_then(Value::as_str) + .or_else(|| body.get("error_description").and_then(Value::as_str)) + .unwrap_or("request failed").to_string(); + return Err(Error::Http { status, message, body }); + } + let mut value = parsed?; + let payload = if value.as_object().is_some_and(|v| v.contains_key("data")) { + value.get_mut("data").expect("checked data").take() + } else { value }; + Ok(serde_json::from_value(payload)?) + } + pub(crate) async fn get(&self, path: &str, query: &[(String, String)]) -> Result { + Self::decode(self.request(Method::GET, path, query)?.send().await?, false).await + } + pub(crate) async fn post(&self, path: &str, body: Value) -> Result { + Self::decode(self.request(Method::POST, path, &[])?.json(&body).send().await?, false).await + } + pub(crate) async fn put(&self, path: &str, body: Value) -> Result { + Self::decode(self.request(Method::PUT, path, &[])?.json(&body).send().await?, false).await + } + pub(crate) async fn patch(&self, path: &str, body: Value) -> Result { + Self::decode(self.request(Method::PATCH, path, &[])?.json(&body).send().await?, false).await + } + pub(crate) async fn delete(&self, path: &str, query: &[(String, String)]) -> Result { + Self::decode(self.request(Method::DELETE, path, query)?.send().await?, false).await + } + pub(crate) async fn delete_body(&self, path: &str, body: Value) -> Result { + Self::decode(self.request(Method::DELETE, path, &[])?.json(&body).send().await?, false).await + } + pub(crate) async fn post_once(&self, path: &str, body: Value) -> Result { + if !matches!(path, "/send-bcc" | "/inbox/drafts-bcc-v1") { + return Err(Error::InvalidInput("invalid compose path".into())); + } + Self::decode(self.request(Method::POST, path, &[])?.json(&body).send().await?, true).await + } + pub(crate) async fn post_form(&self, path: &str, body: reqwest::multipart::Form) -> Result { + Self::decode(self.request(Method::POST, path, &[])?.multipart(body).send().await?, false).await + } + pub(crate) async fn raw_get(&self, path: &str, query: &[(String, String)]) -> Result { + let response = self.request(Method::GET, path, query)?.send().await?; + if !response.status().is_success() { + return match Self::decode::(response, false).await { + Err(e) => Err(e), Ok(_) => Err(Error::InvalidResponse("unsuccessful raw response".into())) + }; + } + Ok(response) + } +} diff --git a/rust/src/lib.rs b/rust/src/lib.rs new file mode 100644 index 0000000..ebc0b9d --- /dev/null +++ b/rust/src/lib.rs @@ -0,0 +1,68 @@ +//! Native async Rust client for the Sentroy platform. +//! See the crate README for authentication, module coverage and recovery behavior. +mod error; +pub(crate) mod http; +pub mod auth; +pub mod mcp; +pub mod resources; +pub mod thumbnails; +pub mod types; +pub mod vault; + +pub use error::{Error, Result}; +pub use resources::{mail::*, storage::*, whatsapp::*}; +use http::{segment, Http}; +use std::time::Duration; + +/// One company-scoped client. Clones share HTTP connection pools. +#[derive(Clone, Debug)] +pub struct Sentroy { + pub domains: Domains, + pub mailboxes: Mailboxes, + pub templates: Templates, + pub inbox: Inbox, + pub send: Send, + pub audience: Audience, + pub suppressions: Suppressions, + pub webhooks: Webhooks, + pub logs: Logs, + pub buckets: Buckets, + pub media: MediaResource, + pub storage: Storage, + pub whatsapp: WhatsApp, +} +impl Sentroy { + pub fn new(base_url: impl Into, company_slug: impl Into, access_token: impl Into) -> Result { + Self::builder(company_slug).base_url(base_url).access_token(access_token).build() + } + pub fn builder(company_slug: impl Into) -> SentroyBuilder { + SentroyBuilder { base_url: "https://sentroy.com".into(), company_slug: company_slug.into(), token: None, timeout: Duration::from_secs(30) } + } +} +/// Client configuration. Tokens are excluded from Debug output. +pub struct SentroyBuilder { base_url: String, company_slug: String, token: Option, timeout: Duration } +impl std::fmt::Debug for SentroyBuilder { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SentroyBuilder").field("base_url", &self.base_url).field("company_slug", &self.company_slug).field("authenticated", &self.token.is_some()).field("timeout", &self.timeout).finish() + } +} +impl SentroyBuilder { + pub fn base_url(mut self, value: impl Into) -> Self { self.base_url = value.into(); self } + pub fn access_token(mut self, value: impl Into) -> Self { self.token = Some(value.into()); self } + pub fn timeout(mut self, value: Duration) -> Self { self.timeout = value; self } + pub fn build(self) -> Result { + let root = self.base_url.trim_end_matches('/'); + let slug = segment(&self.company_slug)?; + let mail = Http::new(format!("{root}/api/mail/companies/{slug}"), self.token.clone(), self.timeout)?; + let storage = Http::new(format!("{root}/api/storage/companies/{slug}"), self.token.clone(), self.timeout)?; + let whatsapp = Http::new(format!("{root}/api/whatsapp/companies/{slug}"), self.token, self.timeout)?; + Ok(Sentroy { + domains: Domains::new(mail.clone()), mailboxes: Mailboxes::new(mail.clone()), + templates: Templates::new(mail.clone()), inbox: Inbox::new(mail.clone()), + send: Send::new(mail.clone()), audience: Audience::new(mail.clone()), + suppressions: Suppressions::new(mail.clone()), webhooks: Webhooks::new(mail.clone()), logs: Logs::new(mail), + buckets: Buckets::new(storage.clone()), media: MediaResource::new(storage.clone()), storage: Storage::new(storage), + whatsapp: WhatsApp::new(whatsapp), + }) + } +} diff --git a/rust/src/resources/mod.rs b/rust/src/resources/mod.rs new file mode 100644 index 0000000..34af966 --- /dev/null +++ b/rust/src/resources/mod.rs @@ -0,0 +1,3 @@ +pub mod mail; +pub mod storage; +pub mod whatsapp; diff --git a/rust/src/types/mod.rs b/rust/src/types/mod.rs new file mode 100644 index 0000000..f0d20dc --- /dev/null +++ b/rust/src/types/mod.rs @@ -0,0 +1,7 @@ +//! Wire-compatible platform request and response models. +pub mod mail; +pub mod storage; +pub mod whatsapp; +pub use mail::*; +pub use storage::*; +pub use whatsapp::*; From 246ce11fe08d49b214c27cbce0840aa27e8dba34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aras=20=C3=96rs?= <83857485+arasors@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:37:24 +0300 Subject: [PATCH 02/10] feat(rust): integrate typed REST, identity, vault and MCP clients --- rust/src/auth/admin.rs | 100 ++++++ rust/src/auth/client.rs | 166 ++++++++++ rust/src/auth/mod.rs | 8 + rust/src/auth/types.rs | 27 ++ rust/src/mcp/file_info.rs | 61 ++++ rust/src/mcp/mod.rs | 150 +++++++++ rust/src/mcp/transfers.rs | 252 +++++++++++++++ rust/src/resources/mail.rs | 176 +++++++++++ rust/src/resources/storage.rs | 66 ++++ rust/src/resources/whatsapp.rs | 49 +++ rust/src/thumbnails.rs | 62 ++++ rust/src/types/mail.rs | 550 +++++++++++++++++++++++++++++++++ rust/src/types/storage.rs | 96 ++++++ rust/src/types/whatsapp.rs | 92 ++++++ rust/src/vault.rs | 235 ++++++++++++++ rust/tests/auth.rs | 29 ++ rust/tests/mail.rs | 80 +++++ rust/tests/storage.rs | 74 +++++ rust/tests/vault.rs | 120 +++++++ rust/tests/whatsapp.rs | 53 ++++ 20 files changed, 2446 insertions(+) create mode 100644 rust/src/auth/admin.rs create mode 100644 rust/src/auth/client.rs create mode 100644 rust/src/auth/mod.rs create mode 100644 rust/src/auth/types.rs create mode 100644 rust/src/mcp/file_info.rs create mode 100644 rust/src/mcp/mod.rs create mode 100644 rust/src/mcp/transfers.rs create mode 100644 rust/src/resources/mail.rs create mode 100644 rust/src/resources/storage.rs create mode 100644 rust/src/resources/whatsapp.rs create mode 100644 rust/src/thumbnails.rs create mode 100644 rust/src/types/mail.rs create mode 100644 rust/src/types/storage.rs create mode 100644 rust/src/types/whatsapp.rs create mode 100644 rust/src/vault.rs create mode 100644 rust/tests/auth.rs create mode 100644 rust/tests/mail.rs create mode 100644 rust/tests/storage.rs create mode 100644 rust/tests/vault.rs create mode 100644 rust/tests/whatsapp.rs diff --git a/rust/src/auth/admin.rs b/rust/src/auth/admin.rs new file mode 100644 index 0000000..4192f3d --- /dev/null +++ b/rust/src/auth/admin.rs @@ -0,0 +1,100 @@ +use std::{sync::Arc, time::{Duration, Instant, SystemTime, UNIX_EPOCH}}; +use serde_json::{json, Value}; +use tokio::sync::Mutex; +use jsonwebtoken::{decode, decode_header, Algorithm, DecodingKey, Validation}; +use crate::{Error, Result}; +use super::{client::auth_error, LoginOutcome, SentroyAuth}; + +/// Server-side, request-scoped administrator. Keep the project API key secret. +#[derive(Clone)] +pub struct SentroyAuthAdmin { + client: SentroyAuth, + api_key: String, + jwks_ttl: Duration, + jwks: Arc>>, +} +impl std::fmt::Debug for SentroyAuthAdmin { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SentroyAuthAdmin").field("client", &self.client) + .field("api_key", &"[REDACTED]").finish() + } +} +impl SentroyAuthAdmin { + pub fn new(base: impl Into, project_slug: impl Into, api_key: impl Into, timeout: Duration, jwks_ttl: Duration) -> Result { + let api_key = api_key.into(); + if api_key.is_empty() { return Err(Error::InvalidInput("API key must not be empty".into())); } + let client = SentroyAuth::new(base, project_slug, Some(api_key.clone()), timeout)?; + Ok(Self { client, api_key, jwks_ttl, jwks: Arc::new(Mutex::new(None)) }) + } + fn path(&self, p: &str) -> String { + self.client.path(p) + } + async fn post(&self, p: &str, body: Value) -> Result { + self.client.http.post(&self.path(p), body).await.map_err(auth_error) + } + pub async fn create_user(&self, input: Value) -> Result { self.post("/signup", input).await } + pub async fn sign_in(&self, input: Value) -> Result { + let v = self.post("/login", input).await?; + if v.get("mfaRequired") == Some(&Value::Bool(true)) { Ok(LoginOutcome::Mfa(v)) } else { Ok(LoginOutcome::Tokens(v)) } + } + pub async fn verify_mfa(&self, input: Value) -> Result { self.post("/login/mfa/verify", input).await } + pub async fn refresh(&self, token: &str) -> Result { self.post("/refresh", json!({"refreshToken":token})).await } + pub async fn sign_out(&self, token: &str) -> Result { self.post("/logout", json!({"refreshToken":token})).await } + pub async fn verify_email(&self, token: &str) -> Result { self.post("/verify-email", json!({"token":token})).await } + pub async fn request_password_reset(&self, email: &str) -> Result { self.post("/password-reset/request", json!({"email":email})).await } + pub async fn confirm_password_reset(&self, input: Value) -> Result { self.post("/password-reset/confirm", input).await } + pub async fn send_magic_link(&self, input: Value) -> Result { self.post("/magic-link/request", input).await } + pub async fn consume_magic_link(&self, token: &str) -> Result { self.post("/magic-link/consume", json!({"token":token})).await } + pub async fn accept_invitation(&self, input: Value) -> Result { self.post("/invitation/accept", input).await } + pub async fn get_user(&self, token: &str) -> Result { + self.client.http.with_token(Some(token.to_owned()))?.get(&self.path("/me"), &[]).await.map_err(auth_error) + } + pub async fn get_userinfo(&self, token: &str) -> Result { + self.client.http.with_token(Some(token.to_owned()))?.get(&self.path("/userinfo"), &[]).await.map_err(auth_error) + } + /// The reference explicitly has no public admin user-list endpoint. + pub fn list_users(&self) -> Result { + Err(Error::InvalidInput("admin users.list requires the session-authenticated dashboard API; no public API is available".into())) + } + pub async fn invalidate_jwks_cache(&self) { *self.jwks.lock().await = None; } + async fn jwks(&self) -> Result { + let mut cache = self.jwks.lock().await; + if let Some((at, keys)) = cache.as_ref() { + if at.elapsed() < self.jwks_ttl { return Ok(keys.clone()); } + } + // Always use the configured auth origin; never follow a JWT-supplied URL. + let v: Value = self.client.http.get(&self.path("/jwks.json"), &[]).await.map_err(auth_error)?; + *cache = Some((Instant::now(), v.clone())); + Ok(v) + } + /// Verify an RS256 signature against the configured project's JWKS, then + /// require the configured issuer and the reference API-key audience rule. + pub async fn verify_id_token(&self, token: &str) -> Result { + let invalid = |s: &str| Error::InvalidResponse(s.to_owned()); + let header = decode_header(token).map_err(|_| invalid("Malformed JWT"))?; + if header.alg != Algorithm::RS256 { return Err(invalid("Only RS256 supported")); } + let jwks = self.jwks().await?; + let keys = jwks.get("keys").and_then(Value::as_array).ok_or_else(|| invalid("Malformed JWKS"))?; + let key = keys.iter().find(|k| k.get("kid").and_then(Value::as_str) == header.kid.as_deref()) + .or_else(|| keys.first()).ok_or_else(|| invalid("No public key in JWKS"))?; + if key.get("kty").and_then(Value::as_str) != Some("RSA") { return Err(invalid("JWKS key is not RSA")); } + let n = key.get("n").and_then(Value::as_str).ok_or_else(|| invalid("Missing RSA modulus"))?; + let e = key.get("e").and_then(Value::as_str).ok_or_else(|| invalid("Missing RSA exponent"))?; + let decoding_key = DecodingKey::from_rsa_components(n, e).map_err(|_| invalid("Invalid RSA key"))?; + let mut validation = Validation::new(Algorithm::RS256); + // Validate the exact reference semantics below, after signature verification. + validation.required_spec_claims.clear(); + validation.validate_exp = false; + validation.validate_aud = false; + let claims = decode::(token, &decoding_key, &validation) + .map_err(|_| invalid("JWT signature or format invalid"))?.claims; + let now = SystemTime::now().duration_since(UNIX_EPOCH).map_err(|_| invalid("System clock before epoch"))?.as_secs_f64(); + let exp = claims.get("exp").and_then(Value::as_f64).ok_or_else(|| invalid("Token expired"))?; + if !exp.is_finite() || exp < now { return Err(invalid("Token expired")); } + let iss = claims.get("iss").and_then(Value::as_str).ok_or_else(|| invalid("Issuer mismatch"))?; + if iss != format!("{}/p/{}", self.client.base, self.client.slug) { return Err(invalid("Issuer mismatch")); } + let aud = claims.get("aud").and_then(Value::as_str).ok_or_else(|| invalid("Audience mismatch"))?; + if aud.is_empty() || !self.api_key.starts_with(aud) { return Err(invalid("Audience mismatch")); } + Ok(claims) + } +} diff --git a/rust/src/auth/client.rs b/rust/src/auth/client.rs new file mode 100644 index 0000000..b2e7fd2 --- /dev/null +++ b/rust/src/auth/client.rs @@ -0,0 +1,166 @@ +use std::{sync::Arc, time::Duration}; +use serde_json::{json, Value}; +use tokio::sync::Mutex; +use crate::{Error, Result, http::{Http, segment}}; +use super::{LoginOutcome, Session}; + +/// Portable identity client. No browser, local storage, timers or WebAuthn runtime is supplied. +/// Clones share one in-memory session and serialize refreshes to avoid rotating a token twice. +#[derive(Clone)] +pub struct SentroyAuth { + pub(crate) http: Http, + pub(crate) base: String, + pub(crate) slug: String, + slug_encoded: String, + session: Arc>>, +} +impl std::fmt::Debug for SentroyAuth { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SentroyAuth").field("base", &self.base).field("slug", &self.slug) + .field("session", &"[REDACTED]").finish() + } +} +pub(crate) fn auth_error(e: Error) -> Error { + match e { + Error::Http { status, message: _, body } => Error::Auth { + code: body.get("error").and_then(Value::as_str).unwrap_or("http_error").to_string(), + description: body.get("error_description").and_then(Value::as_str) + .map(str::to_string).unwrap_or_else(|| format!("HTTP {status}")), + status: Some(status), + }, e => e, + } +} +impl SentroyAuth { + pub fn new(base: impl Into, project_slug: impl Into, api_key: Option, timeout: Duration) -> Result { + let base = base.into().trim_end_matches('/').to_string(); + let slug = project_slug.into(); + if slug.is_empty() { return Err(Error::InvalidInput("project slug must not be empty".into())); } + let slug_encoded = segment(&slug)?; + let url = url::Url::parse(&base).map_err(|e| Error::InvalidInput(e.to_string()))?; + if !matches!(url.scheme(), "https" | "http") || url.host_str().is_none() || url.query().is_some() || url.fragment().is_some() || !url.username().is_empty() || url.password().is_some() { + return Err(Error::InvalidInput("invalid auth base URL".into())); + } + let http = Http::new(base.clone(), api_key, timeout)?; + Ok(Self { http, base, slug, slug_encoded, session: Arc::new(Mutex::new(None)) }) + } + pub(crate) fn path(&self, p: &str) -> String { format!("/api/v1/auth/{}{}", self.slug_encoded, p) } + async fn post(&self, p: &str, body: Value) -> Result { + self.http.post(&self.path(p), body).await.map_err(auth_error) + } + async fn bearer(&self) -> Result { + let token = self.session.lock().await.as_ref().map(|s| s.access_token.clone()) + .ok_or_else(|| Error::InvalidInput("Not signed in — access token missing".into()))?; + self.http.with_token(Some(token)) + } + async fn get_auth(&self, p: &str) -> Result { + self.bearer().await?.get(&self.path(p), &[]).await.map_err(auth_error) + } + async fn post_auth(&self, p: &str, body: Value) -> Result { + self.bearer().await?.post(&self.path(p), body).await.map_err(auth_error) + } + async fn delete_auth(&self, p: &str) -> Result<()> { + let _: Value = self.bearer().await?.delete(&self.path(p), &[]).await.map_err(auth_error)?; + Ok(()) + } + async fn persist_login(&self, v: &Value) -> Result<()> { + let session: Session = serde_json::from_value(v.clone()).map_err(|e| Error::InvalidResponse(e.to_string()))?; + self.set_session(session).await; + Ok(()) + } + pub async fn session(&self) -> Option { self.session.lock().await.clone() } + pub async fn set_session(&self, session: Session) { *self.session.lock().await = Some(session); } + pub async fn clear_session(&self) { *self.session.lock().await = None; } + pub async fn sign_up(&self, input: Value) -> Result { + let v = self.post("/signup", input).await?; + if v.get("accessToken").and_then(Value::as_str).is_some() && v.get("refreshToken").and_then(Value::as_str).is_some() { self.persist_login(&v).await?; } + Ok(v) + } + pub async fn sign_in(&self, input: Value) -> Result { + let v = self.post("/login", input).await?; + if v.get("mfaRequired") == Some(&Value::Bool(true)) { Ok(LoginOutcome::Mfa(v)) } + else { self.persist_login(&v).await?; Ok(LoginOutcome::Tokens(v)) } + } + pub async fn verify_mfa(&self, input: Value) -> Result { + let v = self.post("/login/mfa/verify", input).await?; self.persist_login(&v).await?; Ok(v) + } + /// One refresh request, with no replay on transport or server failure. + /// An indeterminate failure requires application-specific reconciliation. + pub async fn refresh(&self) -> Result { + let mut guard = self.session.lock().await; + let old = guard.as_ref().ok_or_else(|| Error::InvalidInput("Not signed in".into()))?; + let v = self.post("/refresh", json!({"refreshToken": old.refresh_token})).await?; + let access = v.get("accessToken").and_then(Value::as_str).ok_or_else(|| Error::InvalidResponse("missing accessToken".into()))?; + let refresh = v.get("refreshToken").and_then(Value::as_str).ok_or_else(|| Error::InvalidResponse("missing refreshToken".into()))?; + let s = guard.as_mut().expect("session held during refresh"); + s.access_token = access.to_owned(); s.refresh_token = refresh.to_owned(); + Ok(v) + } + pub async fn sign_out(&self) { + let old = self.session.lock().await.take(); + if let Some(s) = old { let _ = self.post("/logout", json!({"refreshToken": s.refresh_token})).await; } + } + pub async fn send_password_reset(&self, email: &str) -> Result { self.post("/password-reset/request", json!({"email":email})).await } + pub async fn confirm_password_reset(&self, input: Value) -> Result { self.post("/password-reset/confirm", input).await } + pub async fn verify_email(&self, token: &str) -> Result { self.post("/verify-email", json!({"token":token})).await } + pub async fn send_magic_link(&self, input: Value) -> Result { self.post("/magic-link/request", input).await } + pub async fn consume_magic_link(&self, token: &str) -> Result { + let v = self.post("/magic-link/consume", json!({"token":token})).await?; self.persist_login(&v).await?; Ok(v) + } + pub async fn accept_invitation(&self, input: Value) -> Result { + let v = self.post("/invitation/accept", input).await?; self.persist_login(&v).await?; Ok(v) + } + /// The server's social URL has no state or PKCE parameters in the reference API. + /// Callers must secure the redirect at the application level. + pub fn social_authorize_url(&self, provider: &str, redirect_uri: Option<&str>, remember_me: bool) -> Result { + if !["google","github","facebook","microsoft","twitter","apple"].contains(&provider) { + return Err(Error::InvalidInput("unsupported social provider".into())); + } + let mut url = url::Url::parse(&format!("{}{}/social/{}/authorize", self.base, self.path(""), provider)) + .map_err(|e| Error::InvalidInput(e.to_string()))?; + { let mut q = url.query_pairs_mut(); if let Some(uri) = redirect_uri { q.append_pair("redirectUri", uri); } if remember_me { q.append_pair("rememberMe", "1"); } } + if url.query() == Some("") { url.set_query(None); } + Ok(url.to_string()) + } + pub async fn current_user(&self) -> Result { + let token = self.session.lock().await.as_ref().map(|s| s.access_token.clone()) + .ok_or_else(|| Error::InvalidInput("Not signed in — access token missing".into()))?; + let v = self.http.with_token(Some(token.clone()))?.get(&self.path("/me"), &[]).await.map_err(auth_error)?; + let mut guard = self.session.lock().await; + if let Some(s) = guard.as_mut() { if s.access_token == token { s.user = v.clone(); } } + Ok(v) + } + pub async fn list_sessions(&self) -> Result { self.get_auth("/me/sessions").await } + pub async fn revoke_session(&self, id: &str) -> Result<()> { self.delete_auth(&format!("/me/sessions/{}", segment(id)?)).await } + pub async fn change_password(&self, input: Value) -> Result { + let v = self.post_auth("/me/password", input).await?; self.clear_session().await; Ok(v) + } + pub async fn request_email_change(&self, input: Value) -> Result { self.post_auth("/me/email/change-request", input).await } + pub async fn confirm_email_change(&self, token: &str) -> Result { + let v = self.post("/me/email/change-confirm", json!({"token": token})).await?; self.clear_session().await; Ok(v) + } + pub async fn request_account_deletion(&self, password: &str) -> Result { self.post_auth("/me/account/delete-request", json!({"currentPassword":password})).await } + pub async fn confirm_account_deletion(&self, token: &str) -> Result { + let v = self.post("/me/account/delete-confirm", json!({"token":token})).await?; self.clear_session().await; Ok(v) + } + pub async fn activity(&self) -> Result { self.get_auth("/me/activity").await } + pub async fn mfa_status(&self) -> Result { self.get_auth("/me/mfa").await } + pub async fn enroll_totp(&self) -> Result { self.post_auth("/me/mfa/totp/enroll", Value::Null).await } + pub async fn verify_totp_enrollment(&self, code: &str) -> Result { self.post_auth("/me/mfa/totp/verify-enrollment", json!({"code":code})).await } + pub async fn disable_totp(&self, password: &str) -> Result { self.post_auth("/me/mfa/totp/disable", json!({"currentPassword":password})).await } + pub async fn list_passkeys(&self) -> Result { self.get_auth("/me/passkey").await } + pub async fn delete_passkey(&self, id: &str) -> Result<()> { self.delete_auth(&format!("/me/passkey/{}",segment(id)?)).await } + /// Returns WebAuthn options and challengeToken for the caller's native authenticator. + pub async fn begin_passkey_registration(&self) -> Result { self.post_auth("/me/passkey/register/begin", Value::Null).await } + pub async fn complete_passkey_registration(&self, challenge_token: &str, response: Value, device_name: Option<&str>) -> Result { + self.post_auth("/me/passkey/register/complete", json!({"challengeToken":challenge_token,"response":response,"deviceName":device_name})).await + } + pub async fn begin_passkey_authentication(&self, email: Option<&str>) -> Result { + let mut body = serde_json::Map::new(); if let Some(e) = email { body.insert("email".into(), json!(e)); } + self.post("/passkey/authenticate/begin", Value::Object(body)).await + } + pub async fn complete_passkey_authentication(&self, challenge_token: &str, response: Value, remember_me: Option) -> Result { + let mut body = json!({"challengeToken":challenge_token,"response":response}); + if let Some(r) = remember_me { body["rememberMe"] = json!(r); } + let v = self.post("/passkey/authenticate/complete", body).await?; self.persist_login(&v).await?; Ok(v) + } +} diff --git a/rust/src/auth/mod.rs b/rust/src/auth/mod.rs new file mode 100644 index 0000000..422cad2 --- /dev/null +++ b/rust/src/auth/mod.rs @@ -0,0 +1,8 @@ +//! Portable identity authentication. Applications own secure session storage, +//! navigation, callbacks and WebAuthn platform interactions. +mod client; +mod admin; +mod types; +pub use admin::SentroyAuthAdmin; +pub use client::SentroyAuth; +pub use types::*; diff --git a/rust/src/auth/types.rs b/rust/src/auth/types.rs new file mode 100644 index 0000000..84dd004 --- /dev/null +++ b/rust/src/auth/types.rs @@ -0,0 +1,27 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +/// A session is sensitive: persist it only in secure application-owned storage. +#[derive(Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Session { + pub access_token: String, + pub refresh_token: String, + pub user: Value, +} +impl std::fmt::Debug for Session { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Session").field("access_token", &"[REDACTED]") + .field("refresh_token", &"[REDACTED]").field("user", &self.user).finish() + } +} + +/// Login may require an MFA challenge before a session can be established. +#[derive(Clone, Serialize, Deserialize)] +#[serde(tag = "kind", content = "data", rename_all = "lowercase")] +pub enum LoginOutcome { Tokens(Value), Mfa(Value) } +impl std::fmt::Debug for LoginOutcome { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { Self::Tokens(_) => f.write_str("Tokens([REDACTED])"), Self::Mfa(_) => f.write_str("Mfa([REDACTED])") } + } +} diff --git a/rust/src/mcp/file_info.rs b/rust/src/mcp/file_info.rs new file mode 100644 index 0000000..6c7ec6d --- /dev/null +++ b/rust/src/mcp/file_info.rs @@ -0,0 +1,61 @@ +//! Local manifests for the two private upload workflows. Computing a manifest +//! never sends a request or publishes a file. +use sha2::{Digest, Sha256}; +use serde::{Deserialize, Serialize}; +use crate::Error; + +fn invalid(code: &str) -> Error { Error::Mcp(code.to_owned()) } + +fn digest(bytes: &[u8]) -> String { + let hash = Sha256::digest(bytes); + hash.iter().map(|byte| format!("{byte:02x}")).collect() +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ContactImportFileInfo { + pub format: String, + pub bytes: usize, + pub file_hash: String, +} + +/// Compute metadata for a contact upload. The body stays local. +pub fn sentroy_mcp_contact_import_file_info(body: &[u8], format: &str) -> Result { + if !matches!(format, "csv" | "jsonl") || body.is_empty() || body.len() > 1024 * 1024 { + return Err(invalid("invalid_mcp_contact_file")); + } + Ok(ContactImportFileInfo { format: format.to_owned(), bytes: body.len(), file_hash: digest(body) }) +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct StorageUploadFileInfo { + pub name: String, + pub mime_type: String, + pub bytes: usize, + pub sha256: String, +} + +fn valid_storage_name(name: &str, mime_type: &str) -> bool { + let extensions: &[&str] = match mime_type { + "application/pdf" => &["pdf"], "image/jpeg" => &["jpg", "jpeg"], + "image/png" => &["png"], "image/webp" => &["webp"], + "text/plain" => &["txt", "log"], "text/csv" => &["csv"], + "audio/mpeg" => &["mp3"], "audio/wav" => &["wav"], "audio/ogg" => &["ogg", "oga"], + "audio/mp4" => &["m4a", "mp4"], "audio/webm" => &["webm"], + "video/mp4" => &["mp4"], "video/webm" => &["webm"], "video/quicktime" => &["mov"], + _ => return false, + }; + if name.is_empty() || name.encode_utf16().count() > 200 || name.trim() != name + || name.chars().any(|c| c == '/' || c == '\\' || c.is_control()) { return false; } + let Some((stem, extension)) = name.rsplit_once('.') else { return false; }; + !stem.is_empty() && extensions.contains(&extension.to_ascii_lowercase().as_str()) +} + +/// Compute metadata for a storage upload. The body stays local. +pub fn sentroy_mcp_storage_upload_file_info(body: &[u8], name: &str, mime_type: &str) -> Result { + if !valid_storage_name(name, mime_type) || body.is_empty() || body.len() > 10 * 1024 * 1024 { + return Err(invalid("invalid_mcp_storage_upload")); + } + Ok(StorageUploadFileInfo { name: name.to_owned(), mime_type: mime_type.to_owned(), bytes: body.len(), sha256: digest(body) }) +} diff --git a/rust/src/mcp/mod.rs b/rust/src/mcp/mod.rs new file mode 100644 index 0000000..0488e05 --- /dev/null +++ b/rust/src/mcp/mod.rs @@ -0,0 +1,150 @@ +//! Native MCP client backed by the official Rust SDK. +//! +//! MCP credentials are a separate namespace from API keys. A connection never +//! acquires credentials, follows redirects, or replays requests on session expiry. + +use std::sync::Arc; +use rmcp::{ + service::{RoleClient, RunningService}, + transport::{ + common::client_side_sse::NeverRetry, + streamable_http_client::{StreamableHttpClientTransport, StreamableHttpClientTransportConfig}, + }, + ServiceExt, +}; +use url::Url; +use crate::Error; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum McpApp { Core, Mail, Storage, Linear, Chaby } + +impl McpApp { + fn origin(self) -> &'static str { + match self { + Self::Core => "https://sentroy.com", + Self::Mail => "https://mail.sentroy.com", + Self::Storage => "https://storage.sentroy.com", + Self::Linear => "https://linear.sentroy.com", + Self::Chaby => "https://chaby.sentroy.com", + } + } +} + +#[derive(Clone, Debug)] +pub struct McpResourceOptions { + pub app: McpApp, + pub company_slug: String, + /// Origin only. HTTPS, or HTTP on a loopback host. + pub origin: Option, +} + +fn invalid(code: &str) -> Error { Error::Mcp(code.to_owned()) } + +fn valid_slug(slug: &str) -> bool { + let bytes = slug.as_bytes(); + !bytes.is_empty() && bytes.len() <= 100 + && (bytes[0].is_ascii_lowercase() || bytes[0].is_ascii_digit()) + && bytes.iter().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || *c == b'_' || *c == b'-') +} + +/// Construct the protected resource URL without making any network requests. +pub fn sentroy_mcp_resource(options: &McpResourceOptions) -> Result { + if !valid_slug(&options.company_slug) { return Err(invalid("invalid_mcp_resource")); } + let origin = Url::parse(options.origin.as_deref().unwrap_or(options.app.origin())) + .map_err(|_| invalid("invalid_mcp_origin"))?; + let loopback = matches!(origin.host_str(), Some("localhost" | "127.0.0.1" | "[::1]")); + if !origin.username().is_empty() || origin.password().is_some() || origin.fragment().is_some() + || origin.query().is_some() || origin.path() != "/" + || !(origin.scheme() == "https" || (origin.scheme() == "http" && loopback)) + || origin.host().is_none() + { return Err(invalid("invalid_mcp_origin")); } + Ok(format!("{}/mcp/{}", origin.origin().ascii_serialization(), options.company_slug)) +} + +/// URL for OAuth protected-resource metadata; this does not perform discovery. +pub fn sentroy_mcp_resource_metadata(options: &McpResourceOptions) -> Result { + let resource = Url::parse(&sentroy_mcp_resource(options)?).map_err(|_| invalid("invalid_mcp_resource"))?; + Ok(format!("{}/.well-known/oauth-protected-resource{}", resource.origin().ascii_serialization(), resource.path())) +} + +fn valid_token(token: &str) -> bool { + let suffix = token.strip_prefix("mcp_at_").or_else(|| token.strip_prefix("mcp_")); + suffix.is_some_and(|s| s.len() == 43 && s.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')) +} + +/// Connected official protocol client. Use `client.peer()` to call MCP methods. +pub struct McpConnection { + pub resource: String, + pub client: RunningService, + app: McpApp, + token: String, + file_client: reqwest::Client, +} + +/// Connect with a personal MCP token or externally managed MCP OAuth access +/// token. This performs the protocol handshake, but no tool calls. +pub async fn connect_sentroy_mcp(options: &McpResourceOptions, token: &str) -> Result { + let resource = sentroy_mcp_resource(options)?; + if !valid_token(token) { return Err(invalid("invalid_mcp_token")); } + let file_client = reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .retry(reqwest::retry::never()) + .build().map_err(|_| invalid("mcp_connection_failed"))?; + let client = reqwest13::Client::builder() + .redirect(reqwest13::redirect::Policy::none()) + .retry(reqwest13::retry::never()) + .build().map_err(|_| invalid("mcp_connection_failed"))?; + let mut config = StreamableHttpClientTransportConfig::with_uri(resource.clone()) + .auth_header(token) + .reinit_on_expired_session(false) + .max_sse_event_size(1024 * 1024); + config.retry_config = Arc::new(NeverRetry::default()); + let transport = StreamableHttpClientTransport::with_client(client, config); + let service = ().serve(transport).await.map_err(|_| invalid("mcp_connection_failed"))?; + Ok(McpConnection { resource, client: service, app: options.app, token: token.to_owned(), file_client }) +} + +mod transfers; +pub use transfers::TransferOptions; +mod file_info; +pub use file_info::{ContactImportFileInfo, StorageUploadFileInfo, sentroy_mcp_contact_import_file_info, sentroy_mcp_storage_upload_file_info}; + +#[cfg(test)] +mod tests { + use super::*; + fn resource() -> McpResourceOptions { + McpResourceOptions { app: McpApp::Mail, company_slug: "example".into(), origin: None } + } + #[test] + fn resource_and_metadata_bind_slug_and_origin() { + let mut opts = resource(); + assert_eq!(sentroy_mcp_resource(&opts).unwrap(), "https://mail.sentroy.com/mcp/example"); + assert_eq!(sentroy_mcp_resource_metadata(&opts).unwrap(), "https://mail.sentroy.com/.well-known/oauth-protected-resource/mcp/example"); + opts.origin = Some("http://localhost:8080".into()); + assert_eq!(sentroy_mcp_resource(&opts).unwrap(), "http://localhost:8080/mcp/example"); + for bad in ["http://example.com", "https://user:pass@example.com", "https://example.com/a", "https://example.com/?q=1", "https://example.com/#f"] { + opts.origin = Some(bad.into()); + assert!(sentroy_mcp_resource(&opts).is_err()); + } + opts.origin = None; + for bad in ["", "A", "../x", "a/b", "with space"] { + opts.company_slug = bad.into(); + assert!(sentroy_mcp_resource(&opts).is_err()); + } + } + #[test] + fn token_namespace_and_manifests() { + assert!(valid_token(&format!("mcp_{}", "a".repeat(43)))); + assert!(valid_token(&format!("mcp_at_{}", "a".repeat(43)))); + assert!(!valid_token(&format!("stk_{}", "a".repeat(43)))); + assert!(!valid_token(&format!("mcp_{}", "a".repeat(42)))); + let contact = sentroy_mcp_contact_import_file_info(b"abc", "csv").unwrap(); + assert_eq!(contact.file_hash, "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"); + assert!(sentroy_mcp_contact_import_file_info(b"", "csv").is_err()); + assert!(sentroy_mcp_contact_import_file_info(b"abc", "txt").is_err()); + let file = sentroy_mcp_storage_upload_file_info(b"abc", "file.PDF", "application/pdf").unwrap(); + assert_eq!(file.sha256, contact.file_hash); + assert!(sentroy_mcp_storage_upload_file_info(b"abc", "../file.pdf", "application/pdf").is_err()); + assert!(sentroy_mcp_storage_upload_file_info(b"abc", "file.txt", "application/pdf").is_err()); + } +} diff --git a/rust/src/mcp/transfers.rs b/rust/src/mcp/transfers.rs new file mode 100644 index 0000000..e717ba3 --- /dev/null +++ b/rust/src/mcp/transfers.rs @@ -0,0 +1,252 @@ +//! Bound private-file transfers. References are validated before any request. +//! Upload transport failures have an unknown outcome; callers must reconcile +//! state through the appropriate MCP tool before deciding whether to retry. +use std::collections::HashMap; +use std::time::Duration; +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use chrono::{DateTime, SecondsFormat, Utc}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use url::Url; +use super::{McpApp, McpConnection}; +use crate::Error; + +const CONTACT_CAP: usize = 1024 * 1024; +const STORAGE_CAP: usize = 10 * 1024 * 1024; +const DOWNLOAD_CAP: usize = 64 * 1024 * 1024; +fn err(s: &str) -> Error { Error::Mcp(s.into()) } +fn field<'a>(v: &'a Value, key: &str) -> Result<&'a str, Error> { + v.get(key).and_then(Value::as_str).ok_or_else(|| err("invalid_mcp_reference")) +} +fn size(v: &Value, key: &str, cap: usize) -> Result { + v.get(key).and_then(Value::as_u64).and_then(|n| usize::try_from(n).ok()) + .filter(|n| *n <= cap).ok_or_else(|| err("invalid_mcp_reference")) +} +fn hex(s: &str, len: usize) -> bool { + s.len() == len && s.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} +fn slug(s: &str) -> bool { + !s.is_empty() && s.len() <= 100 && s.as_bytes()[0].is_ascii_alphanumeric() + && s.bytes().all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-') + && (s.as_bytes()[0].is_ascii_lowercase() || s.as_bytes()[0].is_ascii_digit()) +} +fn uuid(s: &str) -> bool { + let b = s.as_bytes(); + b.len() == 36 && [8,13,18,23].iter().all(|i| b[*i] == b'-') + && b.iter().enumerate().all(|(i,c)| [8,13,18,23].contains(&i) || c.is_ascii_digit() || (b'a'..=b'f').contains(c)) + && (b'1'..=b'8').contains(&b[14]) && matches!(b[19], b'8' | b'9' | b'a' | b'b') +} +fn date(s: &str, future: bool) -> bool { + let Ok(d) = DateTime::parse_from_rfc3339(s) else { return false; }; + d.to_utc().to_rfc3339_opts(SecondsFormat::Millis, true) == s && (!future || d > Utc::now()) +} +fn sha(bytes: &[u8]) -> String { Sha256::digest(bytes).iter().map(|b| format!("{b:02x}")).collect() } +fn flag(v: &Value, key: &str, b: bool) -> bool { v.get(key).and_then(Value::as_bool) == Some(b) } + +fn url_for(connection: &McpConnection, raw: &str, max: usize) -> Result { + if raw.len() > max { return Err(err("invalid_mcp_reference")); } + let url = Url::parse(raw).map_err(|_| err("invalid_mcp_reference"))?; + let resource = Url::parse(&connection.resource).map_err(|_| err("invalid_mcp_reference"))?; + if url.origin() != resource.origin() || !url.username().is_empty() || url.password().is_some() || url.fragment().is_some() { + return Err(err("invalid_mcp_reference")); + } + Ok(url) +} +fn query(url: &Url, keys: &[&str]) -> Result, Error> { + let pairs: Vec<_> = url.query_pairs().collect(); + if pairs.len() != keys.len() { return Err(err("invalid_mcp_reference")); } + let mut out = HashMap::new(); + for (k,v) in pairs { + if !keys.contains(&k.as_ref()) || out.insert(k.into_owned(), v.into_owned()).is_some() { + return Err(err("invalid_mcp_reference")); + } + } + if !out.get("connection").is_some_and(|s| hex(s,24)) { return Err(err("invalid_mcp_reference")); } + Ok(out) +} +fn resource_path(c: &McpConnection) -> Result { + Ok(Url::parse(&c.resource).map_err(|_| err("invalid_mcp_resource"))?.path().to_owned()) +} +fn contact_ref(c: &McpConnection, v: &Value, kind: &str) -> Result<(Url,usize,String,String), Error> { + if c.app != McpApp::Mail { return Err(err("mcp_contact_files_not_supported")); } + if !flag(v, "requiresOriginalBearer", true) { return Err(err("invalid_mcp_reference")); } + let bytes = size(v, "bytes", CONTACT_CAP)?; + let id = field(v, if kind == "exports" { "exportId" } else { "uploadId" })?; + let hash = field(v,"fileHash")?; + let expires = field(v,"expiresAt")?; + if !uuid(id) || !hex(hash,64) || !date(expires,true) || (kind == "imports" && bytes == 0) { + return Err(err("invalid_mcp_reference")); + } + let u = url_for(c, field(v,if kind == "exports" { "downloadUrl" } else { "uploadUrl" })?,4000)?; + if u.path() != format!("{}/contacts/{kind}/{id}",resource_path(c)?) { return Err(err("invalid_mcp_reference")); } + let q = query(&u, if kind == "exports" { &["connection","hash"] } else { &["connection"] })?; + if kind == "exports" && q.get("hash").map(String::as_str) != Some(hash) { return Err(err("invalid_mcp_reference")); } + Ok((u,bytes,hash.to_owned(),expires.to_owned())) +} +/// Transfer timeout covers the complete request and body; defaults are 60s +/// for downloads and 15s for uploads. Maximum is 120s. +#[derive(Clone, Copy, Debug, Default)] +pub struct TransferOptions { pub timeout_ms: Option } +fn timeout(o: Option, default: u64) -> Result { + let ms = o.and_then(|v| v.timeout_ms).unwrap_or(default); + if !(1..=120_000).contains(&ms) { return Err(err("invalid_mcp_timeout")); } + Ok(Duration::from_millis(ms)) +} +async fn bounded(mut response: reqwest::Response, cap: usize, expected: Option) -> Result, Error> { + if let Some(length) = response.content_length() { + if length > cap as u64 || expected.is_some_and(|n| n as u64 != length) { return Err(err("mcp_file_size_mismatch")); } + } + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(|_| err("mcp_download_failed"))? { + if chunk.len() > cap.saturating_sub(bytes.len()) || expected.is_some_and(|n| chunk.len() > n.saturating_sub(bytes.len())) { + return Err(err("mcp_file_size_mismatch")); + } + bytes.extend_from_slice(&chunk); + } + if expected.is_some_and(|n| n != bytes.len()) { return Err(err("mcp_file_size_mismatch")); } + Ok(bytes) +} +impl McpConnection { + async fn get_file(&self, u: Url, cap: usize, expected: Option, duration: Duration) -> Result, Error> { + tokio::time::timeout(duration, async { + let r = self.file_client.get(u).bearer_auth(&self.token).send().await.map_err(|_| err("mcp_download_failed"))?; + if r.status() != 200 { return Err(err("mcp_download_failed")); } + bounded(r,cap,expected).await + }).await.map_err(|_| err("mcp_download_aborted"))? + } + async fn put_file(&self, u: Url, body: &[u8], duration: Duration) -> Result, Error> { + // Once send starts, timeout, connection loss, or invalid response cannot + // determine whether the server accepted the bytes. Never replay it. + tokio::time::timeout(duration, async { + let r = self.file_client.put(u).bearer_auth(&self.token) + .header(reqwest::header::CONTENT_TYPE,"application/octet-stream").body(body.to_vec()) + .send().await.map_err(|_| err("mcp_upload_outcome_unknown"))?; + if r.status() != 200 { return Err(err("mcp_upload_failed")); } + bounded(r,4096,None).await.map_err(|_| err("mcp_upload_outcome_unknown")) + }).await.map_err(|_| err("mcp_upload_outcome_unknown"))? + } + + /// Download a storage file, bound to the file ID, version and connection. + pub async fn download_file(&self, v: &Value, options: Option) -> Result,Error> { + if self.app != McpApp::Storage { return Err(err("mcp_download_not_supported")); } + if !flag(v,"requiresAuthorization",true) || field(v,"method")? != "GET" { return Err(err("invalid_mcp_reference")); } + let file = v.get("file").ok_or_else(|| err("invalid_mcp_reference"))?; + let id = field(file,"id")?; let n = size(file,"size",DOWNLOAD_CAP)?; + if !hex(id,24) { return Err(err("invalid_mcp_reference")); } + let u = url_for(self,field(v,"downloadUrl")?,4000)?; + let prefix = format!("{}/files/",resource_path(self)?); + let path = u.path().strip_prefix(&prefix).ok_or_else(|| err("invalid_mcp_reference"))?; + let parts: Vec<_> = path.split('/').collect(); + let q = query(&u,&["connection","version"])?; + if parts.len()!=2 || !slug(parts[0]) || parts[1]!=id || !q.get("version").is_some_and(|s| date(s,false)) { + return Err(err("invalid_mcp_reference")); + } + self.get_file(u,DOWNLOAD_CAP,Some(n),timeout(options,60_000)?).await + } + + /// Download a mail attachment after validating its encoded message reference. + pub async fn download_mail_attachment(&self, v: &Value, options: Option) -> Result,Error> { + if self.app != McpApp::Mail { return Err(err("mcp_download_not_supported")); } + if !flag(v,"requiresAuthorization",true) || field(v,"method")? != "GET" { return Err(err("invalid_mcp_reference")); } + let max = size(v,"maxBytes",8*1024*1024)?; + let a = v.get("attachment").ok_or_else(|| err("invalid_mcp_reference"))?; + size(a,"encodedSize",12*1024*1024)?; + let part = field(a,"partId")?; + if max == 0 || part.len()>80 || part.is_empty() || part.split('.').any(|p| p.is_empty() || !p.bytes().all(|b| b.is_ascii_digit())) { + return Err(err("invalid_mcp_reference")); + } + let u = url_for(self,field(v,"downloadUrl")?,5000)?; + let prefix = format!("{}/attachments/",resource_path(self)?); + let encoded = u.path().strip_prefix(&prefix).ok_or_else(|| err("invalid_mcp_reference"))?; + if encoded.is_empty() || encoded.len()>2500 || !encoded.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-') { + return Err(err("invalid_mcp_reference")); + } + let q=query(&u,&["connection","version"])?; + if !q.get("version").is_some_and(|s|hex(s,64)) { return Err(err("invalid_mcp_reference")); } + let data=URL_SAFE_NO_PAD.decode(encoded).map_err(|_|err("invalid_mcp_reference"))?; + let decoded: Value=serde_json::from_slice(&data).map_err(|_|err("invalid_mcp_reference"))?; + let obj=decoded.as_object().ok_or_else(||err("invalid_mcp_reference"))?; + let reference=decoded.get("reference").and_then(Value::as_object).ok_or_else(||err("invalid_mcp_reference"))?; + if obj.len()!=2 || decoded.get("partId").and_then(Value::as_str)!=Some(part) || reference.len()!=5 + || !["companyId","mailbox","folder","uidValidity","uid"].iter().all(|k|reference.contains_key(*k)) { + return Err(err("invalid_mcp_reference")); + } + let r=decoded.get("reference").ok_or_else(||err("invalid_mcp_reference"))?; + let company=field(r,"companyId")?; let mailbox=field(r,"mailbox")?; + let folder=field(r,"folder")?; let validity=field(r,"uidValidity")?; + let uid=r.get("uid").and_then(Value::as_u64).unwrap_or(0); + if company.is_empty() || company.len()>100 || company.contains(['\r','\n','\0']) + || mailbox.len()>320 || mailbox.split('@').count()!=2 || mailbox.split('@').any(|s| s.is_empty() || s.chars().any(char::is_whitespace)) + || folder.is_empty() || folder.len()>512 || folder.starts_with("__") || folder.contains(['\r','\n','\0']) + || validity.is_empty() || validity.len()>20 || validity.starts_with('0') || !validity.bytes().all(|b|b.is_ascii_digit()) + || uid==0 || uid>u32::MAX as u64 { return Err(err("invalid_mcp_reference")); } + self.get_file(u,max,None,timeout(options,60_000)?).await + } + + /// Download and hash-check an original-grant contacts export. + pub async fn download_mail_contacts_export(&self, v: &Value, options: Option) -> Result,Error> { + let (u,n,hash,_) = contact_ref(self,v,"exports")?; + if !matches!(field(v,"format")?, "csv" | "jsonl") { return Err(err("invalid_mcp_reference")); } + let data=self.get_file(u,CONTACT_CAP,Some(n),timeout(options,60_000)?).await?; + if sha(&data)!=hash { return Err(err("mcp_contact_file_hash_mismatch")); } + Ok(data) + } + + /// Upload an original-grant contacts import. A transport failure returns + /// `mcp_upload_outcome_unknown`; query its state before any further action. + pub async fn upload_mail_contacts_import(&self, v: &Value, body: &[u8], options: Option) -> Result { + let (u,n,hash,expires)=contact_ref(self,v,"imports")?; + if body.len()!=n { return Err(err("mcp_contact_file_size_mismatch")); } + if sha(body)!=hash { return Err(err("mcp_contact_file_hash_mismatch")); } + if !date(&expires,true) { return Err(err("invalid_mcp_reference")); } + let data=self.put_file(u,body,timeout(options,15_000)?).await?; + let result: Value=serde_json::from_slice(&data).map_err(|_|err("mcp_upload_outcome_unknown"))?; + if result.get("uploadId")!=v.get("uploadId") || result.get("state").and_then(Value::as_str)!=Some("ready") + || result.get("fileHash").and_then(Value::as_str)!=Some(hash.as_str()) + || result.get("expiresAt").and_then(Value::as_str)!=Some(expires.as_str()) + || !result.get("count").and_then(Value::as_u64).is_some_and(|n|(1..=100).contains(&n)) + || !date(&expires,true) { return Err(err("mcp_upload_outcome_unknown")); } + Ok(json!({"uploadId":result["uploadId"],"state":"ready","count":result["count"],"fileHash":hash,"expiresAt":expires})) + } + + /// Upload a storage file under the original grant. A transport failure + /// returns `mcp_upload_outcome_unknown` and is never automatically retried. + pub async fn upload_storage_file(&self, v: &Value, body: &[u8], options: Option) -> Result { + if self.app != McpApp::Storage { return Err(err("mcp_storage_upload_not_supported")); } + let id=field(v,"uploadId")?; let name=field(v,"name")?; let mime=field(v,"mimeType")?; + let n=size(v,"bytes",STORAGE_CAP)?; let hash=field(v,"sha256")?; let expires=field(v,"expiresAt")?; + let bucket=v.get("bucket").ok_or_else(||err("invalid_mcp_reference"))?; + if !uuid(id) || !hex(hash,64) || !hex(field(v,"revision")?,64) || !date(expires,true) + || !matches!(field(v,"state")?, "waiting" | "ready") || !flag(v,"published",false) + || !flag(v,"malwareScanned",false) || !flag(v,"requiresHumanApproval",true) + || !flag(v,"requiresOriginalBearer",true) || field(v,"method")?!="PUT" + || field(v,"contentType")?!="application/octet-stream" + || !hex(field(bucket,"id")?,24) || !slug(field(bucket,"slug")?) || !hex(field(bucket,"revision")?,64) { + return Err(err("invalid_mcp_reference")); + } + let info=super::sentroy_mcp_storage_upload_file_info(body,name,mime)?; + if info.bytes!=n { return Err(err("mcp_storage_upload_size_mismatch")); } + if info.sha256!=hash { return Err(err("mcp_storage_upload_hash_mismatch")); } + let u=url_for(self,field(v,"uploadUrl")?,4000)?; + if u.path()!=format!("{}/uploads/{id}",resource_path(self)?) { return Err(err("invalid_mcp_reference")); } + query(&u,&["connection"])?; + if !date(expires,true) { return Err(err("invalid_mcp_reference")); } + let data=self.put_file(u,body,timeout(options,15_000)?).await?; + let result: Value=serde_json::from_slice(&data).map_err(|_|err("mcp_upload_outcome_unknown"))?; + let rb=result.get("bucket").ok_or_else(||err("mcp_upload_outcome_unknown"))?; + for key in ["uploadId","name","mimeType","bytes","sha256","expiresAt"] { + if result.get(key)!=v.get(key) { return Err(err("mcp_upload_outcome_unknown")); } + } + for key in ["id","slug","revision"] { + if rb.get(key)!=bucket.get(key) { return Err(err("mcp_upload_outcome_unknown")); } + } + if result.get("state").and_then(Value::as_str)!=Some("ready") + || !result.get("revision").and_then(Value::as_str).is_some_and(|s|hex(s,64)) + || !flag(&result,"published",false) || !flag(&result,"malwareScanned",false) + || !flag(&result,"requiresHumanApproval",true) || !date(expires,true) { + return Err(err("mcp_upload_outcome_unknown")); + } + Ok(json!({"uploadId":id,"name":name,"mimeType":mime,"bytes":n,"sha256":hash,"expiresAt":expires, + "bucket":bucket,"state":"ready","revision":result["revision"],"published":false,"malwareScanned":false,"requiresHumanApproval":true})) + } +} diff --git a/rust/src/resources/mail.rs b/rust/src/resources/mail.rs new file mode 100644 index 0000000..039e8ef --- /dev/null +++ b/rust/src/resources/mail.rs @@ -0,0 +1,176 @@ +//! Mail resources. +use serde::Serialize; +use serde_json::{json, Value}; +use crate::{Error, Result}; +use crate::http::{Http, segment, query}; +use crate::types::mail::*; + +fn body(v: &T) -> Result { Ok(serde_json::to_value(v)?) } +fn path(base: &str, id: &str) -> Result { Ok(format!("{}/{}", base, segment(id)?)) } +fn q(v: &T) -> Result> { query(v) } +fn nonempty_bcc(value: &Option) -> bool { + match value { Some(Addresses::One(s)) => !s.trim().is_empty(), Some(Addresses::Many(v)) => !v.is_empty(), None => false } +} +fn request_id(value: &Option) -> Result { + match value { + None => Ok(uuid::Uuid::new_v4().to_string()), + Some(v) if v.is_empty() => Ok(uuid::Uuid::new_v4().to_string()), + Some(v) => { + let parsed = uuid::Uuid::parse_str(v).map_err(|_| Error::InvalidInput("requestId must be a UUID".into()))?; + let canonical = parsed.to_string(); + if !canonical.eq_ignore_ascii_case(v) || !(1..=8).contains(&(parsed.as_bytes()[6] >> 4)) || (parsed.as_bytes()[8] & 0xc0) != 0x80 { + return Err(Error::InvalidInput("requestId must be a UUID".into())); + } + Ok(canonical) + } + } +} +async fn compose(http: &Http, endpoint: &str, mut data: Value, id: String) -> Result { + data["requestId"] = json!(id); + http.post_once(endpoint, data).await.map_err(|source| Error::ComposeRequest { request_id:id, source:Box::new(source) }) +} +macro_rules! resource { + ($name:ident) => { + #[derive(Clone, Debug)] + pub struct $name { http: Http } + impl $name { pub(crate) fn new(http: Http) -> Self { Self { http } } } + } +} +resource!(Domains); +impl Domains { + pub async fn list(&self) -> Result> { self.http.get("/domains", &[]).await } + pub async fn get(&self, id:&str) -> Result { self.http.get(&path("/domains", id)?, &[]).await } +} +resource!(Mailboxes); +impl Mailboxes { + pub async fn list(&self) -> Result> { self.http.get("/mailboxes", &[]).await } +} +resource!(Templates); +impl Templates { + pub async fn list(&self, params:&TemplateListParams) -> Result> { self.http.get("/templates", &q(params)?).await } + pub async fn get(&self, id:&str) -> Result