diff --git a/.repository-projection.json b/.repository-projection.json index 6059c438a..2b562df31 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "43eb0edd715c489c97e0fb1fe22f52a44dbcfd95", + "sourceSha": "dc2633f387866330906a8eb01a53a05f29ddb6a4", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "44c5b0b9c35bc88ea604bae376b1a1b9c4d0d7a5", + "priorProjectedBase": "1a45f8314db3701fad50948cc68c11393130024f", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "45502ff0478e541d02f34cc39ac332935f8d3c0030fa221afe5bcc3b5e51f88e", - "contentDigest": "fd67fb0c9b781bf5e60fd477ccb05c531bb038c0c4bcb52ccf50947d8c5bcd1e", + "contentDigest": "696812e7a6f2c121810ace9a63cf0def9e81291515a6a367a08d4639e0b630de", "publicationEligible": true } diff --git a/packages/local-host-rs/src/credential_mode.rs b/packages/local-host-rs/src/credential_mode.rs index 831d5ca2b..5df4ec75d 100644 --- a/packages/local-host-rs/src/credential_mode.rs +++ b/packages/local-host-rs/src/credential_mode.rs @@ -157,6 +157,18 @@ fn test_identity_introspect_response(request: &str) -> (u16, &'static str) { (!token.is_empty()).then_some(token) }); match token { + Some("desktop-hosted-token") => ( + 200, + r#"{"active":true,"subject":"user-test","token_type":"access","organization_id":"org-test","workspace_id":"workspace-test","scopes":["llm_gateway:invoke","console:read","console:write"]}"#, + ), + Some("desktop-readonly-token") => ( + 200, + r#"{"active":true,"subject":"user-test","token_type":"access","organization_id":"org-test","workspace_id":"workspace-test","scopes":["llm_gateway:invoke","console:read"]}"#, + ), + Some("desktop-writeonly-token") => ( + 200, + r#"{"active":true,"subject":"user-test","token_type":"access","organization_id":"org-test","workspace_id":"workspace-test","scopes":["llm_gateway:invoke","console:write"]}"#, + ), Some("inactive-token") => ( 200, r#"{"active":false,"subject":"user-test","token_type":"access","organization_id":"org-test","workspace_id":"workspace-test","scopes":["llm_gateway:invoke"]}"#, @@ -431,7 +443,7 @@ fn current_verified_identity_session_with_env() false, ); } - let identity = verify_live_runtime_identity(snapshot.as_ref(), &env)?; + let identity = verify_live_runtime_identity(snapshot.as_ref(), &env, &[])?; Ok((identity, env)) } @@ -448,6 +460,41 @@ pub fn verified_current_identity_session() -> Result { verify_live_identity_session(snapshot.as_ref(), &env) } +/// Verify a desktop-held access token without reading or replacing the CLI +/// credential. The expected tenant is compared after live introspection. +pub fn verified_desktop_identity_session( + access_token: &str, + organization_id: &str, + workspace_id: &str, +) -> Result { + crate::safety::require_vendor_network()?; + if access_token.trim().is_empty() + || organization_id.trim().is_empty() + || workspace_id.trim().is_empty() + { + bail!("desktop Identity credential is incomplete"); + } + let mut env = std::env::vars() + .filter(|(key, _)| { + matches!( + key.as_str(), + "MAESTRO_IDENTITY_URL" | "EVALOPS_IDENTITY_URL" + ) || key == crate::init_cli::TEST_IDENTITY_AUTHORITY_ENV + }) + .collect::>(); + env.insert(ACCESS_TOKEN_ENV.to_owned(), access_token.to_owned()); + env.insert(ORG_ID_ENV.to_owned(), organization_id.to_owned()); + env.insert(WORKSPACE_ID_ENV.to_owned(), workspace_id.to_owned()); + let session = verify_live_runtime_identity(None, &env, &["console:read", "console:write"])? + .require_human()?; + if session.organization_id != organization_id + || session.workspace_id.as_deref() != Some(workspace_id) + { + bail!("desktop Identity tenant differs from verified token"); + } + Ok(session) +} + /// Return a replacement verified session only when the credential backing a /// long-running capture hook changed or reached its refresh window. /// @@ -595,12 +642,13 @@ fn verify_live_identity_session( snapshot: Option<&EvalOpsCredentialSnapshot>, env: &HashMap, ) -> Result { - verify_live_runtime_identity(snapshot, env)?.require_human() + verify_live_runtime_identity(snapshot, env, &[])?.require_human() } fn verify_live_runtime_identity( snapshot: Option<&EvalOpsCredentialSnapshot>, env: &HashMap, + additional_scopes: &[&str], ) -> Result { let Some(unverified) = platform_session_from(snapshot, env) else { bail!("{IDENTITY_REQUIRED_MESSAGE}"); @@ -653,6 +701,16 @@ fn verify_live_runtime_identity( .join() .map_err(|_| anyhow::anyhow!("EvalOps Identity verification thread panicked"))? .with_context(|| IDENTITY_REQUIRED_MESSAGE.to_owned())?; + if additional_scopes.iter().any(|required| { + !introspection + .scopes + .iter() + .map(String::as_str) + .chain(introspection.scope.split_whitespace()) + .any(|present| present == *required) + }) { + bail!("desktop Identity token lacks required hosted thread scope"); + } verified_runtime_identity(unverified, introspection, hosted) } @@ -732,6 +790,7 @@ pub(crate) fn verified_platform_session_for_scope( // select a tenant different from the signed Identity token. session.organization_id = organization_id.expect("checked above").to_owned(); session.workspace_id = workspace_id.map(str::to_owned); + session.user_id = Some(introspection.subject); Ok(session) } @@ -1725,6 +1784,53 @@ mod tests { ); } + #[test] + fn desktop_credential_is_live_verified_and_cannot_select_another_tenant() { + let _guard = crate::config::test_process_env_lock(); + let _restore = EnvRestore::capture(&[ + "MAESTRO_IDENTITY_URL", + crate::init_cli::TEST_IDENTITY_AUTHORITY_ENV, + ]); + std::env::set_var("MAESTRO_IDENTITY_URL", test_identity_base_url()); + std::env::set_var(crate::init_cli::TEST_IDENTITY_AUTHORITY_ENV, "1"); + let session = + verified_desktop_identity_session("desktop-hosted-token", "org-test", "workspace-test") + .expect("matching desktop token"); + assert_eq!(session.user_id.as_deref(), Some("user-test")); + assert!( + verified_desktop_identity_session( + "desktop-hosted-token", + "other-org", + "workspace-test" + ) + .is_err() + ); + assert!( + verified_desktop_identity_session( + "desktop-hosted-token", + "org-test", + "other-workspace" + ) + .is_err() + ); + assert!( + verified_desktop_identity_session("valid-token", "org-test", "workspace-test").is_err() + ); + for token in ["desktop-readonly-token", "desktop-writeonly-token"] { + assert!( + verified_desktop_identity_session(token, "org-test", "workspace-test").is_err() + ); + } + assert!( + verified_desktop_identity_session("inactive-token", "org-test", "workspace-test") + .is_err() + ); + assert!( + verified_desktop_identity_session("unscoped-token", "org-test", "workspace-test") + .is_err() + ); + } + #[test] fn live_introspection_rejects_inactive_and_unscoped_tokens() { let _guard = crate::config::test_process_env_lock(); diff --git a/packages/local-host-rs/src/hosted_thread.rs b/packages/local-host-rs/src/hosted_thread.rs new file mode 100644 index 000000000..dc576e43d --- /dev/null +++ b/packages/local-host-rs/src/hosted_thread.rs @@ -0,0 +1,449 @@ +//! Tenant-scoped Platform operating-thread client shared by the terminal and desktop gateway. + +use crate::credential_mode::PlatformSession; +use anyhow::{Context, Result, bail}; +use prost::Message; +use reqwest::{Client, Url}; +use std::time::Duration; +use uuid::Uuid; + +const SERVICE: &str = "/deixic.v1.DeixicService"; +const MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024; + +// A narrow wire projection of console.v1. Field numbers and kinds mirror the +// canonical protobuf contract; unknown fields are ignored by prost. +#[derive(Clone, PartialEq, Message)] +pub struct Query { + #[prost(string, tag = "1")] + pub workspace_id: String, + #[prost(string, tag = "13")] + pub organization_id: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct GetRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(string, tag = "2")] + pub channel_id: String, + #[prost(int32, tag = "3")] + pub limit: i32, +} + +#[derive(Clone, PartialEq, Message)] +pub struct Channel { + #[prost(string, tag = "1")] + pub id: String, + #[prost(string, tag = "2")] + pub label: String, + #[prost(int32, tag = "5")] + pub unread_count: i32, + #[prost(int32, tag = "6")] + pub open_count: i32, + #[prost(bool, tag = "11")] + pub archived: bool, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ListChannelsRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(int32, tag = "2")] + pub archive_filter: i32, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ListChannelsResponse { + #[prost(message, repeated, tag = "1")] + pub channels: Vec, +} + +#[derive(Clone, PartialEq, Message)] +pub struct RenameThreadRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(string, tag = "2")] + pub channel_id: String, + #[prost(string, tag = "3")] + pub title: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct RenameThreadResponse { + #[prost(message, optional, tag = "1")] + pub channel: Option, + #[prost(bool, tag = "2")] + pub changed: bool, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ArchiveThreadRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(string, tag = "2")] + pub channel_id: String, + #[prost(bool, tag = "3")] + pub archived: bool, + #[prost(string, tag = "4")] + pub idempotency_key: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ArchiveThreadResponse { + #[prost(message, optional, tag = "1")] + pub channel: Option, + #[prost(bool, tag = "2")] + pub changed: bool, + #[prost(bool, tag = "3")] + pub replayed: bool, +} + +#[derive(Clone, PartialEq, Message)] +pub struct OperatingMessage { + #[prost(string, tag = "1")] + pub id: String, + #[prost(string, tag = "3")] + pub role: String, + #[prost(string, tag = "5")] + pub body: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct GetResponse { + #[prost(message, optional, tag = "1")] + pub channel: Option, + #[prost(message, repeated, tag = "2")] + pub messages: Vec, + #[prost(int64, tag = "7")] + pub replay_cursor: i64, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ListRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(string, tag = "2")] + pub channel_id: String, + #[prost(int64, tag = "3")] + pub after_cursor: i64, + #[prost(int32, tag = "4")] + pub limit: i32, +} + +#[derive(Clone, PartialEq, Message)] +pub struct Event { + #[prost(int64, tag = "1")] + pub cursor: i64, + #[prost(string, tag = "2")] + pub event_id: String, + #[prost(string, tag = "3")] + pub turn_id: String, + #[prost(int32, tag = "4")] + pub kind: i32, + #[prost(string, tag = "5")] + pub safe_text: String, + #[prost(string, tag = "9")] + pub request_id: String, + #[prost(int32, tag = "10")] + pub request_type: i32, + #[prost(string, tag = "12")] + pub request_call_id: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ListResponse { + #[prost(message, repeated, tag = "1")] + pub events: Vec, + #[prost(int64, tag = "2")] + pub next_cursor: i64, + #[prost(bool, tag = "3")] + pub has_more: bool, + #[prost(bool, tag = "4")] + pub reset_required: bool, +} + +#[derive(Clone, PartialEq, Message)] +pub struct SubmitRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(string, tag = "2")] + pub channel_id: String, + #[prost(string, tag = "3")] + pub body: String, + #[prost(string, tag = "4")] + pub idempotency_key: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct Turn { + #[prost(string, tag = "1")] + pub turn_id: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct SubmitResponse { + #[prost(message, optional, tag = "6")] + pub accepted_turn: Option, + #[prost(int64, tag = "7")] + pub replay_cursor: i64, +} + +#[derive(Clone, PartialEq, Message)] +pub struct ThreadResponse { + #[prost(string, tag = "1")] + pub request_id: String, + #[prost(string, tag = "2")] + pub call_id: String, + #[prost(int32, tag = "3")] + pub request_type: i32, + #[prost(int32, tag = "4")] + pub action: i32, + #[prost(string, tag = "5")] + pub text: String, + #[prost(string, tag = "7")] + pub idempotency_key: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct RespondRequest { + #[prost(message, optional, tag = "1")] + pub query: Option, + #[prost(string, tag = "2")] + pub channel_id: String, + #[prost(string, tag = "3")] + pub turn_id: String, + #[prost(message, optional, tag = "4")] + pub response: Option, + #[prost(string, tag = "5")] + pub idempotency_key: String, +} + +#[derive(Clone, PartialEq, Message)] +pub struct RespondResponse { + #[prost(int64, tag = "5")] + pub replay_cursor: i64, +} + +pub struct ThreadClient { + http: Client, + base: Url, + session: PlatformSession, + channel_id: String, +} + +impl ThreadClient { + pub fn new(session: PlatformSession, channel_id: String, base_url: &str) -> Result { + let base = Url::parse(base_url).context("invalid Platform URL")?; + if !matches!(base.scheme(), "https" | "http") + || base.host_str().is_none() + || (base.scheme() == "http" + && !matches!(base.host_str(), Some("127.0.0.1" | "localhost" | "::1"))) + { + bail!("Platform URL must use HTTPS or loopback HTTP"); + } + let http = Client::builder() + .timeout(Duration::from_secs(30)) + .redirect(reqwest::redirect::Policy::none()) + .build()?; + Ok(Self { + http, + base, + session, + channel_id, + }) + } + + pub fn query(&self) -> Result { + let workspace_id = self + .session + .workspace_id + .as_ref() + .filter(|id| !id.trim().is_empty()) + .context("managed login must select a workspace")? + .clone(); + Ok(Query { + workspace_id, + organization_id: self.session.organization_id.clone(), + }) + } + + async fn call( + &self, + method: &str, + request: Req, + ) -> Result { + let url = self.base.join(&format!("{SERVICE}/{method}"))?; + let workspace = self.query()?.workspace_id; + let mut response = self + .http + .post(url) + .bearer_auth(&self.session.access_token) + .header("X-Organization-ID", &self.session.organization_id) + .header("X-Workspace-ID", workspace) + .header("Connect-Protocol-Version", "1") + .header("Content-Type", "application/proto") + .header("Accept", "application/proto") + .body(request.encode_to_vec()) + .send() + .await + .context("Platform request failed")?; + let status = response.status(); + if !status.is_success() { + bail!("Platform {method} returned HTTP {status}"); + } + if response + .content_length() + .is_some_and(|size| size as usize > MAX_RESPONSE_BYTES) + { + bail!("Platform response exceeds limit"); + } + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await? { + if bytes.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES { + bail!("Platform response exceeds limit"); + } + bytes.extend_from_slice(&chunk); + } + Resp::decode(bytes.as_slice()).context("invalid Platform protobuf response") + } + + pub async fn get(&self) -> Result { + let result: GetResponse = self + .call( + "GetOperatingThread", + GetRequest { + query: Some(self.query()?), + channel_id: self.channel_id.clone(), + limit: 50, + }, + ) + .await?; + if result + .channel + .as_ref() + .is_none_or(|channel| channel.id != self.channel_id) + { + bail!("Platform returned a different or missing thread"); + } + Ok(result) + } + + pub async fn list_channels(&self, archived: bool) -> Result { + self.call( + "ListOperatingChannels", + ListChannelsRequest { + query: Some(self.query()?), + archive_filter: if archived { 2 } else { 1 }, + }, + ) + .await + } + + pub async fn rename(&self, title: String) -> Result { + let result: RenameThreadResponse = self + .call( + "RenameOperatingThread", + RenameThreadRequest { + query: Some(self.query()?), + channel_id: self.channel_id.clone(), + title, + }, + ) + .await?; + if result + .channel + .as_ref() + .is_none_or(|channel| channel.id != self.channel_id) + { + bail!("Platform returned a different or missing renamed thread"); + } + Ok(result) + } + + pub async fn archive(&self, archived: bool) -> Result { + let result: ArchiveThreadResponse = self + .call( + "ArchiveOperatingThread", + ArchiveThreadRequest { + query: Some(self.query()?), + channel_id: self.channel_id.clone(), + archived, + idempotency_key: Uuid::new_v4().to_string(), + }, + ) + .await?; + if result + .channel + .as_ref() + .is_none_or(|channel| channel.id != self.channel_id || channel.archived != archived) + { + bail!("Platform returned a different or unmodified archive state"); + } + Ok(result) + } + + pub async fn events(&self, cursor: i64) -> Result { + self.call( + "ListOperatingThreadEvents", + ListRequest { + query: Some(self.query()?), + channel_id: self.channel_id.clone(), + after_cursor: cursor, + limit: 200, + }, + ) + .await + } + + pub async fn submit(&self, body: String) -> Result { + if body.trim().is_empty() || body.len() > 20_000 { + bail!("message must contain 1 to 20000 bytes"); + } + let result: SubmitResponse = self + .call( + "SubmitOperatingMessage", + SubmitRequest { + query: Some(self.query()?), + channel_id: self.channel_id.clone(), + body, + idempotency_key: Uuid::new_v4().to_string(), + }, + ) + .await?; + if result + .accepted_turn + .as_ref() + .is_none_or(|turn| turn.turn_id.is_empty()) + { + bail!("Platform did not return an accepted turn"); + } + Ok(result) + } + + pub async fn respond( + &self, + pending: &Event, + action: i32, + text: String, + ) -> Result { + let key = Uuid::new_v4().to_string(); + self.call( + "RespondOperatingThread", + RespondRequest { + query: Some(self.query()?), + channel_id: self.channel_id.clone(), + turn_id: pending.turn_id.clone(), + response: Some(ThreadResponse { + request_id: pending.request_id.clone(), + call_id: pending.request_call_id.clone(), + request_type: pending.request_type, + action, + text, + idempotency_key: key.clone(), + }), + idempotency_key: key, + }, + ) + .await + } +} diff --git a/packages/local-host-rs/src/lib.rs b/packages/local-host-rs/src/lib.rs index f0ad46140..6eabcccd6 100644 --- a/packages/local-host-rs/src/lib.rs +++ b/packages/local-host-rs/src/lib.rs @@ -78,3 +78,5 @@ pub mod hosted_runner; pub mod hosted_runner_cli; pub mod hosted_runner_conformance; pub mod subagents; + +pub mod hosted_thread; diff --git a/packages/runtime-gateway-rs/src/auth.rs b/packages/runtime-gateway-rs/src/auth.rs index 89265af8c..4b7c1ae29 100644 --- a/packages/runtime-gateway-rs/src/auth.rs +++ b/packages/runtime-gateway-rs/src/auth.rs @@ -158,6 +158,7 @@ fn is_chat_websocket_request(head: &RequestHead) -> bool { fn runtime_tenant_resource_path(path: &str) -> bool { path.starts_with("/api/chat") + || path.starts_with("/api/hosted-threads/") || path.starts_with("/api/sessions") || matches!( path, diff --git a/packages/runtime-gateway-rs/src/hosted_threads.rs b/packages/runtime-gateway-rs/src/hosted_threads.rs new file mode 100644 index 000000000..1c1d4bcc9 --- /dev/null +++ b/packages/runtime-gateway-rs/src/hosted_threads.rs @@ -0,0 +1,502 @@ +//! Tenant-bound desktop access to Platform-owned operating threads. + +use super::*; +use maestro_local_host::credential_mode::{ + PlatformSession, verified_current_identity_session, verified_desktop_identity_session, +}; +use maestro_local_host::hosted_thread::ThreadClient; +use serde::de::DeserializeOwned; + +const ROOT: &str = "/api/hosted-threads"; +const PREFIX: &str = "/api/hosted-threads/"; +const MAX_HOSTED_BODY_BYTES: usize = 70 * 1024; + +#[derive(Debug, PartialEq, Eq)] +enum Operation { + List, + Rename, + Archive, + Snapshot, + Events, + Message, + Response, +} + +#[derive(Deserialize)] +struct MessageBody { + body: String, +} + +#[derive(Deserialize)] +struct RenameBody { + title: String, +} + +#[derive(Deserialize)] +struct ArchiveBody { + archived: bool, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ResponseBody { + cursor: String, + request_id: String, + action: String, + #[serde(default)] + text: String, +} + +pub(crate) fn is_hosted_thread_endpoint(head: &RequestHead) -> bool { + head.path == ROOT || head.path.starts_with(PREFIX) +} + +fn valid_thread_id(id: &str) -> bool { + let Some(suffix) = id.strip_prefix("thread:") else { + return false; + }; + !suffix.is_empty() + && suffix.len() <= 249 + && suffix + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_')) +} + +fn parse_route(path: &str) -> Option<(String, Operation)> { + let suffix = path.strip_prefix(PREFIX)?; + let (encoded_id, operation) = match suffix.split_once('/') { + Some((id, "rename")) => (id, Operation::Rename), + Some((id, "archive")) => (id, Operation::Archive), + Some((id, "events")) => (id, Operation::Events), + Some((id, "messages")) => (id, Operation::Message), + Some((id, "responses")) => (id, Operation::Response), + Some(_) => return None, + None => (suffix, Operation::Snapshot), + }; + let raw_id = percent_decode_component(encoded_id); + let id = raw_id.strip_prefix("thread:").unwrap_or(&raw_id); + let channel_id = format!("thread:{id}"); + if !valid_thread_id(&channel_id) { + return None; + } + Some((channel_id, operation)) +} + +fn private_json(status: u16, value: &impl Serialize) -> Vec { + let body = serde_json::to_vec(value).expect("hosted thread response is JSON serializable"); + response_with_no_store(status, "application/json", &body) +} + +fn error(status: u16, message: &str) -> Vec { + private_json(status, &serde_json::json!({ "error": message })) +} + +async fn parse_body( + stream: &mut TcpStream, + initial: &mut Vec, + head: &RequestHead, +) -> Result> { + let body = read_request_body_with_limit(stream, initial, head, MAX_HOSTED_BODY_BYTES) + .await + .map_err(|_| error(400, "invalid request body"))?; + serde_json::from_slice(&body).map_err(|_| error(400, "invalid JSON request")) +} + +fn session_matches_auth(auth: &AuthContext, session: &PlatformSession) -> bool { + let subject_matches = auth + .subject + .as_deref() + .is_none_or(|subject| session.user_id.as_deref() == Some(subject)); + let org_matches = auth + .organization_id + .as_deref() + .is_none_or(|org| org == session.organization_id); + let workspace_matches = auth + .workspace_id + .as_deref() + .is_none_or(|workspace| session.workspace_id.as_deref() == Some(workspace)); + subject_matches && org_matches && workspace_matches +} + +fn desktop_credential( + head: &RequestHead, + auth: &AuthContext, + loopback: bool, +) -> Result, Vec> { + let token = head.headers.get("x-maestro-identity-token"); + let org = head.headers.get("x-maestro-identity-organization"); + let workspace = head.headers.get("x-maestro-identity-workspace"); + if token.is_none() && org.is_none() && workspace.is_none() { + return Ok(None); + } + if !loopback || auth.source != AuthSource::StaticGatewayKey { + return Err(error( + 403, + "desktop credential requires local gateway authority", + )); + } + match (token, org, workspace) { + (Some(token), Some(org), Some(workspace)) + if !token.trim().is_empty() + && !org.trim().is_empty() + && !workspace.trim().is_empty() => + { + Ok(Some((token.clone(), org.clone(), workspace.clone()))) + } + _ => Err(error(401, "desktop credential is incomplete")), + } +} + +fn response_action(action: &str, text: &str) -> Option { + if text.len() > 65_536 { + return None; + } + match action { + "approve" => Some(1), + "deny" => Some(2), + "answer" if !text.trim().is_empty() => Some(3), + "retry" => Some(4), + "skip" => Some(5), + "abort" => Some(6), + _ => None, + } +} + +pub(crate) async fn handle_hosted_thread_endpoint( + stream: &mut TcpStream, + initial: &mut Vec, + head: RequestHead, + state: &AppState, +) -> Vec { + if let Err(response) = validate_csrf(&head, &state.config) { + return response; + } + let auth = match authorized_context(&head, &state.config) { + Ok(auth) => auth, + Err(response) => return response, + }; + let route = if head.path == ROOT { + Some((String::new(), Operation::List)) + } else { + parse_route(&head.path) + }; + let Some((channel_id, operation)) = route else { + return error(404, "hosted thread route not found"); + }; + let valid_method = matches!( + (&operation, head.method.as_str()), + ( + Operation::List | Operation::Snapshot | Operation::Events, + "GET" + ) | ( + Operation::Rename | Operation::Archive | Operation::Message | Operation::Response, + "POST" + ) + ); + if !valid_method { + return error(405, "method not allowed"); + } + + // Electron main can supply its live keychain token only over the private + // loopback gateway key. Otherwise, use the independent CLI credential. + let desktop = match desktop_credential(&head, &auth, state.config.listen_host_is_loopback()) { + Ok(credential) => credential, + Err(response) => return response, + }; + let session = match tokio::task::spawn_blocking(move || match desktop { + Some((token, org, workspace)) => { + verified_desktop_identity_session(&token, &org, &workspace) + } + None => verified_current_identity_session(), + }) + .await + { + Ok(Ok(session)) => session, + _ => return error(401, "managed sign-in required"), + }; + if !session_matches_auth(&auth, &session) { + return error(403, "caller tenant differs from managed sign-in"); + } + let Some(base_url) = maestro_local_host::managed_setup::platform_base_url() else { + return error(503, "Platform URL unavailable"); + }; + let client = match ThreadClient::new(session, channel_id, &base_url) { + Ok(client) => client, + Err(_) => return error(503, "Platform configuration unavailable"), + }; + + match operation { + Operation::List => { + let archived = match head.query.get("archived").map(String::as_str) { + None | Some("false") => false, + Some("true") => true, + _ => return error(400, "invalid archive filter"), + }; + match client.list_channels(archived).await { + Ok(result) => { + let threads: Vec<_> = result + .channels + .iter() + .filter(|channel| { + channel.archived == archived && valid_thread_id(&channel.id) + }) + .collect(); + private_json( + 200, + &serde_json::json!({ + "threads": threads.iter().take(200).map(|channel| serde_json::json!({ + "id": channel.id, + "label": channel.label, + "unreadCount": channel.unread_count, + "openCount": channel.open_count, + })).collect::>(), + "truncated": threads.len() > 200, + }), + ) + } + Err(_) => error(503, "Platform threads unavailable"), + } + } + Operation::Rename => { + let request: RenameBody = match parse_body(stream, initial, &head).await { + Ok(request) => request, + Err(response) => return response, + }; + let title = request.title.trim(); + if title.is_empty() || title.chars().count() > 80 { + return error(400, "title must contain 1 to 80 characters"); + } + match client.rename(title.to_owned()).await { + Ok(result) => { + let channel = result.channel.expect("rename validates channel identity"); + private_json( + 200, + &serde_json::json!({ + "id": channel.id, + "label": channel.label, + "changed": result.changed, + }), + ) + } + Err(_) => error(503, "Platform thread rename was not accepted"), + } + } + Operation::Archive => { + let request: ArchiveBody = match parse_body(stream, initial, &head).await { + Ok(request) => request, + Err(response) => return response, + }; + match client.archive(request.archived).await { + Ok(result) => { + let channel = result.channel.expect("archive validates channel identity"); + private_json( + 200, + &serde_json::json!({ + "id": channel.id, + "archived": channel.archived, + "changed": result.changed, + }), + ) + } + Err(_) => error(503, "Platform thread archive was not accepted"), + } + } + Operation::Snapshot => match client.get().await { + Ok(snapshot) => { + let archived = snapshot + .channel + .as_ref() + .is_some_and(|channel| channel.archived); + private_json( + 200, + &serde_json::json!({ + "channelId": snapshot.channel.map(|channel| channel.id), + "archived": archived, + "messages": snapshot.messages.iter().map(|message| serde_json::json!({ + "id": message.id, "role": message.role, "body": message.body + })).collect::>(), + "replayCursor": snapshot.replay_cursor.to_string() + }), + ) + } + Err(_) => error(503, "Platform thread unavailable"), + }, + Operation::Events => { + let cursor = match head.query.get("cursor").map(String::as_str) { + None => 0, + Some(value) => match value.parse::() { + Ok(cursor) if cursor >= 0 => cursor, + _ => return error(400, "invalid event cursor"), + }, + }; + match client.events(cursor).await { + Ok(page) if page.next_cursor >= cursor || page.reset_required => private_json( + 200, + &serde_json::json!({ + "events": page.events.iter().map(|event| serde_json::json!({ + "cursor": event.cursor.to_string(), + "eventId": event.event_id, + "turnId": event.turn_id, + "kind": event.kind, + "safeText": event.safe_text, + "requestId": event.request_id, + })).collect::>(), + "nextCursor": page.next_cursor.to_string(), + "hasMore": page.has_more, + "resetRequired": page.reset_required, + }), + ), + Err(_) => error(503, "Platform events unavailable"), + Ok(_) => error(503, "Platform event cursor moved backwards"), + } + } + Operation::Message => { + let request: MessageBody = match parse_body(stream, initial, &head).await { + Ok(request) => request, + Err(response) => return response, + }; + match client.submit(request.body).await { + Ok(result) => private_json( + 202, + &serde_json::json!({ + "turnId": result.accepted_turn.map(|turn| turn.turn_id), + "replayCursor": result.replay_cursor.to_string(), + }), + ), + Err(_) => error(503, "Platform message was not accepted"), + } + } + Operation::Response => { + let request: ResponseBody = match parse_body(stream, initial, &head).await { + Ok(request) => request, + Err(response) => return response, + }; + let Some(action) = response_action(&request.action, &request.text) else { + return error(400, "invalid response action"); + }; + let cursor = match request.cursor.parse::() { + Ok(cursor) if cursor > 0 => cursor, + _ => return error(400, "invalid request cursor"), + }; + if request.request_id.is_empty() { + return error(400, "request cursor and id required"); + } + // Resolve identity from Platform immediately before responding. + // A renderer may name a request, but cannot supply its type, call id, + // or turn id to the owner mutation. + let page = match client.events(cursor - 1).await { + Ok(page) if !page.reset_required => page, + _ => return error(409, "request must be refreshed"), + }; + let Some(event) = page.events.iter().find(|event| { + event.cursor == cursor + && event.request_id == request.request_id + && matches!(event.kind, 4 | 5 | 14 | 15) + }) else { + return error(409, "request must be refreshed"); + }; + match client.respond(event, action, request.text).await { + Ok(result) => private_json( + 200, + &serde_json::json!({ "replayCursor": result.replay_cursor.to_string() }), + ), + Err(_) => error(503, "Platform response was not accepted"), + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn rejects_encoded_path_escape_and_unrecognized_actions() { + assert!(parse_route("/api/hosted-threads/thread%3Aone%2Ftwo").is_none()); + assert!(parse_route("/api/hosted-threads/thread%3Aone/events/extra").is_none()); + assert_eq!(response_action("answer", " "), None); + assert_eq!(response_action("approve", ""), Some(1)); + } + + #[test] + fn routes_are_tenant_scoped_thread_ids() { + assert_eq!( + parse_route("/api/hosted-threads/thread%3Aabc_1/events"), + Some(("thread:abc_1".to_string(), Operation::Events)) + ); + assert_eq!( + parse_route("/api/hosted-threads/thread%3Aabc_1/rename"), + Some(("thread:abc_1".to_string(), Operation::Rename)) + ); + assert_eq!( + parse_route("/api/hosted-threads/thread%3Aabc_1/archive"), + Some(("thread:abc_1".to_string(), Operation::Archive)) + ); + assert!(valid_thread_id("thread:abc_1")); + assert!(!valid_thread_id("thread:../escape")); + assert!(!valid_thread_id("other:abc")); + assert!(is_hosted_thread_endpoint(&RequestHead { + method: "GET".into(), + path: ROOT.into(), + query: Default::default(), + headers: Default::default(), + })); + } + + #[test] + fn hosted_identity_cannot_cross_caller_subject_or_tenant() { + let session = PlatformSession { + access_token: "secret".into(), + organization_id: "org-a".into(), + workspace_id: Some("ws-a".into()), + provider_ref: Value::Null, + email: None, + user_id: Some("user-a".into()), + }; + let mut auth = AuthContext { + subject: Some("user-a".into()), + organization_id: Some("org-a".into()), + workspace_id: Some("ws-a".into()), + ..AuthContext::default() + }; + assert!(session_matches_auth(&auth, &session)); + auth.subject = Some("user-b".into()); + assert!(!session_matches_auth(&auth, &session)); + auth.subject = Some("user-a".into()); + auth.workspace_id = Some("ws-b".into()); + assert!(!session_matches_auth(&auth, &session)); + } + + #[test] + fn desktop_credential_requires_private_loopback_key_and_all_fields() { + let mut head = RequestHead { + method: "GET".into(), + path: "/api/hosted-threads/thread%3Aone".into(), + query: Default::default(), + headers: Default::default(), + }; + head.headers + .insert("x-maestro-identity-token".into(), "token".into()); + head.headers + .insert("x-maestro-identity-organization".into(), "org".into()); + head.headers + .insert("x-maestro-identity-workspace".into(), "workspace".into()); + let static_key = AuthContext { + source: AuthSource::StaticGatewayKey, + ..AuthContext::default() + }; + assert!( + desktop_credential(&head, &static_key, true) + .unwrap() + .is_some() + ); + assert!(desktop_credential(&head, &static_key, false).is_err()); + let jwt = AuthContext { + source: AuthSource::IdentityJwt, + ..AuthContext::default() + }; + assert!(desktop_credential(&head, &jwt, true).is_err()); + head.headers.remove("x-maestro-identity-workspace"); + assert!(desktop_credential(&head, &static_key, true).is_err()); + } +} diff --git a/packages/runtime-gateway-rs/src/lib.rs b/packages/runtime-gateway-rs/src/lib.rs index 7ced6d53f..5d1d46efc 100644 --- a/packages/runtime-gateway-rs/src/lib.rs +++ b/packages/runtime-gateway-rs/src/lib.rs @@ -47,6 +47,7 @@ mod codex_bridge; mod codex_compat; mod codex_subagent_dispatch; mod extended; +mod hosted_threads; mod http; mod local; mod markitdown; @@ -101,6 +102,7 @@ pub(crate) use chat::{ }; pub(crate) use codex_bridge::*; use extended::{ExtendedApiState, handle_extended_endpoint, is_extended_endpoint}; +use hosted_threads::{handle_hosted_thread_endpoint, is_hosted_thread_endpoint}; pub(crate) use http::MAX_JSON_BODY_BYTES; #[cfg(test)] use http::parse_request_head; @@ -926,6 +928,17 @@ async fn handle_connection(mut stream: TcpStream, state: AppState) -> anyhow::Re return Ok(()); } + if is_hosted_thread_endpoint(&head) { + let response = + handle_hosted_thread_endpoint(&mut stream, &mut initial, head, &state).await; + stream + .write_all(&response) + .await + .context("failed to write hosted thread response")?; + let _ = stream.shutdown().await; + return Ok(()); + } + if is_local_endpoint(&head) { let response = handle_local_endpoint(&mut stream, &mut initial, head, &state).await; stream diff --git a/packages/runtime-gateway-rs/src/tests.rs b/packages/runtime-gateway-rs/src/tests.rs index f70de5e70..8221482be 100644 --- a/packages/runtime-gateway-rs/src/tests.rs +++ b/packages/runtime-gateway-rs/src/tests.rs @@ -14219,6 +14219,8 @@ fn extended_admin_mcp_and_pending_paths_require_tenant_binding_on_remote() { ("POST", "/api/admin/enterprise-policy/publish"), ("GET", "/api/admin/enterprise-policy/status"), ("POST", "/api/pending-requests/req-1/resume"), + ("GET", "/api/hosted-threads/thread:one"), + ("POST", "/api/hosted-threads/thread:one/messages"), ] { assert!( authorize(&auth_head(method, path, &write_no_tenant), &remote_config).is_err(), diff --git a/packages/tui-rs/src/thread_cli.rs b/packages/tui-rs/src/thread_cli.rs index 62699fcf6..c5e84a149 100644 --- a/packages/tui-rs/src/thread_cli.rs +++ b/packages/tui-rs/src/thread_cli.rs @@ -1,180 +1,32 @@ //! Attach the native terminal to a Platform-owned Dex operating thread. //! -//! The wire projections below retain only the fields this client reads. Their -//! field numbers come from proto/console/v1/console.proto. Unknown fields stay -//! opaque; Platform remains the sole owner of execution and approval state. +//! Wire projections shared with the native desktop retain only fields these +//! clients read. Unknown fields stay opaque; Platform owns execution and approvals. use std::collections::{HashMap, HashSet}; use std::io::{self, IsTerminal, Write}; use std::time::{Duration, Instant}; use anyhow::{Context, Result, bail}; -use maestro_local_host::credential_mode::{PlatformSession, verified_current_identity_session}; +#[cfg(test)] +use maestro_local_host::credential_mode::PlatformSession; +use maestro_local_host::credential_mode::verified_current_identity_session; +#[cfg(test)] +use maestro_local_host::hosted_thread::{ + Channel, GetRequest, GetResponse, ListRequest, ListResponse, OperatingMessage, RespondRequest, + RespondResponse, SubmitRequest, SubmitResponse, Turn, +}; +use maestro_local_host::hosted_thread::{Event, ThreadClient}; +#[cfg(test)] use prost::Message; -use reqwest::{Client, Url}; use serde::Serialize; use tokio::time::sleep; -use uuid::Uuid; -const SERVICE: &str = "/deixic.v1.DeixicService"; -const MAX_RESPONSE_BYTES: usize = 4 * 1024 * 1024; const FOLLOW_LIMIT: Duration = Duration::from_mins(2); const USAGE: &str = "Usage: deixic-code thread attach [--message ] [--json] [--base-url ]\n\ The thread id is a Deixic thread id (thread:...). A managed Deixic login is required.\n\ Interactive commands: /quit, /respond [text]."; -// A narrow wire projection of console.v1. Field numbers and kinds mirror the -// canonical protobuf contract; unknown fields are ignored by prost. -#[derive(Clone, PartialEq, Message)] -struct Query { - #[prost(string, tag = "1")] - workspace_id: String, - #[prost(string, tag = "13")] - organization_id: String, -} - -#[derive(Clone, PartialEq, Message)] -struct GetRequest { - #[prost(message, optional, tag = "1")] - query: Option, - #[prost(string, tag = "2")] - channel_id: String, - #[prost(int32, tag = "3")] - limit: i32, -} - -#[derive(Clone, PartialEq, Message)] -struct Channel { - #[prost(string, tag = "1")] - id: String, -} - -#[derive(Clone, PartialEq, Message)] -struct OperatingMessage { - #[prost(string, tag = "1")] - id: String, - #[prost(string, tag = "3")] - role: String, - #[prost(string, tag = "5")] - body: String, -} - -#[derive(Clone, PartialEq, Message)] -struct GetResponse { - #[prost(message, optional, tag = "1")] - channel: Option, - #[prost(message, repeated, tag = "2")] - messages: Vec, - #[prost(int64, tag = "7")] - replay_cursor: i64, -} - -#[derive(Clone, PartialEq, Message)] -struct ListRequest { - #[prost(message, optional, tag = "1")] - query: Option, - #[prost(string, tag = "2")] - channel_id: String, - #[prost(int64, tag = "3")] - after_cursor: i64, - #[prost(int32, tag = "4")] - limit: i32, -} - -#[derive(Clone, PartialEq, Message)] -struct Event { - #[prost(int64, tag = "1")] - cursor: i64, - #[prost(string, tag = "2")] - event_id: String, - #[prost(string, tag = "3")] - turn_id: String, - #[prost(int32, tag = "4")] - kind: i32, - #[prost(string, tag = "5")] - safe_text: String, - #[prost(string, tag = "9")] - request_id: String, - #[prost(int32, tag = "10")] - request_type: i32, - #[prost(string, tag = "12")] - request_call_id: String, -} - -#[derive(Clone, PartialEq, Message)] -struct ListResponse { - #[prost(message, repeated, tag = "1")] - events: Vec, - #[prost(int64, tag = "2")] - next_cursor: i64, - #[prost(bool, tag = "3")] - has_more: bool, - #[prost(bool, tag = "4")] - reset_required: bool, -} - -#[derive(Clone, PartialEq, Message)] -struct SubmitRequest { - #[prost(message, optional, tag = "1")] - query: Option, - #[prost(string, tag = "2")] - channel_id: String, - #[prost(string, tag = "3")] - body: String, - #[prost(string, tag = "4")] - idempotency_key: String, -} - -#[derive(Clone, PartialEq, Message)] -struct Turn { - #[prost(string, tag = "1")] - turn_id: String, -} - -#[derive(Clone, PartialEq, Message)] -struct SubmitResponse { - #[prost(message, optional, tag = "6")] - accepted_turn: Option, - #[prost(int64, tag = "7")] - replay_cursor: i64, -} - -#[derive(Clone, PartialEq, Message)] -struct ThreadResponse { - #[prost(string, tag = "1")] - request_id: String, - #[prost(string, tag = "2")] - call_id: String, - #[prost(int32, tag = "3")] - request_type: i32, - #[prost(int32, tag = "4")] - action: i32, - #[prost(string, tag = "5")] - text: String, - #[prost(string, tag = "7")] - idempotency_key: String, -} - -#[derive(Clone, PartialEq, Message)] -struct RespondRequest { - #[prost(message, optional, tag = "1")] - query: Option, - #[prost(string, tag = "2")] - channel_id: String, - #[prost(string, tag = "3")] - turn_id: String, - #[prost(message, optional, tag = "4")] - response: Option, - #[prost(string, tag = "5")] - idempotency_key: String, -} - -#[derive(Clone, PartialEq, Message)] -struct RespondResponse { - #[prost(int64, tag = "5")] - replay_cursor: i64, -} - #[derive(Debug)] struct Options { channel_id: String, @@ -251,171 +103,6 @@ fn parse(args: &[String]) -> Result> { })) } -struct ThreadClient { - http: Client, - base: Url, - session: PlatformSession, - channel_id: String, -} - -impl ThreadClient { - fn new(session: PlatformSession, channel_id: String, base_url: &str) -> Result { - let base = Url::parse(base_url).context("invalid Platform URL")?; - if !matches!(base.scheme(), "https" | "http") - || base.host_str().is_none() - || (base.scheme() == "http" - && !matches!(base.host_str(), Some("127.0.0.1" | "localhost" | "::1"))) - { - bail!("Platform URL must use HTTPS or loopback HTTP"); - } - let http = Client::builder() - .timeout(Duration::from_secs(30)) - .redirect(reqwest::redirect::Policy::none()) - .build()?; - Ok(Self { - http, - base, - session, - channel_id, - }) - } - - fn query(&self) -> Result { - let workspace_id = self - .session - .workspace_id - .as_ref() - .filter(|id| !id.trim().is_empty()) - .context("managed login must select a workspace")? - .clone(); - Ok(Query { - workspace_id, - organization_id: self.session.organization_id.clone(), - }) - } - - async fn call( - &self, - method: &str, - request: Req, - ) -> Result { - let url = self.base.join(&format!("{SERVICE}/{method}"))?; - let workspace = self.query()?.workspace_id; - let mut response = self - .http - .post(url) - .bearer_auth(&self.session.access_token) - .header("X-Organization-ID", &self.session.organization_id) - .header("X-Workspace-ID", workspace) - .header("Connect-Protocol-Version", "1") - .header("Content-Type", "application/proto") - .header("Accept", "application/proto") - .body(request.encode_to_vec()) - .send() - .await - .context("Platform request failed")?; - let status = response.status(); - if !status.is_success() { - bail!("Platform {method} returned HTTP {status}"); - } - if response - .content_length() - .is_some_and(|size| size as usize > MAX_RESPONSE_BYTES) - { - bail!("Platform response exceeds limit"); - } - let mut bytes = Vec::new(); - while let Some(chunk) = response.chunk().await? { - if bytes.len().saturating_add(chunk.len()) > MAX_RESPONSE_BYTES { - bail!("Platform response exceeds limit"); - } - bytes.extend_from_slice(&chunk); - } - Resp::decode(bytes.as_slice()).context("invalid Platform protobuf response") - } - - async fn get(&self) -> Result { - let result: GetResponse = self - .call( - "GetOperatingThread", - GetRequest { - query: Some(self.query()?), - channel_id: self.channel_id.clone(), - limit: 50, - }, - ) - .await?; - if result - .channel - .as_ref() - .is_none_or(|channel| channel.id != self.channel_id) - { - bail!("Platform returned a different or missing thread"); - } - Ok(result) - } - - async fn events(&self, cursor: i64) -> Result { - self.call( - "ListOperatingThreadEvents", - ListRequest { - query: Some(self.query()?), - channel_id: self.channel_id.clone(), - after_cursor: cursor, - limit: 200, - }, - ) - .await - } - - async fn submit(&self, body: String) -> Result { - if body.trim().is_empty() || body.len() > 20_000 { - bail!("message must contain 1 to 20000 bytes"); - } - let result: SubmitResponse = self - .call( - "SubmitOperatingMessage", - SubmitRequest { - query: Some(self.query()?), - channel_id: self.channel_id.clone(), - body, - idempotency_key: Uuid::new_v4().to_string(), - }, - ) - .await?; - if result - .accepted_turn - .as_ref() - .is_none_or(|turn| turn.turn_id.is_empty()) - { - bail!("Platform did not return an accepted turn"); - } - Ok(result) - } - - async fn respond(&self, pending: &Event, action: i32, text: String) -> Result { - let key = Uuid::new_v4().to_string(); - self.call( - "RespondOperatingThread", - RespondRequest { - query: Some(self.query()?), - channel_id: self.channel_id.clone(), - turn_id: pending.turn_id.clone(), - response: Some(ThreadResponse { - request_id: pending.request_id.clone(), - call_id: pending.request_call_id.clone(), - request_type: pending.request_type, - action, - text, - idempotency_key: key.clone(), - }), - idempotency_key: key, - }, - ) - .await - } -} - #[derive(Serialize)] #[serde(tag = "type", rename_all = "snake_case")] enum Output<'a> { @@ -814,6 +501,7 @@ mod tests { GetResponse { channel: Some(Channel { id: "thread:one".into(), + ..Channel::default() }), messages: vec![OperatingMessage { id: "msg-1".into(),