workdir: add authenticated runtime session transport
This commit is contained in:
Generated
+2
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<String>) -> Result<Self, WorkdirError> {
|
||||
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<String>,
|
||||
}
|
||||
|
||||
#[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("<remote>".into()),
|
||||
Code::Conflict => WorkdirError::Conflict(self.message),
|
||||
Code::Unsupported => WorkdirError::Unavailable(self.message),
|
||||
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".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<String, WorkdirError>;
|
||||
}
|
||||
|
||||
struct FixedBearerToken(Arc<str>);
|
||||
|
||||
impl std::fmt::Debug for FixedBearerToken {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
formatter.write_str("FixedBearerToken(<redacted>)")
|
||||
}
|
||||
}
|
||||
|
||||
impl WorkdirHttpAuthorization for FixedBearerToken {
|
||||
fn bearer_token(&self) -> Result<String, WorkdirError> {
|
||||
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<dyn WorkdirHttpAuthorization>,
|
||||
workdir: Workdir,
|
||||
session_id: WorkdirSessionId,
|
||||
capabilities: WorkdirSessionCapabilities,
|
||||
closed: AtomicBool,
|
||||
}
|
||||
|
||||
impl RemoteWorkdirSession {
|
||||
pub async fn open(
|
||||
client: Client,
|
||||
base_url: Url,
|
||||
bearer_token: impl Into<Arc<str>>,
|
||||
workdir_id: WorkdirId,
|
||||
request: OpenWorkdirSessionRequest,
|
||||
) -> Result<Self, WorkdirError> {
|
||||
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<dyn WorkdirHttpAuthorization>,
|
||||
workdir_id: WorkdirId,
|
||||
request: OpenWorkdirSessionRequest,
|
||||
) -> Result<Self, WorkdirError> {
|
||||
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<WorkdirSessionOperationResult, WorkdirError> {
|
||||
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<StatResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::Stat(request)).await? {
|
||||
WorkdirSessionOperationResult::Stat(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("stat")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::Read(request)).await? {
|
||||
WorkdirSessionOperationResult::Read(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("read")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError> {
|
||||
match self
|
||||
.operate(WorkdirSessionOperation::Write(request))
|
||||
.await?
|
||||
{
|
||||
WorkdirSessionOperationResult::Write(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("write")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn edit(&self, request: EditRequest) -> Result<EditResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::Edit(request)).await? {
|
||||
WorkdirSessionOperationResult::Edit(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("edit")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::List(request)).await? {
|
||||
WorkdirSessionOperationResult::List(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("list")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::Glob(request)).await? {
|
||||
WorkdirSessionOperationResult::Glob(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("glob")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
|
||||
match self.operate(WorkdirSessionOperation::Grep(request)).await? {
|
||||
WorkdirSessionOperationResult::Grep(result) => Ok(result),
|
||||
_ => Err(Self::mismatch("grep")),
|
||||
}
|
||||
}
|
||||
|
||||
async fn start_command(
|
||||
&self,
|
||||
request: CommandRequest,
|
||||
) -> Result<CommandHandle, WorkdirError> {
|
||||
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<CommandStatus, WorkdirError> {
|
||||
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<CommandOutput, WorkdirError> {
|
||||
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<Url, WorkdirError> {
|
||||
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<T: serde::de::DeserializeOwned>(
|
||||
response: reqwest::Response,
|
||||
) -> Result<T, WorkdirError> {
|
||||
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::<WorkdirTransportError>()
|
||||
.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"));
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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<WorkdirSessionHandle, WorkingDirectoryDiagnostic> {
|
||||
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<WorkdirSessionHandle, WorkingDirectoryDiagnostic> {
|
||||
self.backend.open_workdir_session(working_directory_id)
|
||||
}
|
||||
|
||||
pub(crate) fn cleanup_working_directory(
|
||||
&self,
|
||||
working_directory_id: &str,
|
||||
|
||||
@@ -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::<str>::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<Arc<str>>,
|
||||
auth: Option<Arc<RuntimeHttpAuthConfig>>,
|
||||
workdir_sessions: Arc<Mutex<HashMap<String, RuntimeHttpWorkdirSession>>>,
|
||||
}
|
||||
|
||||
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<RuntimeHttpState>,
|
||||
Path(working_directory_id): Path<String>,
|
||||
auth: Option<Extension<RuntimeAuthContext>>,
|
||||
body: Result<Json<OpenWorkdirSessionRequest>, JsonRejection>,
|
||||
) -> Result<Json<OpenWorkdirSessionResponse>, 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<RuntimeHttpState>,
|
||||
Path(session_id): Path<String>,
|
||||
auth: Option<Extension<RuntimeAuthContext>>,
|
||||
body: Result<Json<WorkdirSessionOperation>, JsonRejection>,
|
||||
) -> Result<Json<WorkdirSessionOperationResult>, 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<RuntimeHttpState>,
|
||||
Path(session_id): Path<String>,
|
||||
auth: Option<Extension<RuntimeAuthContext>>,
|
||||
) -> Result<StatusCode, RuntimeHttpWorkdirError> {
|
||||
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<Extension<RuntimeAuthContext>>,
|
||||
) -> Result<RuntimeWorkspaceScope, RuntimeHttpWorkdirError> {
|
||||
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<RuntimeHttpState>,
|
||||
Path(working_directory_id): Path<String>,
|
||||
@@ -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<String>,
|
||||
) -> 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<workdir::WorkdirError> 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 {
|
||||
|
||||
@@ -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<workdir::WorkdirSessionHandle, RuntimeError> {
|
||||
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,
|
||||
|
||||
@@ -845,6 +845,31 @@ where
|
||||
materializer.working_directory_status(working_directory_id)
|
||||
}
|
||||
|
||||
fn open_workdir_session(
|
||||
&self,
|
||||
working_directory_id: &str,
|
||||
) -> Result<WorkdirSessionHandle, WorkingDirectoryDiagnostic> {
|
||||
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<dyn Future<Output = Result<workdir::WorkdirSessionHandle, WorkdirError>> + 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<RemoteRuntimeAuthConfig>,
|
||||
fallback_bearer_token: Option<String>,
|
||||
}
|
||||
|
||||
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<String, WorkdirError> {
|
||||
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<String> {
|
||||
@@ -2274,6 +2341,7 @@ fn all_remote_runtime_permissions() -> Vec<String> {
|
||||
"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<RemoteWorkdirSession, WorkdirError> {
|
||||
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<dyn WorkdirHttpAuthorization> =
|
||||
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<dyn Future<Output = Result<workdir::WorkdirSessionHandle, WorkdirError>> + 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<dyn WorkspaceWorkerRuntime> = 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(
|
||||
|
||||
Reference in New Issue
Block a user