From e5f0c4168f1a8c237529f695cc8202cfa83b7549 Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 3 Aug 2026 23:37:15 +0900 Subject: [PATCH] workdir: add authenticated runtime session transport --- Cargo.lock | 2 + crates/workdir/Cargo.toml | 5 + crates/workdir/src/http.rs | 499 ++++++++++++++++++++ crates/workdir/src/lib.rs | 1 + crates/worker-runtime/src/execution.rs | 20 + crates/worker-runtime/src/http_server.rs | 427 ++++++++++++++++- crates/worker-runtime/src/runtime.rs | 45 ++ crates/worker-runtime/src/worker_backend.rs | 25 + crates/workspace-server/Cargo.toml | 1 + crates/workspace-server/src/hosts.rs | 200 +++++++- 10 files changed, 1220 insertions(+), 5 deletions(-) create mode 100644 crates/workdir/src/http.rs diff --git a/Cargo.lock b/Cargo.lock index fe549e90..3a7b90bd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5930,6 +5930,7 @@ dependencies = [ "grep-searcher", "ignore", "manifest", + "reqwest", "serde", "serde_json", "sha2 0.11.0", @@ -6130,6 +6131,7 @@ dependencies = [ "url", "uuid", "webauthn-rs", + "workdir", "worker", "worker-runtime", ] diff --git a/crates/workdir/Cargo.toml b/crates/workdir/Cargo.toml index 2da86af3..ec8687f7 100644 --- a/crates/workdir/Cargo.toml +++ b/crates/workdir/Cargo.toml @@ -4,6 +4,10 @@ version = "0.1.0" edition.workspace = true license.workspace = true +[features] +default = [] +http-client = ["dep:reqwest"] + [dependencies] async-trait.workspace = true globset = "0.4.18" @@ -12,6 +16,7 @@ grep-regex = "0.1.14" grep-searcher = "0.1.16" ignore = "0.4.25" manifest.workspace = true +reqwest = { version = "0.13", default-features = false, features = ["json", "rustls"], optional = true } serde = { workspace = true, features = ["derive"] } sha2.workspace = true tempfile.workspace = true diff --git a/crates/workdir/src/http.rs b/crates/workdir/src/http.rs new file mode 100644 index 00000000..63c34662 --- /dev/null +++ b/crates/workdir/src/http.rs @@ -0,0 +1,499 @@ +//! HTTP transport contract and client for remote Workdir sessions. +//! +//! The protocol keeps filesystem/search/process operations provider-side: one +//! [`WorkdirSessionOperation`] is one bounded HTTP request. The HTTP client is +//! optional so Runtime servers can share these DTOs without depending on a +//! client stack. + +use serde::{Deserialize, Serialize}; + +use crate::{ + CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest, + EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, + ReadRequest, ReadResult, StatRequest, StatResult, WorkdirError, WorkdirId, + WorkdirSessionCapabilities, WriteRequest, WriteResult, +}; + +/// Opaque Runtime-owned identifier for one ephemeral Workdir session. +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct WorkdirSessionId(String); + +impl WorkdirSessionId { + pub fn new(value: impl Into) -> Result { + let value = value.into(); + if value.trim().is_empty() { + return Err(WorkdirError::InvalidArgument( + "Workdir session id must not be empty".to_string(), + )); + } + Ok(Self(value)) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +/// Open a fresh session for a persisted Workdir identity. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct OpenWorkdirSessionRequest { + /// Optional Runtime Worker whose persisted binding establishes workspace + /// ownership of the Workdir. Runtime servers reject cross-workspace owners. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub owner_worker_id: Option, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct OpenWorkdirSessionResponse { + pub session_id: WorkdirSessionId, + pub workdir_id: WorkdirId, + pub capabilities: WorkdirSessionCapabilities, +} + +/// One provider-side Workdir operation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "operation", content = "request", rename_all = "snake_case")] +pub enum WorkdirSessionOperation { + Stat(StatRequest), + Read(ReadRequest), + Write(WriteRequest), + Edit(EditRequest), + List(ListRequest), + Glob(GlobRequest), + Grep(GrepRequest), + CommandStart(CommandRequest), + CommandStatus(CommandHandle), + CommandOutput(CommandOutputRequest), + CommandCancel(CommandHandle), +} + +/// Typed result paired with [`WorkdirSessionOperation`]. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "operation", content = "result", rename_all = "snake_case")] +pub enum WorkdirSessionOperationResult { + Stat(StatResult), + Read(ReadResult), + Write(WriteResult), + Edit(EditResult), + List(ListResult), + Glob(GlobResult), + Grep(GrepResult), + CommandStart(CommandHandle), + CommandStatus(CommandStatus), + CommandOutput(CommandOutput), + CommandCancel, +} + +/// Stable, host-path-free error code crossing the Runtime boundary. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum WorkdirTransportErrorCode { + NotFound, + Conflict, + Unsupported, + InvalidRequest, + UnknownCommand, + Unavailable, + Internal, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct WorkdirTransportError { + pub code: WorkdirTransportErrorCode, + pub message: String, +} + +impl WorkdirTransportError { + /// Convert a provider error without exposing materialization paths or raw I/O errors. + pub fn from_workdir_error(error: &WorkdirError) -> Self { + use WorkdirTransportErrorCode as Code; + let (code, message) = match error { + WorkdirError::NotFound(_) => (Code::NotFound, "Workdir path was not found"), + WorkdirError::Conflict(_) => (Code::Conflict, "Workdir content changed"), + WorkdirError::Unsupported(capability) => { + return Self { + code: Code::Unsupported, + message: format!("Workdir capability {capability:?} is not available"), + }; + } + WorkdirError::UnknownCommand(_) => { + (Code::UnknownCommand, "Workdir command was not found") + } + WorkdirError::Unavailable(_) => (Code::Unavailable, "Workdir session is unavailable"), + WorkdirError::InvalidPath(_) + | WorkdirError::RelativePath(_) + | WorkdirError::InvalidGlob(_) + | WorkdirError::InvalidRegex(_) + | WorkdirError::InvalidArgument(_) => { + (Code::InvalidRequest, "Workdir operation request is invalid") + } + WorkdirError::OutOfScope(_) + | WorkdirError::SymlinkOutOfScope { .. } + | WorkdirError::BrokenSymlink { .. } + | WorkdirError::SymlinkTargetIsDirectory { .. } + | WorkdirError::ReadOnly(_) + | WorkdirError::IsDirectory(_) + | WorkdirError::SymlinkDirectoryNotTraversed { .. } + | WorkdirError::Io { .. } => (Code::Internal, "Workdir operation failed"), + }; + Self { + code, + message: message.to_string(), + } + } + + pub fn into_workdir_error(self) -> WorkdirError { + use WorkdirTransportErrorCode as Code; + match self.code { + Code::NotFound => WorkdirError::NotFound("".into()), + Code::Conflict => WorkdirError::Conflict(self.message), + Code::Unsupported => WorkdirError::Unavailable(self.message), + Code::UnknownCommand => WorkdirError::UnknownCommand("".to_string()), + Code::InvalidRequest => WorkdirError::InvalidArgument(self.message), + Code::Unavailable | Code::Internal => WorkdirError::Unavailable(self.message), + } + } +} + +#[cfg(feature = "http-client")] +mod client { + use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }; + + use async_trait::async_trait; + use reqwest::{Client, StatusCode, Url}; + + use super::*; + use crate::{Workdir, WorkdirSession}; + + /// Provides a fresh bearer token for each Runtime request. Backend + /// implementations can mint short-lived capability tokens without making a + /// Worker-bound session expire with the token used to open it. + pub trait WorkdirHttpAuthorization: std::fmt::Debug + Send + Sync { + fn bearer_token(&self) -> Result; + } + + struct FixedBearerToken(Arc); + + impl std::fmt::Debug for FixedBearerToken { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("FixedBearerToken()") + } + } + + impl WorkdirHttpAuthorization for FixedBearerToken { + fn bearer_token(&self) -> Result { + Ok(self.0.to_string()) + } + } + + /// Authenticated HTTP implementation of [`WorkdirSession`]. + /// + /// Clone and reuse one `reqwest::Client` per Runtime to preserve connection + /// pooling and keep-alive across Worker-bound sessions. + #[derive(Debug)] + pub struct RemoteWorkdirSession { + client: Client, + base_url: Url, + authorization: Arc, + workdir: Workdir, + session_id: WorkdirSessionId, + capabilities: WorkdirSessionCapabilities, + closed: AtomicBool, + } + + impl RemoteWorkdirSession { + pub async fn open( + client: Client, + base_url: Url, + bearer_token: impl Into>, + workdir_id: WorkdirId, + request: OpenWorkdirSessionRequest, + ) -> Result { + Self::open_with_authorization( + client, + base_url, + Arc::new(FixedBearerToken(bearer_token.into())), + workdir_id, + request, + ) + .await + } + + pub async fn open_with_authorization( + client: Client, + base_url: Url, + authorization: Arc, + workdir_id: WorkdirId, + request: OpenWorkdirSessionRequest, + ) -> Result { + let url = endpoint( + &base_url, + &["v1", "working-directories", workdir_id.as_str(), "sessions"], + )?; + let response = client + .post(url) + .bearer_auth(authorization.bearer_token()?) + .json(&request) + .send() + .await + .map_err(http_unavailable)?; + let opened: OpenWorkdirSessionResponse = decode_response(response).await?; + if opened.workdir_id.as_str() != workdir_id.as_str() { + return Err(WorkdirError::Unavailable( + "Runtime opened a session for a different Workdir".to_string(), + )); + } + Ok(Self { + client, + base_url, + authorization, + workdir: Workdir::new(opened.workdir_id.as_str()), + session_id: opened.session_id, + capabilities: opened.capabilities, + closed: AtomicBool::new(false), + }) + } + + pub fn session_id(&self) -> &WorkdirSessionId { + &self.session_id + } + + async fn operate( + &self, + operation: WorkdirSessionOperation, + ) -> Result { + if self.closed.load(Ordering::Acquire) { + return Err(WorkdirError::Unavailable( + "Workdir session is closed".to_string(), + )); + } + let url = endpoint( + &self.base_url, + &[ + "v1", + "workdir-sessions", + self.session_id.as_str(), + "operations", + ], + )?; + let response = self + .client + .post(url) + .bearer_auth(self.authorization.bearer_token()?) + .json(&operation) + .send() + .await + .map_err(http_unavailable)?; + decode_response(response).await + } + + fn mismatch(expected: &str) -> WorkdirError { + WorkdirError::Unavailable(format!( + "Runtime returned a mismatched Workdir operation result; expected {expected}" + )) + } + } + + #[async_trait] + impl WorkdirSession for RemoteWorkdirSession { + fn workdir(&self) -> &Workdir { + &self.workdir + } + + fn capabilities(&self) -> WorkdirSessionCapabilities { + self.capabilities + } + + async fn stat(&self, request: StatRequest) -> Result { + match self.operate(WorkdirSessionOperation::Stat(request)).await? { + WorkdirSessionOperationResult::Stat(result) => Ok(result), + _ => Err(Self::mismatch("stat")), + } + } + + async fn read(&self, request: ReadRequest) -> Result { + match self.operate(WorkdirSessionOperation::Read(request)).await? { + WorkdirSessionOperationResult::Read(result) => Ok(result), + _ => Err(Self::mismatch("read")), + } + } + + async fn write(&self, request: WriteRequest) -> Result { + match self + .operate(WorkdirSessionOperation::Write(request)) + .await? + { + WorkdirSessionOperationResult::Write(result) => Ok(result), + _ => Err(Self::mismatch("write")), + } + } + + async fn edit(&self, request: EditRequest) -> Result { + match self.operate(WorkdirSessionOperation::Edit(request)).await? { + WorkdirSessionOperationResult::Edit(result) => Ok(result), + _ => Err(Self::mismatch("edit")), + } + } + + async fn list(&self, request: ListRequest) -> Result { + match self.operate(WorkdirSessionOperation::List(request)).await? { + WorkdirSessionOperationResult::List(result) => Ok(result), + _ => Err(Self::mismatch("list")), + } + } + + async fn glob(&self, request: GlobRequest) -> Result { + match self.operate(WorkdirSessionOperation::Glob(request)).await? { + WorkdirSessionOperationResult::Glob(result) => Ok(result), + _ => Err(Self::mismatch("glob")), + } + } + + async fn grep(&self, request: GrepRequest) -> Result { + match self.operate(WorkdirSessionOperation::Grep(request)).await? { + WorkdirSessionOperationResult::Grep(result) => Ok(result), + _ => Err(Self::mismatch("grep")), + } + } + + async fn start_command( + &self, + request: CommandRequest, + ) -> Result { + match self + .operate(WorkdirSessionOperation::CommandStart(request)) + .await? + { + WorkdirSessionOperationResult::CommandStart(result) => Ok(result), + _ => Err(Self::mismatch("command_start")), + } + } + + async fn command_status( + &self, + handle: CommandHandle, + ) -> Result { + match self + .operate(WorkdirSessionOperation::CommandStatus(handle)) + .await? + { + WorkdirSessionOperationResult::CommandStatus(result) => Ok(result), + _ => Err(Self::mismatch("command_status")), + } + } + + async fn command_output( + &self, + request: CommandOutputRequest, + ) -> Result { + let wait = request.wait; + loop { + match self + .operate(WorkdirSessionOperation::CommandOutput(request.clone())) + .await? + { + WorkdirSessionOperationResult::CommandOutput(result) + if wait && result.status == CommandStatus::Running => {} + WorkdirSessionOperationResult::CommandOutput(result) => return Ok(result), + _ => return Err(Self::mismatch("command_output")), + } + } + } + + async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> { + match self + .operate(WorkdirSessionOperation::CommandCancel(handle)) + .await? + { + WorkdirSessionOperationResult::CommandCancel => Ok(()), + _ => Err(Self::mismatch("command_cancel")), + } + } + + async fn close(&self) -> Result<(), WorkdirError> { + if self.closed.swap(true, Ordering::AcqRel) { + return Ok(()); + } + let url = endpoint( + &self.base_url, + &["v1", "workdir-sessions", self.session_id.as_str()], + )?; + let response = self + .client + .delete(url) + .bearer_auth(self.authorization.bearer_token()?) + .send() + .await + .map_err(http_unavailable)?; + if response.status() == StatusCode::NO_CONTENT || response.status().is_success() { + Ok(()) + } else { + Err(decode_error(response).await) + } + } + } + + fn endpoint(base_url: &Url, segments: &[&str]) -> Result { + let mut url = base_url.clone(); + { + let mut path = url.path_segments_mut().map_err(|_| { + WorkdirError::InvalidArgument( + "Runtime base URL cannot be used for path-based Workdir operations".to_string(), + ) + })?; + path.pop_if_empty(); + path.extend(segments.iter().copied()); + } + Ok(url) + } + + async fn decode_response( + response: reqwest::Response, + ) -> Result { + if response.status().is_success() { + response.json().await.map_err(http_unavailable) + } else { + Err(decode_error(response).await) + } + } + + async fn decode_error(response: reqwest::Response) -> WorkdirError { + response + .json::() + .await + .map(WorkdirTransportError::into_workdir_error) + .unwrap_or_else(|error| { + WorkdirError::Unavailable(format!("Runtime HTTP error: {error}")) + }) + } + + fn http_unavailable(error: reqwest::Error) -> WorkdirError { + WorkdirError::Unavailable(format!("Runtime Workdir HTTP request failed: {error}")) + } + + pub use self::RemoteWorkdirSession as ClientSession; +} + +#[cfg(feature = "http-client")] +pub use client::{ClientSession as RemoteWorkdirSession, WorkdirHttpAuthorization}; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn transport_error_does_not_expose_host_path() { + let error = WorkdirError::Io { + path: "/secret/runtime/root/file".into(), + source: std::io::Error::new(std::io::ErrorKind::PermissionDenied, "host detail"), + }; + let transport = WorkdirTransportError::from_workdir_error(&error); + assert_eq!(transport.code, WorkdirTransportErrorCode::Internal); + assert!(!transport.message.contains("/secret")); + assert!(!transport.message.contains("host detail")); + } +} diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 3deff787..5a576651 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -5,6 +5,7 @@ //! bound to one Worker. Tools consume sessions; they do not own Workdir //! materialization or cleanup. +pub mod http; mod local; mod operation; mod search; diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 6f5144a8..b75315da 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -10,6 +10,7 @@ use protocol::Method; use serde::{Deserialize, Serialize}; use std::fmt; use std::sync::Arc; +use workdir::WorkdirSessionHandle; /// Current execution-side run state for a Worker. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] @@ -293,6 +294,18 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static { )) } + fn open_workdir_session( + &self, + working_directory_id: &str, + ) -> Result { + Err(WorkingDirectoryDiagnostic::rejected( + "workdir_session_unsupported", + format!( + "working directory `{working_directory_id}` does not expose operation sessions" + ), + )) + } + fn cleanup_working_directory( &self, working_directory_id: &str, @@ -403,6 +416,13 @@ impl WorkerExecutionBackendRef { self.backend.working_directory(working_directory_id) } + pub(crate) fn open_workdir_session( + &self, + working_directory_id: &str, + ) -> Result { + self.backend.open_workdir_session(working_directory_id) + } + pub(crate) fn cleanup_working_directory( &self, working_directory_id: &str, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 7a9c9114..4e925a47 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -7,7 +7,7 @@ //! credentials, registration, and policy. use crate::auth::{ - RuntimeAuthContext, RuntimeAuthError, RuntimeHttpAuthConfig, unix_now_seconds, + RuntimeAuthContext, RuntimeAuthError, RuntimeHttpAuthConfig, new_token_id, unix_now_seconds, verify_capability_token, }; use crate::catalog::{ @@ -32,7 +32,7 @@ use axum::extract::{Extension, Path, Query, State}; use axum::http::{Method, Request, StatusCode, header}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; -use axum::routing::{get, post}; +use axum::routing::{delete, get, post}; use axum::{Json, Router}; #[cfg(feature = "ws-server")] use futures::{SinkExt, StreamExt}; @@ -45,14 +45,21 @@ use protocol::subscription::{ SubscriptionTerminationCode, }; use serde::{Deserialize, Serialize}; -#[cfg(feature = "ws-server")] use std::collections::HashMap; use std::fmt; use std::net::SocketAddr; #[cfg(feature = "fs-store")] use std::path::PathBuf; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use tokio::net::TcpListener; +use workdir::{ + CommandOutput, CommandStatus, WorkdirSessionHandle, + http::{ + OpenWorkdirSessionRequest, OpenWorkdirSessionResponse, WorkdirSessionId, + WorkdirSessionOperation, WorkdirSessionOperationResult, WorkdirTransportError, + WorkdirTransportErrorCode, + }, +}; const DEFAULT_RUNTIME_HTTP_PORT: u16 = 38800; @@ -172,6 +179,7 @@ fn runtime_http_router_with_optional_auth( runtime, local_token: local_token.map(Arc::::from), auth: auth.map(Arc::new), + workdir_sessions: Arc::new(Mutex::new(HashMap::new())), }; let router = Router::new() @@ -188,6 +196,18 @@ fn runtime_http_router_with_optional_auth( "/v1/working-directories", get(list_working_directories).post(create_working_directory), ) + .route( + "/v1/working-directories/{working_directory_id}/sessions", + post(open_workdir_session), + ) + .route( + "/v1/workdir-sessions/{session_id}/operations", + post(run_workdir_session_operation), + ) + .route( + "/v1/workdir-sessions/{session_id}", + delete(close_workdir_session), + ) .route( "/v1/working-directories/{working_directory_id}", get(get_working_directory).delete(cleanup_working_directory), @@ -228,6 +248,12 @@ struct RuntimeHttpState { runtime: Runtime, local_token: Option>, auth: Option>, + workdir_sessions: Arc>>, +} + +struct RuntimeHttpWorkdirSession { + owner: RuntimeWorkspaceScope, + session: WorkdirSessionHandle, } /// `GET /v1/runtime` response. @@ -472,6 +498,164 @@ async fn get_working_directory( })) } +async fn open_workdir_session( + State(state): State, + Path(working_directory_id): Path, + auth: Option>, + body: Result, JsonRejection>, +) -> Result, RuntimeHttpWorkdirError> { + let Json(request) = body.map_err(|_| RuntimeHttpWorkdirError::invalid_request())?; + let owner = required_workdir_owner(auth)?; + let owner_worker_ref = request + .owner_worker_id + .as_deref() + .map(|worker_id| { + WorkerId::parse(worker_id) + .map(WorkerRef::new) + .ok_or_else(RuntimeHttpWorkdirError::invalid_request) + }) + .transpose()?; + let session = state + .runtime + .open_workdir_session_scoped(&owner, &working_directory_id, owner_worker_ref.as_ref()) + .map_err(RuntimeHttpWorkdirError::runtime)?; + let session_id = + WorkdirSessionId::new(new_token_id().map_err(|_| RuntimeHttpWorkdirError::internal())?) + .map_err(|_| RuntimeHttpWorkdirError::internal())?; + let response = OpenWorkdirSessionResponse { + session_id: session_id.clone(), + workdir_id: session.workdir().id().clone(), + capabilities: session.capabilities(), + }; + state + .workdir_sessions + .lock() + .map_err(|_| RuntimeHttpWorkdirError::internal())? + .insert( + session_id.as_str().to_string(), + RuntimeHttpWorkdirSession { owner, session }, + ); + Ok(Json(response)) +} + +async fn run_workdir_session_operation( + State(state): State, + Path(session_id): Path, + auth: Option>, + body: Result, JsonRejection>, +) -> Result, RuntimeHttpWorkdirError> { + let Json(operation) = body.map_err(|_| RuntimeHttpWorkdirError::invalid_request())?; + let owner = required_workdir_owner(auth)?; + let session = { + let sessions = state + .workdir_sessions + .lock() + .map_err(|_| RuntimeHttpWorkdirError::internal())?; + let record = sessions + .get(&session_id) + .filter(|record| record.owner == owner) + .ok_or_else(RuntimeHttpWorkdirError::not_found)?; + record.session.clone() + }; + + let result = match operation { + WorkdirSessionOperation::Stat(request) => { + WorkdirSessionOperationResult::Stat(session.stat(request).await?) + } + WorkdirSessionOperation::Read(request) => { + WorkdirSessionOperationResult::Read(session.read(request).await?) + } + WorkdirSessionOperation::Write(request) => { + WorkdirSessionOperationResult::Write(session.write(request).await?) + } + WorkdirSessionOperation::Edit(request) => { + WorkdirSessionOperationResult::Edit(session.edit(request).await?) + } + WorkdirSessionOperation::List(request) => { + WorkdirSessionOperationResult::List(session.list(request).await?) + } + WorkdirSessionOperation::Glob(request) => { + WorkdirSessionOperationResult::Glob(session.glob(request).await?) + } + WorkdirSessionOperation::Grep(request) => { + WorkdirSessionOperationResult::Grep(session.grep(request).await?) + } + WorkdirSessionOperation::CommandStart(request) => { + WorkdirSessionOperationResult::CommandStart(session.start_command(request).await?) + } + WorkdirSessionOperation::CommandStatus(handle) => { + WorkdirSessionOperationResult::CommandStatus(session.command_status(handle).await?) + } + WorkdirSessionOperation::CommandOutput(request) if request.wait => { + let cursor = request.cursor; + let output = match tokio::time::timeout( + std::time::Duration::from_secs(20), + session.command_output(request), + ) + .await + { + Ok(result) => result?, + Err(_) => CommandOutput { + exit_code: None, + timed_out: false, + status: CommandStatus::Running, + content: String::new(), + next_cursor: Some(cursor), + truncated: false, + }, + }; + WorkdirSessionOperationResult::CommandOutput(output) + } + WorkdirSessionOperation::CommandOutput(request) => { + WorkdirSessionOperationResult::CommandOutput(session.command_output(request).await?) + } + WorkdirSessionOperation::CommandCancel(handle) => { + session.cancel_command(handle).await?; + WorkdirSessionOperationResult::CommandCancel + } + }; + Ok(Json(result)) +} + +async fn close_workdir_session( + State(state): State, + Path(session_id): Path, + auth: Option>, +) -> Result { + let owner = required_workdir_owner(auth)?; + let session = { + let mut sessions = state + .workdir_sessions + .lock() + .map_err(|_| RuntimeHttpWorkdirError::internal())?; + if sessions + .get(&session_id) + .is_some_and(|record| record.owner == owner) + { + sessions.remove(&session_id).map(|record| record.session) + } else { + None + } + }; + if let Some(session) = session { + session.close().await?; + } + Ok(StatusCode::NO_CONTENT) +} + +fn required_workdir_owner( + auth: Option>, +) -> Result { + let Extension(auth) = auth.ok_or_else(RuntimeHttpWorkdirError::forbidden)?; + if auth.workspace_id.trim().is_empty() || auth.server_id.trim().is_empty() { + return Err(RuntimeHttpWorkdirError::forbidden()); + } + Ok(RuntimeWorkspaceScope::new( + auth.workspace_id, + auth.server_id, + )) +} + async fn cleanup_working_directory( State(state): State, Path(working_directory_id): Path, @@ -1263,6 +1447,11 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s if path == "/v1/workers" && *method == Method::POST { return Some("workers:create"); } + if path.starts_with("/v1/workdir-sessions") + || (path.starts_with("/v1/working-directories/") && path.ends_with("/sessions")) + { + return Some("workdirs:operate"); + } if path.starts_with("/v1/config-bundles") || path.starts_with("/v1/working-directories") { return Some("workers:create"); } @@ -1293,6 +1482,101 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s None } +#[derive(Debug)] +struct RuntimeHttpWorkdirError { + status: StatusCode, + payload: WorkdirTransportError, +} + +impl RuntimeHttpWorkdirError { + fn new( + status: StatusCode, + code: WorkdirTransportErrorCode, + message: impl Into, + ) -> Self { + Self { + status, + payload: WorkdirTransportError { + code, + message: message.into(), + }, + } + } + + fn invalid_request() -> Self { + Self::new( + StatusCode::BAD_REQUEST, + WorkdirTransportErrorCode::InvalidRequest, + "Workdir operation request is invalid", + ) + } + + fn forbidden() -> Self { + Self::new( + StatusCode::FORBIDDEN, + WorkdirTransportErrorCode::Unavailable, + "Workdir session is unavailable", + ) + } + + fn not_found() -> Self { + Self::new( + StatusCode::NOT_FOUND, + WorkdirTransportErrorCode::NotFound, + "Workdir session was not found", + ) + } + + fn internal() -> Self { + Self::new( + StatusCode::INTERNAL_SERVER_ERROR, + WorkdirTransportErrorCode::Internal, + "Workdir operation failed", + ) + } + + fn runtime(error: RuntimeError) -> Self { + match error { + RuntimeError::WorkingDirectory(diagnostic) + if diagnostic.code == "working_directory_not_found" => + { + Self::not_found() + } + RuntimeError::WorkspaceOwnerMismatch { .. } => Self::not_found(), + RuntimeError::InvalidRequest(_) => Self::invalid_request(), + _ => Self::new( + StatusCode::SERVICE_UNAVAILABLE, + WorkdirTransportErrorCode::Unavailable, + "Workdir session is unavailable", + ), + } + } +} + +impl From for RuntimeHttpWorkdirError { + fn from(error: workdir::WorkdirError) -> Self { + let payload = WorkdirTransportError::from_workdir_error(&error); + let status = match payload.code { + WorkdirTransportErrorCode::NotFound | WorkdirTransportErrorCode::UnknownCommand => { + StatusCode::NOT_FOUND + } + WorkdirTransportErrorCode::Conflict => StatusCode::CONFLICT, + WorkdirTransportErrorCode::Unsupported | WorkdirTransportErrorCode::InvalidRequest => { + StatusCode::BAD_REQUEST + } + WorkdirTransportErrorCode::Unavailable => StatusCode::SERVICE_UNAVAILABLE, + WorkdirTransportErrorCode::Internal => StatusCode::INTERNAL_SERVER_ERROR, + }; + Self { status, payload } + } +} + +impl IntoResponse for RuntimeHttpWorkdirError { + fn into_response(self) -> Response { + (self.status, Json(self.payload)).into_response() + } +} + #[derive(Debug)] struct RuntimeHttpRestError { status: StatusCode, @@ -1434,7 +1718,11 @@ mod tests { use crate::management::RuntimeOptions; use axum::body::to_bytes; use axum::http::Method; + use manifest::{Scope, SharedScope}; use tower::ServiceExt; + use workdir::{ + LocalWorkdirSession, StatRequest, Workdir, WorkdirPath, WorkdirSessionCapabilities, + }; fn test_bundle(profile: ProfileSelector) -> ConfigBundle { ConfigBundle { @@ -1525,6 +1813,7 @@ mod tests { "workers:stop", "workers:protocol", "workers:delete", + "workdirs:operate", ], ) } @@ -1719,6 +2008,27 @@ mod tests { assert_eq!(response.status(), StatusCode::FORBIDDEN); } + #[tokio::test] + async fn capability_token_without_workdir_permission_is_forbidden() { + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(AcceptingBackend)) + .unwrap(); + let (auth, signer) = auth_config_and_signer(); + let token = token_for_workspace_with_permissions(&signer, "workspace-a", ["workers:read"]); + let app = runtime_http_router_with_auth(runtime, None, auth); + + let response = app + .oneshot(bearer_request( + Method::POST, + "/v1/working-directories/wd-1/sessions", + &token, + serde_json::to_vec(&OpenWorkdirSessionRequest::default()).unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + fn task_request(_objective: &str) -> CreateWorkerRequest { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle(profile.clone()); @@ -1755,6 +2065,115 @@ mod tests { } } + #[test] + fn workdir_routes_require_dedicated_operation_permission() { + assert_eq!( + required_runtime_permission(&Method::POST, "/v1/working-directories/wd-1/sessions"), + Some("workdirs:operate") + ); + assert_eq!( + required_runtime_permission(&Method::POST, "/v1/workdir-sessions/session-1/operations"), + Some("workdirs:operate") + ); + assert_eq!( + required_runtime_permission(&Method::DELETE, "/v1/workdir-sessions/session-1"), + Some("workdirs:operate") + ); + } + + #[tokio::test] + async fn workdir_session_operations_enforce_owner_and_close_terminally() { + let temp = tempfile::tempdir().expect("tempdir"); + std::fs::write(temp.path().join("hello.txt"), "hello").expect("write fixture"); + let scope = SharedScope::new(Scope::writable(temp.path()).expect("scope")); + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound( + Workdir::new("wd-1"), + temp.path().to_path_buf(), + temp.path().to_path_buf(), + scope, + WorkdirSessionCapabilities::ALL, + )); + let owner = RuntimeWorkspaceScope::new("workspace-a", "server-a"); + let state = RuntimeHttpState { + runtime: Runtime::with_execution_backend( + RuntimeOptions::default(), + Arc::new(AcceptingBackend), + ) + .expect("runtime"), + local_token: Some(Arc::from("token")), + auth: None, + workdir_sessions: Arc::new(Mutex::new(HashMap::from([( + "session-1".to_string(), + RuntimeHttpWorkdirSession { + owner: owner.clone(), + session: session.clone(), + }, + )]))), + }; + let auth = RuntimeAuthContext { + server_id: "server-a".to_string(), + workspace_id: "workspace-a".to_string(), + permissions: vec!["workdirs:operate".to_string()], + token_id: "token-a".to_string(), + expires_at: u64::MAX, + }; + let operation = WorkdirSessionOperation::Stat(StatRequest { + path: WorkdirPath::new("hello.txt").expect("logical path"), + }); + + let Json(result) = run_workdir_session_operation( + State(state.clone()), + Path("session-1".to_string()), + Some(Extension(auth.clone())), + Ok(Json(operation.clone())), + ) + .await + .expect("owned operation"); + assert!(matches!(result, WorkdirSessionOperationResult::Stat(_))); + + let wrong_owner = RuntimeAuthContext { + workspace_id: "workspace-b".to_string(), + ..auth.clone() + }; + let error = run_workdir_session_operation( + State(state.clone()), + Path("session-1".to_string()), + Some(Extension(wrong_owner)), + Ok(Json(operation)), + ) + .await + .expect_err("cross-workspace session access must fail"); + assert_eq!(error.status, StatusCode::NOT_FOUND); + + assert_eq!( + close_workdir_session( + State(state.clone()), + Path("session-1".to_string()), + Some(Extension(auth)), + ) + .await + .expect("close"), + StatusCode::NO_CONTENT + ); + let closed_error = session + .stat(StatRequest { + path: WorkdirPath::new("hello.txt").expect("logical path"), + }) + .await + .expect_err("close must be terminal"); + assert!(matches!( + closed_error, + workdir::WorkdirError::Unavailable(_) + )); + assert!( + state + .workdir_sessions + .lock() + .expect("session registry") + .is_empty() + ); + } + struct AcceptingBackend; impl WorkerExecutionBackend for AcceptingBackend { diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 0980e5af..b7b2efd9 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -360,6 +360,51 @@ impl Runtime { self.annotate_working_directory_status(status) } + /// Open a fresh Workdir operation session after proving that the persisted + /// materialization is assigned to a Worker in the authorized workspace. + pub fn open_workdir_session_scoped( + &self, + scope: &RuntimeWorkspaceScope, + working_directory_id: &str, + owner_worker_ref: Option<&WorkerRef>, + ) -> Result { + let backend = { + let mut state = self.lock()?; + state.ensure_running()?; + state.ensure_workspace_owner(scope, false)?; + + let owns_workdir = |worker: &WorkerRecord| { + worker.belongs_to_workspace(&scope.workspace_id) + && worker.working_directory.as_ref().is_some_and(|status| { + status.summary.working_directory_id == working_directory_id + }) + }; + let authorized = match owner_worker_ref { + Some(worker_ref) => state + .workers + .get(&worker_ref.worker_id) + .is_some_and(owns_workdir), + None => state.workers.values().any(owns_workdir), + }; + if !authorized { + return Err(RuntimeError::WorkingDirectory( + crate::working_directory::WorkingDirectoryDiagnostic::rejected( + "working_directory_not_found", + "working directory was not found in the authorized workspace", + ), + )); + } + state.execution_backend.clone().ok_or_else(|| { + RuntimeError::ExecutionBackendUnavailable { + message: "opening a Workdir session requires an execution backend".to_string(), + } + })? + }; + backend + .open_workdir_session(working_directory_id) + .map_err(RuntimeError::WorkingDirectory) + } + /// Cleanup a Runtime-owned working directory. pub fn cleanup_working_directory( &self, diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index fcf6b49d..8aa9c5c7 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -845,6 +845,31 @@ where materializer.working_directory_status(working_directory_id) } + fn open_workdir_session( + &self, + working_directory_id: &str, + ) -> Result { + let Some(materializer) = self.working_directory_materializer.as_ref() else { + return Err(WorkingDirectoryDiagnostic::rejected( + "working_directory_materializer_unavailable", + "Workdir session requested, but no materializer is configured for this runtime backend", + )); + }; + let binding = materializer.bind_working_directory(working_directory_id, None)?; + let scope = manifest::Scope::writable(binding.root()).map_err(|error| { + WorkingDirectoryDiagnostic::rejected( + "workdir_session_scope_invalid", + format!("failed to create Workdir session scope: {error}"), + ) + })?; + Ok(runtime_local_workdir_session( + working_directory_id, + binding.root(), + binding.cwd(), + manifest::SharedScope::new(scope), + )) + } + fn cleanup_working_directory( &self, working_directory_id: &str, diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 89a35e72..e22cbe8f 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -34,6 +34,7 @@ memory.workspace = true tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "sync", "time"] } tokio-tungstenite.workspace = true worker.workspace = true +workdir = { workspace = true, features = ["http-client"] } worker-runtime.workspace = true toml.workspace = true tracing.workspace = true diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index e0216136..73148196 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1,18 +1,24 @@ use crate::Error; use crate::resource_broker::BackendResourceBroker; use chrono::Utc; -use reqwest::StatusCode; use reqwest::blocking::{Client as BlockingHttpClient, RequestBuilder}; use reqwest::header::{AUTHORIZATION, CONTENT_TYPE}; +use reqwest::{Client as AsyncHttpClient, StatusCode, Url}; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use std::{ collections::BTreeMap, + future::Future, path::PathBuf, + pin::Pin, sync::{Arc, RwLock}, time::Duration, }; +use workdir::{ + Workdir, WorkdirError, + http::{OpenWorkdirSessionRequest, RemoteWorkdirSession, WorkdirHttpAuthorization}, +}; use worker_runtime::auth::{CapabilityTokenSigner, capability_claims}; use worker_runtime::catalog::{ ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef, @@ -690,6 +696,20 @@ pub trait WorkspaceWorkerRuntime: Send + Sync { } } + fn open_workdir_session<'a>( + &'a self, + _working_directory_id: &'a str, + _owner_worker_id: Option<&'a str>, + ) -> Pin< + Box> + Send + 'a>, + > { + Box::pin(async { + Err(WorkdirError::Unavailable( + "Runtime does not expose Workdir operation sessions".to_string(), + )) + }) + } + fn cleanup_working_directory( &self, working_directory_id: &str, @@ -2249,6 +2269,52 @@ impl RemoteRuntimeConfig { } } +#[derive(Clone)] +struct RemoteWorkdirAuthorization { + runtime_id: String, + workspace_id: String, + auth: Option, + fallback_bearer_token: Option, +} + +impl std::fmt::Debug for RemoteWorkdirAuthorization { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("RemoteWorkdirAuthorization") + .field("runtime_id", &self.runtime_id) + .field("workspace_id", &self.workspace_id) + .field("auth", &self.auth.as_ref().map(|_| "capability_token")) + .field( + "fallback_bearer_token", + &self.fallback_bearer_token.as_ref().map(|_| "configured"), + ) + .finish() + } +} + +impl WorkdirHttpAuthorization for RemoteWorkdirAuthorization { + fn bearer_token(&self) -> Result { + if let Some(auth) = self.auth.as_ref() { + let claims = capability_claims( + &auth.server_id, + &self.runtime_id, + &self.workspace_id, + all_remote_runtime_permissions(), + 300, + ) + .map_err(|error| WorkdirError::Unavailable(error.to_string()))?; + return CapabilityTokenSigner::new(&auth.server_id, &auth.server_private_key) + .sign(&claims) + .map_err(|error| WorkdirError::Unavailable(error.to_string())); + } + self.fallback_bearer_token.clone().ok_or_else(|| { + WorkdirError::Unavailable( + "remote Runtime does not have bearer authorization configured".to_string(), + ) + }) + } +} + #[derive(Clone)] pub struct RemoteWorkerRuntime { runtime_id: String, @@ -2263,6 +2329,7 @@ pub struct RemoteWorkerRuntime { host_id: String, resource_broker: BackendResourceBroker, http: BlockingHttpClient, + async_http: AsyncHttpClient, } fn all_remote_runtime_permissions() -> Vec { @@ -2274,6 +2341,7 @@ fn all_remote_runtime_permissions() -> Vec { "workers:input", "workers:stop", "workers:protocol", + "workdirs:operate", ] .into_iter() .map(str::to_string) @@ -2296,6 +2364,17 @@ impl RemoteWorkerRuntime { code: "remote_runtime_client_build_failed".to_string(), message: err.to_string(), })?; + // Workdir command-output waits are bounded to 20 seconds by Runtime; + // leave transport margin while retaining a finite client timeout. + let workdir_timeout = timeout.max(Duration::from_secs(30)); + let async_http = AsyncHttpClient::builder() + .timeout(workdir_timeout) + .build() + .map_err(|err| RuntimeRegistryError::RuntimeOperationFailed { + runtime_id: config.runtime_id.clone(), + code: "remote_runtime_async_client_build_failed".to_string(), + message: err.to_string(), + })?; Ok(Self { host_id: host_id_for_remote_runtime(&config.runtime_id), runtime_id: config.runtime_id, @@ -2309,6 +2388,7 @@ impl RemoteWorkerRuntime { cached_status: config.cached_status, resource_broker: BackendResourceBroker::default(), http, + async_http, }) } @@ -2317,6 +2397,33 @@ impl RemoteWorkerRuntime { self } + pub async fn open_workdir_session( + &self, + working_directory_id: &str, + owner_worker_id: Option<&str>, + ) -> Result { + let base_url = Url::parse(&self.base_url) + .map_err(|error| WorkdirError::InvalidArgument(error.to_string()))?; + let workdir_id = Workdir::new(working_directory_id).id().clone(); + let authorization: Arc = + Arc::new(RemoteWorkdirAuthorization { + runtime_id: self.runtime_id.clone(), + workspace_id: self.workspace_id.clone(), + auth: self.auth.clone(), + fallback_bearer_token: self.bearer_token.clone(), + }); + RemoteWorkdirSession::open_with_authorization( + self.async_http.clone(), + base_url, + authorization, + workdir_id, + OpenWorkdirSessionRequest { + owner_worker_id: owner_worker_id.map(str::to_string), + }, + ) + .await + } + fn endpoint(&self, path: &str) -> String { format!("{}{}", self.base_url, path) } @@ -2739,6 +2846,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { } } + fn open_workdir_session<'a>( + &'a self, + working_directory_id: &'a str, + owner_worker_id: Option<&'a str>, + ) -> Pin< + Box> + Send + 'a>, + > { + Box::pin(async move { + let session = RemoteWorkerRuntime::open_workdir_session( + self, + working_directory_id, + owner_worker_id, + ) + .await?; + Ok(Arc::new(session) as workdir::WorkdirSessionHandle) + }) + } + fn cleanup_working_directory( &self, working_directory_id: &str, @@ -4867,6 +4992,79 @@ mod tests { server.join().expect("mock remote server finished"); } + #[tokio::test(flavor = "multi_thread")] + async fn remote_workdir_session_uses_authenticated_http_operations_and_closes() { + use workdir::{EntryKind, StatRequest, StatResult, WorkdirPath}; + + let opened = workdir::http::OpenWorkdirSessionResponse { + session_id: workdir::http::WorkdirSessionId::new("session-1").unwrap(), + workdir_id: Workdir::new("wd-1").id().clone(), + capabilities: workdir::WorkdirSessionCapabilities::ALL, + }; + let stat = workdir::http::WorkdirSessionOperationResult::Stat(StatResult { + path: WorkdirPath::new("hello.txt").unwrap(), + kind: EntryKind::File, + size: 5, + }); + let (base_url, server) = serve_mock_http(vec![ + mock_response( + "POST", + "/v1/working-directories/wd-1/sessions", + true, + 200, + serde_json::to_string(&opened).unwrap(), + ), + mock_response( + "POST", + "/v1/workdir-sessions/session-1/operations", + true, + 200, + serde_json::to_string(&stat).unwrap(), + ), + mock_response( + "DELETE", + "/v1/workdir-sessions/session-1", + true, + 204, + String::new(), + ), + ]); + let runtime = RemoteWorkerRuntime::new( + RemoteRuntimeConfig::new( + "runtime-a", + "Runtime A", + base_url, + Some("secret-token".to_string()), + ), + "workspace-a".to_string(), + "http://backend.invalid".to_string(), + ) + .unwrap(); + + let runtime: Arc = Arc::new(runtime); + let session = runtime + .open_workdir_session("wd-1", Some("1")) + .await + .expect("open remote Workdir session"); + let result = session + .stat(StatRequest { + path: WorkdirPath::new("hello.txt").unwrap(), + }) + .await + .expect("remote stat"); + assert_eq!(result.size, 5); + session.close().await.expect("close remote session"); + session.close().await.expect("idempotent close"); + let error = session + .stat(StatRequest { + path: WorkdirPath::new("hello.txt").unwrap(), + }) + .await + .expect_err("closed session must reject local operation without another request"); + assert!(matches!(error, WorkdirError::Unavailable(_))); + server.join().expect("mock remote server finished"); + } + #[test] fn remote_runtime_auth_errors_map_to_typed_backend_error() { let (base_url, server) = serve_mock_http(vec![mock_response(