Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

chore(ticket): incorporate integrated Drive and structured WIP contracts

Browse Source 
# Conflicts:
#	openapi/server-api.json
#	web/workspace/src/lib/generated/companion-api.ts
#	web/workspace/src/lib/generated/repository-api.ts
#	web/workspace/src/lib/generated/runtime-api.ts
#	web/workspace/src/lib/generated/ticket-api.ts
#	web/workspace/src/lib/generated/worker-launch-api.ts
This commit is contained in:
Hare
2026-10-09 02:54:09 +09:00
parent 5244308c0d 32c2227e8e
commit e67d75615d
62 changed files with 11501 additions and 1008 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/Cargo.toml
+1
View File
@@ -11,6 +11,7 @@ name = "yoi-server"
path = "src/main.rs"
[dependencies]
api-macros = { workspace = true, features = ["axum"] }
async-trait.workspace = true
axum = { workspace = true, features = ["ws"] }
chrono = { version = "0.4", default-features = false, features = ["clock"] }
crates/workspace-server/src/latest_schema.sql
+25
View File
@@ -1629,3 +1629,28 @@ CREATE TABLE workspace_config_grants (
);
CREATE UNIQUE INDEX workspace_config_grants_active_worker
ON workspace_config_grants(workspace_id, runtime_id, worker_id) WHERE revoked = 0;
CREATE TABLE workspace_drive_grants (
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
grant_id INTEGER PRIMARY KEY AUTOINCREMENT,
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
access TEXT NOT NULL CHECK(access IN ('read_only', 'read_write')),
revoked INTEGER NOT NULL DEFAULT 0 CHECK(revoked IN (0, 1)),
created_by TEXT NOT NULL,
created_at TEXT NOT NULL,
revoked_by TEXT,
revoked_at TEXT,
CHECK((revoked=0 AND revoked_by IS NULL AND revoked_at IS NULL)
OR (revoked=1 AND revoked_by IS NOT NULL AND revoked_at IS NOT NULL))
);
CREATE UNIQUE INDEX workspace_drive_grants_active_worker
ON workspace_drive_grants(workspace_id, runtime_id, worker_id) WHERE revoked=0;
-- Retain grant audit across Worker deletion, but never carry its authority into
-- a later lifetime reusing the same Runtime/Worker identity.
CREATE TRIGGER workspace_drive_grants_worker_deleted BEFORE DELETE ON worker_registry BEGIN
UPDATE workspace_drive_grants SET revoked=1, revoked_by='server:worker-removal',
revoked_at=strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE workspace_id=OLD.workspace_id AND runtime_id=OLD.runtime_id
AND worker_id=OLD.worker_id AND revoked=0;
END;
crates/workspace-server/src/server.rs
+131
View File
@@ -6,6 +6,10 @@ pub(crate) use worker_operations::{
WorkerOperationContext, WorkspaceWorker, WorkspaceWorkerMethodSender,
};
#[path = "server_drive.rs"]
mod drive;
#[path = "server_drive_grants.rs"]
mod drive_grants;
#[path = "server_workspace_config.rs"]
mod workspace_config;
use std::collections::{BTreeMap, HashMap, HashSet};
@@ -2601,6 +2605,9 @@ fn signed_request_target(uri: &Uri) -> &str {
}
fn repository_api_rejection(path: &str, status: StatusCode, message: &str) -> Response {
if path.starts_with("/api/w/") && path.contains("/drive/") {
return (status, Json(drive::failure(status.as_u16()))).into_response();
}
if path == "/api/repositories"
|| path.starts_with("/api/repositories/")
|| (path.starts_with("/api/w/") && path.contains("/repositories"))
@@ -6303,6 +6310,7 @@ fn generated_workspace_catalog_contract_router(service: ServerApiContractService
fn generated_workspace_contract_router(service: ServerApiContractService) -> Router {
let service = Arc::new(service);
Router::new()
.merge(drive::router(service.clone()))
.merge(server_api::server_api_axum::workspace_current(
service.clone(),
))
@@ -11112,6 +11120,128 @@ impl server_api::ServerApi for ServerApiContractService {
.map_err(ApiError::into_repository_api_error)
}
async fn drive_root(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
) -> std::result::Result<server_api::DriveEntry, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::root(api, &context, &workspace_id).await
}
async fn drive_metadata(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveEntryQuery,
) -> std::result::Result<server_api::DriveEntry, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::metadata(api, &context, &workspace_id, query).await
}
async fn drive_list(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveListQuery,
) -> std::result::Result<server_api::DriveListResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::list(api, &context, &workspace_id, query).await
}
async fn drive_search(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveSearchQuery,
) -> std::result::Result<server_api::DriveListResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::search(api, &context, &workspace_id, query).await
}
async fn drive_read_text(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveReadTextQuery,
) -> std::result::Result<server_api::DriveReadTextResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::read_text(api, &context, &workspace_id, query).await
}
async fn drive_read_chunk(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveReadChunkQuery,
) -> std::result::Result<server_api::BinaryBody, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::read_chunk(api, &context, &workspace_id, query).await
}
async fn drive_download(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveDownloadQuery,
) -> std::result::Result<
server_api::server_api_responses::DriveDownload,
server_api::DriveApiError,
> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::download(api, &context, &workspace_id, query).await
}
async fn drive_mutate(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
request: server_api::DriveMutationRequest,
) -> std::result::Result<server_api::DriveMutationResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::mutate(api, &context, &workspace_id, request).await
}
async fn drive_upload(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveUploadQuery,
body: server_api::BinaryBody,
) -> std::result::Result<server_api::DriveMutationResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::upload(api, &context, &workspace_id, query, body).await
}
async fn drive_request_status(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
request_id: String,
) -> std::result::Result<server_api::DriveRequestStatusResponse, server_api::DriveApiError>
{
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive::request_status(api, &context, &workspace_id, request_id).await
}
async fn drive_grant_create(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
request: server_api::DriveGrantCreateRequest,
) -> std::result::Result<server_api::DriveGrantResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive_grants::create_grant(api, &context, &workspace_id, request).await
}
async fn drive_grant_revoke(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
grant_id: String,
) -> std::result::Result<server_api::DriveGrantResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive_grants::revoke_grant(api, &context, &workspace_id, &grant_id).await
}
async fn drive_grant_list(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::DriveGrantListQuery,
) -> std::result::Result<server_api::DriveGrantListResponse, server_api::DriveApiError> {
let api = self.workspace_api().map_err(|_| drive::failure(404))?;
drive_grants::list_grants(api, &context, &workspace_id, query).await
}
async fn current_worker_workspace_config_get(
&self,
context: server_api::ServerRequestContext,
@@ -32591,6 +32721,7 @@ impl IntoResponse for ApiError {
#[cfg(test)]
mod tests {
mod auth_logging_tests;
mod drive_api_tests;
mod drive_tests;
mod subject_spawn_tests;
mod subjektiv_jobs_tests;
crates/workspace-server/src/server/tests/drive_api_tests.rs
+1006
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/server_drive.rs
+736
View File
@@ -0,0 +1,736 @@
//! Authorized Workspace document API. Storage, CAS and receipts stay in workspace-drive.
//! HTTP upload consumes frames with a hard bound; DB publication occurs only after validation.
use super::*;
use futures::StreamExt;
use server_api::*;
use sha2::{Digest, Sha256};
use workspace_drive::{Drive, Mutation, NodeId, Revision};
fn mutation_actor(context: &ServerRequestContext) -> DriveResult<String> {
if let Some(source) = &context.runtime_source {
let worker = source.worker_id.as_ref().ok_or_else(|| failure(403))?;
Ok(format!(
"worker:{}",
serde_json::to_string(&(&source.runtime_id, worker)).map_err(|_| failure(400))?
))
} else {
Ok(format!(
"user:{}",
context.actor.as_ref().ok_or_else(|| failure(403))?.user_id
))
}
}
type DriveResult<T> = std::result::Result<T, DriveApiError>;
const TRANSFER_CHUNK: usize = 64 * 1024;
pub(super) fn failure(status: u16) -> DriveApiError {
let code = match status {
401 | 403 => DriveApiErrorCode::Denied,
404 => DriveApiErrorCode::NotFound,
409 => DriveApiErrorCode::Conflict,
413 => DriveApiErrorCode::Limit,
500..=599 => DriveApiErrorCode::StorageUnavailable,
_ => DriveApiErrorCode::Invalid,
};
DriveApiError::new(code)
}
fn domain_error(error: workspace_drive::Error, mutation: bool) -> DriveApiError {
use workspace_drive::Error;
match error {
Error::Denied => failure(403),
Error::NotFound => failure(404),
Error::Conflict | Error::Fenced => failure(409),
Error::Invalid(_) => failure(400),
Error::Blob(_) => failure(503),
_ if mutation => DriveApiError::new(DriveApiErrorCode::OutcomeUnknown),
_ => failure(503),
}
}
async fn blocking<T: Send + 'static>(
mutation: bool,
action: impl FnOnce() -> workspace_drive::Result<T> + Send + 'static,
) -> DriveResult<T> {
tokio::task::spawn_blocking(action)
.await
.map_err(|_| {
if mutation {
DriveApiError::new(DriveApiErrorCode::OutcomeUnknown)
} else {
failure(503)
}
})?
.map_err(|error| domain_error(error, mutation))
}
fn node_id(value: &str) -> DriveResult<NodeId> {
value.to_owned().try_into().map_err(|_| failure(400))
}
fn revision(value: &str) -> DriveResult<Revision> {
value.to_owned().try_into().map_err(|_| failure(400))
}
fn entry(workspace: &str, reference: &DriveEntryRef) -> DriveResult<NodeId> {
if reference.workspace_id != workspace {
return Err(failure(403));
}
node_id(&reference.node_id)
}
fn scoped_id(workspace: &str, entry_workspace: &str, id: &str) -> DriveResult<NodeId> {
if workspace != entry_workspace {
return Err(failure(403));
}
node_id(id)
}
fn safe_content_type(value: &str) -> DriveResult<String> {
// Deliberately accept a bare MIME type only, not attacker-controlled header parameters.
if value.len() > 128
|| !value.contains('/')
|| value.matches('/').count() != 1
|| value.split('/').any(str::is_empty)
|| !value
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b"!#$&^_.+-/".contains(&b))
{
return Err(failure(400));
}
Ok(value.to_ascii_lowercase())
}
fn disposition(name: &str) -> String {
// ASCII fallback plus RFC 5987 UTF-8 filename. Never place raw names in headers.
let encoded: String = name
.as_bytes()
.iter()
.map(|b| {
if b.is_ascii_alphanumeric() || b"-._".contains(b) {
(*b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect();
format!("attachment; filename=\"download\"; filename*=UTF-8''{encoded}")
}
fn error_response(error: DriveApiError) -> Response {
let status = match error.code {
DriveApiErrorCode::Denied => 403,
DriveApiErrorCode::NotFound => 404,
DriveApiErrorCode::Conflict => 409,
DriveApiErrorCode::Limit => 413,
DriveApiErrorCode::Invalid => 400,
DriveApiErrorCode::StorageUnavailable | DriveApiErrorCode::OutcomeUnknown => 503,
};
(StatusCode::from_u16(status).unwrap(), Json(error)).into_response()
}
// No raw storage diagnostic is ever placed in a public error or stream failure.
fn stream_error() -> std::io::Error {
std::io::Error::other("Drive transfer unavailable")
}
/// Stream HTTP frames while enforcing both announced and observed limits. Storage currently
/// accepts a bounded whole-file Vec; this is binary buffering, never JSON/base64 expansion.
async fn receive(body: axum::body::Body, declared: u64, digest: &str) -> DriveResult<Vec<u8>> {
if declared > workspace_drive::MAX_FILE_BYTES as u64 {
return Err(failure(413));
}
if digest.len() != 64
|| !digest
.bytes()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
{
return Err(failure(400));
}
let mut stream = body.into_data_stream();
let mut bytes = Vec::new();
let mut hasher = Sha256::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|_| failure(400))?;
let size = bytes
.len()
.checked_add(chunk.len())
.ok_or_else(|| failure(413))?;
if size > workspace_drive::MAX_FILE_BYTES {
return Err(failure(413));
}
if size > declared as usize {
return Err(failure(400));
}
hasher.update(&chunk);
bytes.extend_from_slice(&chunk);
}
if bytes.len() != declared as usize
|| hasher
.finalize()
.iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
!= digest
{
return Err(failure(400));
}
Ok(bytes)
}
fn public_node(node: workspace_drive::Node) -> DriveEntry {
let workspace_id = node.workspace_id.clone();
let time = chrono::DateTime::from_timestamp_millis(node.updated_at_ms)
.map(|t| t.to_rfc3339())
.unwrap_or_default();
DriveEntry {
entry: DriveEntryRef {
workspace_id: workspace_id.clone(),
node_id: node.id.get().to_string(),
},
parent: node.parent_id.map(|id| DriveEntryRef {
workspace_id,
node_id: id.get().to_string(),
}),
name: node.name,
kind: match node.kind {
workspace_drive::Kind::File => DriveEntryKind::File,
workspace_drive::Kind::Directory => DriveEntryKind::Folder,
},
revision: node.revision.get().to_string(),
size: (node.kind == workspace_drive::Kind::File).then_some(node.size as u32),
content_type: node.content_type,
updated_by: node.updated_by,
updated_at: time,
latest_url: format!(
"/api/w/{}/drive/{}?entry_workspace_id={}&id={}",
node.workspace_id,
if node.kind == workspace_drive::Kind::File {
"download"
} else {
"metadata"
},
node.workspace_id,
node.id.get()
),
}
}
fn cursor(workspace: &str, id: Option<NodeId>) -> Option<String> {
id.map(|id| format!("{workspace}:{}", id.get()))
}
fn after(workspace: &str, value: Option<String>) -> DriveResult<Option<NodeId>> {
value
.map(|value| {
let (scope, id) = value.rsplit_once(':').ok_or_else(|| failure(400))?;
scoped_id(workspace, scope, id)
})
.transpose()
}
fn limit(value: Option<u32>, maximum: usize) -> DriveResult<usize> {
let limit = value.unwrap_or(maximum as u32) as usize;
if limit == 0 || limit > maximum {
return Err(failure(413));
}
Ok(limit)
}
pub(super) async fn root(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
) -> DriveResult<DriveEntry> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
blocking(false, move || drive.root()).await.map(public_node)
}
pub(super) async fn metadata(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveEntryQuery,
) -> DriveResult<DriveEntry> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
blocking(false, move || drive.metadata(id))
.await
.map(public_node)
}
pub(super) async fn list(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveListQuery,
) -> DriveResult<DriveListResponse> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
let after = after(workspace, query.after)?;
let limit = limit(query.limit, workspace_drive::MAX_PAGE)?;
let page = blocking(false, move || drive.list(id, after, limit)).await?;
Ok(DriveListResponse {
entries: page.nodes.into_iter().map(public_node).collect(),
next_after: cursor(workspace, page.next_after),
})
}
pub(super) async fn search(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveSearchQuery,
) -> DriveResult<DriveListResponse> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let after = after(workspace, query.after)?;
let limit = limit(query.limit, workspace_drive::MAX_SEARCH_NODES)?;
let page = blocking(false, move || {
drive.search(&query.query, query.include_text, after, limit)
})
.await?;
Ok(DriveListResponse {
entries: page.nodes.into_iter().map(public_node).collect(),
next_after: cursor(workspace, page.next_after),
})
}
pub(super) async fn read_text(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveReadTextQuery,
) -> DriveResult<DriveReadTextResponse> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
let max = query.max_bytes as usize;
if max == 0 || max > DRIVE_TEXT_MAX_BYTES {
return Err(failure(413));
}
let (node, chunk) = blocking(false, move || {
let node = drive.metadata(id)?;
let chunk = drive.read(id, node.revision, 0, max)?;
Ok((node, chunk))
})
.await?;
let text = match std::str::from_utf8(&chunk.bytes) {
Ok(text) => text,
Err(error) if !chunk.eof && error.error_len().is_none() => {
std::str::from_utf8(&chunk.bytes[..error.valid_up_to()]).map_err(|_| failure(400))?
}
Err(_) => return Err(failure(400)),
};
Ok(DriveReadTextResponse {
entry: public_node(node),
text: text.into(),
truncated: !chunk.eof,
})
}
pub(super) async fn read_chunk(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveReadChunkQuery,
) -> DriveResult<BinaryBody> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
let revision = revision(&query.expected_revision)?;
if query.length == 0 || query.length > DRIVE_CHUNK_MAX_BYTES {
return Err(failure(413));
}
let chunk = blocking(false, move || {
drive.read(id, revision, query.offset.into(), query.length as usize)
})
.await?;
Ok(chunk.bytes.into())
}
fn text_bytes(text: String) -> DriveResult<Vec<u8>> {
if text.len() > DRIVE_TEXT_MAX_BYTES {
return Err(failure(413));
}
Ok(text.into_bytes())
}
fn mutation(workspace: &str, mutation: DriveMutation) -> DriveResult<Mutation> {
Ok(match mutation {
DriveMutation::CreateFolder { parent, name } => Mutation::CreateFolder {
parent: entry(workspace, &parent)?,
name,
},
DriveMutation::CreateText {
parent,
name,
text,
content_type,
} => Mutation::CreateFile {
parent: entry(workspace, &parent)?,
name,
content_type: safe_content_type(&content_type)?,
bytes: text_bytes(text)?,
},
DriveMutation::UpdateText {
id,
expected_revision,
text,
content_type,
} => Mutation::Update {
id: entry(workspace, &id)?,
expected_revision: revision(&expected_revision)?,
content_type: safe_content_type(&content_type)?,
bytes: text_bytes(text)?,
},
DriveMutation::Relocate {
id,
expected_revision,
parent,
name,
} => Mutation::Relocate {
id: entry(workspace, &id)?,
expected_revision: revision(&expected_revision)?,
parent: entry(workspace, &parent)?,
name,
},
DriveMutation::Delete {
id,
expected_revision,
} => Mutation::Delete {
id: entry(workspace, &id)?,
expected_revision: revision(&expected_revision)?,
},
})
}
fn result(request_id: String, result: workspace_drive::MutationResult) -> DriveMutationResponse {
DriveMutationResponse {
request_id,
entry: (!result.deleted).then(|| public_node(result.node)),
}
}
pub(super) async fn mutate(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
request: DriveMutationRequest,
) -> DriveResult<DriveMutationResponse> {
let drive = drive_grants::authorized_drive(api, context, workspace, true).await?;
let actor = mutation_actor(context)?;
let operation = mutation(workspace, request.mutation)?;
let request_id = request.request_id.clone();
let mutated = blocking(true, move || {
drive.mutate(&request.request_id, &actor, operation)
})
.await?;
Ok(result(request_id, mutated))
}
fn upload_mutation(
workspace: &str,
query: &DriveUploadQuery,
bytes: Vec<u8>,
) -> DriveResult<Mutation> {
if query.entry_workspace_id != workspace {
return Err(failure(403));
}
let content_type = safe_content_type(&query.content_type)?;
match query.operation {
DriveUploadOperation::Create if query.id.is_none() && query.expected_revision.is_none() => {
Ok(Mutation::CreateFile {
parent: node_id(query.parent_id.as_deref().ok_or_else(|| failure(400))?)?,
name: query.name.clone().ok_or_else(|| failure(400))?,
content_type,
bytes,
})
}
DriveUploadOperation::Update if query.parent_id.is_none() && query.name.is_none() => {
Ok(Mutation::Update {
id: node_id(query.id.as_deref().ok_or_else(|| failure(400))?)?,
expected_revision: revision(
query
.expected_revision
.as_deref()
.ok_or_else(|| failure(400))?,
)?,
content_type,
bytes,
})
}
_ => Err(failure(400)),
}
}
pub(super) async fn upload(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveUploadQuery,
body: BinaryBody,
) -> DriveResult<DriveMutationResponse> {
let drive = drive_grants::authorized_drive(api, context, workspace, true).await?;
let actor = mutation_actor(context)?;
let bytes = receive(
axum::body::Body::from(body.into_bytes()),
query.size.into(),
&query.sha256,
)
.await?;
publish_upload(drive, actor, workspace, query, bytes).await
}
async fn publish_upload(
drive: Drive,
actor: String,
workspace: &str,
query: DriveUploadQuery,
bytes: Vec<u8>,
) -> DriveResult<DriveMutationResponse> {
let mutation = upload_mutation(workspace, &query, bytes)?;
let request_id = query.request_id.clone();
let mutated = blocking(true, move || {
drive.mutate(&query.request_id, &actor, mutation)
})
.await?;
Ok(result(request_id, mutated))
}
pub(super) async fn request_status(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
request_id: String,
) -> DriveResult<DriveRequestStatusResponse> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let id = request_id.clone();
let status = blocking(false, move || drive.request_status(&id)).await?;
Ok(match status {
workspace_drive::RequestStatus::Uncommitted => DriveRequestStatusResponse {
request_id,
state: DriveRequestState::Uncommitted,
response: None,
},
workspace_drive::RequestStatus::Committed { result: mutation } => {
DriveRequestStatusResponse {
request_id: request_id.clone(),
state: DriveRequestState::Committed,
response: Some(result(request_id, mutation)),
}
}
})
}
async fn select_download(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveDownloadQuery,
) -> DriveResult<(Drive, workspace_drive::Node)> {
let drive = drive_grants::authorized_drive(api, context, workspace, false).await?;
let id = scoped_id(workspace, &query.entry_workspace_id, &query.id)?;
let for_node = drive.clone();
let node = blocking(false, move || for_node.metadata(id)).await?;
if node.kind != workspace_drive::Kind::File {
return Err(failure(400));
}
if query
.expected_revision
.as_deref()
.is_some_and(|v| v != node.revision.get().to_string())
{
return Err(failure(409));
}
safe_content_type(node.content_type.as_deref().unwrap_or("")).map_err(|_| failure(503))?;
Ok((drive, node))
}
pub(super) async fn download(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace: &str,
query: DriveDownloadQuery,
) -> DriveResult<server_api::server_api_responses::DriveDownload> {
let (drive, node) = select_download(api, context, workspace, query).await?;
let mut bytes = Vec::new();
let mut offset = 0;
loop {
let copy = drive.clone();
let chunk = blocking(false, move || {
copy.read(node.id, node.revision, offset, TRANSFER_CHUNK)
})
.await?;
offset += chunk.bytes.len() as u64;
bytes.extend_from_slice(&chunk.bytes);
if chunk.eof {
break;
}
}
Ok(server_api::server_api_responses::DriveDownload::Status200 {
body: bytes.into(),
header_content_disposition: disposition(&node.name),
header_cache_control: "private, no-store".into(),
header_etag: validator(&node),
header_x_content_type_options: "nosniff".into(),
header_content_security_policy: "sandbox; default-src 'none'".into(),
})
}
fn validator(node: &workspace_drive::Node) -> String {
format!(
"\"drive-{}-{}-{}\"",
node.workspace_id,
node.id.get(),
node.revision.get()
)
}
pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
Router::new()
.merge(streaming_router(service.clone()))
.merge(server_api::server_api_axum::drive_root(service.clone()))
.merge(server_api::server_api_axum::drive_metadata(service.clone()))
.merge(server_api::server_api_axum::drive_list(service.clone()))
.merge(server_api::server_api_axum::drive_search(service.clone()))
.merge(server_api::server_api_axum::drive_read_text(
service.clone(),
))
.merge(server_api::server_api_axum::drive_read_chunk(
service.clone(),
))
.merge(server_api::server_api_axum::drive_mutate(service.clone()))
.merge(server_api::server_api_axum::drive_request_status(
service.clone(),
))
.merge(server_api::server_api_axum::drive_grant_create(
service.clone(),
))
.merge(server_api::server_api_axum::drive_grant_revoke(
service.clone(),
))
.merge(server_api::server_api_axum::drive_grant_list(
service.clone(),
))
.layer(DefaultBodyLimit::max(512 * 1024))
.layer(middleware::from_fn(normalize_rejections))
}
async fn normalize_rejections(request: Request, next: Next) -> Response {
let response = next.run(request).await;
if matches!(response.status().as_u16(), 415 | 422) {
error_response(failure(400))
} else if matches!(response.status().as_u16(), 400 | 413)
&& response
.headers()
.get(CONTENT_TYPE)
.is_none_or(|value| value != "application/json")
{
error_response(failure(response.status().as_u16()))
} else {
response
}
}
fn streaming_router(service: Arc<ServerApiContractService>) -> Router {
// The operation inventory is the route authority, not duplicated string literals.
use api_macros::Operation;
use axum::routing::put;
use server_api::server_api_operations::{DriveDownload, DriveUpload};
Router::new()
.route(DriveUpload::METADATA.path, put(streaming_upload))
.route(DriveDownload::METADATA.path, get(streaming_download))
.with_state(service)
}
async fn streaming_upload(
State(service): State<Arc<ServerApiContractService>>,
AxumPath(workspace): AxumPath<String>,
Extension(context): Extension<ServerRequestContext>,
query: std::result::Result<Query<DriveUploadQuery>, axum::extract::rejection::QueryRejection>,
request: Request,
) -> Response {
let result = async {
let api = service.workspace_api().map_err(|_| failure(404))?;
let drive = drive_grants::authorized_drive(api, &context, &workspace, true).await?;
let actor = mutation_actor(&context)?;
let Query(query) = query.map_err(|_| failure(400))?;
// Reject malformed metadata, foreign refs and over-limit declarations before reading.
upload_mutation(&workspace, &query, Vec::new())?;
if query.size as usize > workspace_drive::MAX_FILE_BYTES {
return Err(failure(413));
}
if let Some(length) = request.headers().get(axum::http::header::CONTENT_LENGTH) {
let length: u64 = length
.to_str()
.ok()
.and_then(|s| s.parse().ok())
.ok_or_else(|| failure(400))?;
if length > workspace_drive::MAX_FILE_BYTES as u64 {
return Err(failure(413));
}
if length != u64::from(query.size) {
return Err(failure(400));
}
}
let bytes = receive(request.into_body(), query.size.into(), &query.sha256).await?;
// The bound Drive authorizer runs again under attached-authority IMMEDIATE,
// so a grant revoked during reception cannot publish or replay a receipt.
publish_upload(drive, actor, &workspace, query, bytes).await
}
.await;
match result {
Ok(result) => Json(result).into_response(),
Err(error) => error_response(error),
}
}
async fn streaming_download(
State(service): State<Arc<ServerApiContractService>>,
AxumPath(workspace): AxumPath<String>,
Extension(context): Extension<ServerRequestContext>,
query: std::result::Result<Query<DriveDownloadQuery>, axum::extract::rejection::QueryRejection>,
) -> Response {
let result = async {
let api = service.workspace_api().map_err(|_| failure(404))?;
let Query(query) = query.map_err(|_| failure(400))?;
let (drive, node) = select_download(api, &context, &workspace, query).await?;
// Validate the referenced blob before HTTP success, including a zero-byte file.
let first_drive = drive.clone();
let first = blocking(false, move || {
first_drive.read(node.id, node.revision, 0, TRANSFER_CHUNK)
})
.await?;
let id = node.id;
let revision = node.revision;
let state = (drive, id, revision, Some(first), 0_u64, false);
let stream = futures::stream::try_unfold(
state,
|(drive, id, revision, first, offset, done)| async move {
if done {
return Ok(None);
}
let chunk = match first {
Some(chunk) => chunk,
None => {
let copy = drive.clone();
blocking(false, move || {
copy.read(id, revision, offset, TRANSFER_CHUNK)
})
.await
.map_err(|_| stream_error())?
}
};
let next = offset + chunk.bytes.len() as u64;
let eof = chunk.eof;
Ok::<_, std::io::Error>(Some((
axum::body::Bytes::from(chunk.bytes),
(drive, id, revision, None, next, eof),
)))
},
);
let mut response = axum::body::Body::from_stream(stream).into_response();
let headers = response.headers_mut();
headers.insert(
CONTENT_TYPE,
HeaderValue::from_str(&safe_content_type(
node.content_type.as_deref().unwrap_or(""),
)?)
.map_err(|_| failure(503))?,
);
headers.insert(
axum::http::header::CONTENT_LENGTH,
HeaderValue::from_str(&node.size.to_string()).map_err(|_| failure(503))?,
);
headers.insert(
axum::http::header::CONTENT_DISPOSITION,
HeaderValue::from_str(&disposition(&node.name)).map_err(|_| failure(503))?,
);
headers.insert(
axum::http::header::ETAG,
HeaderValue::from_str(&validator(&node)).map_err(|_| failure(503))?,
);
headers.insert(CACHE_CONTROL, HeaderValue::from_static("private, no-store"));
headers.insert(
"x-content-type-options",
HeaderValue::from_static("nosniff"),
);
headers.insert(
"content-security-policy",
HeaderValue::from_static("sandbox; default-src 'none'"),
);
Ok::<_, DriveApiError>(response)
}
.await;
match result {
Ok(response) => response,
Err(error) => error_response(error),
}
}
crates/workspace-server/src/server_drive_grants.rs
+229
View File
@@ -0,0 +1,229 @@
//! Drive authority is rechecked inside the metadata transaction against the
//! attached Server DB. A bound handle is not an authorization cache or lease.
use super::*;
use rusqlite::params;
use server_api::{
DriveApiError, DriveApiErrorCode, DriveGrantCreateRequest, DriveGrantListQuery,
DriveGrantListResponse, DriveGrantResponse, ServerRequestContext,
};
type DriveResult<T> = std::result::Result<T, DriveApiError>;
fn fail(code: DriveApiErrorCode) -> DriveApiError {
DriveApiError::new(code)
}
pub(super) fn pre_error(error: impl Into<ApiError>) -> DriveApiError {
let error = error.into();
fail(match api_error_status(&error.error).as_u16() {
401 | 403 => DriveApiErrorCode::Denied,
404 => DriveApiErrorCode::NotFound,
409 => DriveApiErrorCode::Conflict,
413 => DriveApiErrorCode::Limit,
500..=599 => DriveApiErrorCode::StorageUnavailable,
_ => DriveApiErrorCode::Invalid,
})
}
fn mutation_error(error: Error) -> DriveApiError {
if matches!(error, Error::Sqlite(_) | Error::Store(_)) {
fail(DriveApiErrorCode::OutcomeUnknown)
} else {
pre_error(error)
}
}
fn worker_request(context: &ServerRequestContext) -> bool {
context.runtime_source.is_some()
|| context.worker_source.is_some()
|| context.transport_headers.iter().any(|(name, _)| {
name.eq_ignore_ascii_case("x-yoi-runtime-id")
|| name.eq_ignore_ascii_case("x-yoi-worker-id")
})
}
fn browser_actor(context: &ServerRequestContext) -> DriveResult<&RequestActor> {
if worker_request(context) {
return Err(fail(DriveApiErrorCode::Denied));
}
context
.actor
.as_ref()
.ok_or_else(|| fail(DriveApiErrorCode::Denied))
}
pub(super) async fn authorized_drive(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace_id: &str,
write: bool,
) -> DriveResult<workspace_drive::Drive> {
validate_workspace_scope(api, workspace_id).map_err(pre_error)?;
let workspace = workspace_id.to_owned();
let drive = if worker_request(context) {
// Never fall back to an actor when a Worker/Runtime source is present,
// incomplete, or inconsistent with its transport identity.
let headers = current_worker_contract_headers(context)
.map_err(|_| fail(DriveApiErrorCode::Denied))?;
let worker = current_worker_identity(api, workspace_id, &headers)
.map_err(|_| fail(DriveApiErrorCode::Denied))?;
if context.worker_source.as_ref().is_some_and(|source| {
source.runtime_id != worker.runtime_id || source.worker_id != worker.worker_id
}) {
return Err(fail(DriveApiErrorCode::Denied));
}
let drive = api.drive().await.map_err(pre_error)?;
drive.with_authorizer(move |conn, write| {
if !crate::store::drive_worker_is_live(conn, true, &workspace, &worker)? {
return Err(workspace_drive::Error::Denied);
}
let granted: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM yoi_workspace_authority.workspace_drive_grants
WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND revoked=0 AND (?4=0 OR access='read_write'))",
params![workspace,worker.runtime_id,worker.worker_id,write],|row|row.get(0))?;
if !granted { return Err(workspace_drive::Error::Denied); }
Ok(())
})
} else {
let actor = browser_actor(context)?;
let account = actor.account_id.clone();
let user = actor.user_id.clone();
let drive = api.drive().await.map_err(pre_error)?;
// Ordinary Workspace use is member-facing: the current Server policy
// admits persisted authenticated users, not only the Workspace owner.
drive.with_authorizer(move |conn, _write| {
let member: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM yoi_workspace_authority.workspaces ws
CROSS JOIN yoi_workspace_authority.users u
WHERE ws.workspace_id=?1 AND ws.state='active' AND u.account_id=?2 AND u.user_id=?3)",
params![workspace,account,user],|row|row.get(0))?;
if !member { return Err(workspace_drive::Error::Denied); }
Ok(())
})
};
// Reject missing/read-only grants before upload body reception. This check
// is admission only: every later operation rechecks under its own lock.
tokio::task::spawn_blocking(move || {
drive.check_authorized(write)?;
Ok::<_, workspace_drive::Error>(drive)
})
.await
.map_err(|_| fail(DriveApiErrorCode::StorageUnavailable))?
.map_err(|error| {
fail(match error {
workspace_drive::Error::Denied => DriveApiErrorCode::Denied,
workspace_drive::Error::Fenced | workspace_drive::Error::Conflict => {
DriveApiErrorCode::Conflict
}
workspace_drive::Error::NotFound => DriveApiErrorCode::NotFound,
workspace_drive::Error::Invalid(_) => DriveApiErrorCode::Invalid,
_ => DriveApiErrorCode::StorageUnavailable,
})
})
}
pub(super) async fn create_grant(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace_id: &str,
request: DriveGrantCreateRequest,
) -> DriveResult<DriveGrantResponse> {
let actor = browser_actor(context)?;
require_workspace_owner(api, workspace_id, actor, "Drive grants")
.await
.map_err(pre_error)?;
for identity in [&request.runtime_id, &request.worker_id] {
if identity.is_empty() || identity.len() > 256 || identity.chars().any(char::is_control) {
return Err(fail(DriveApiErrorCode::Invalid));
}
}
let worker = RuntimeWorkerRef::new(&request.runtime_id, &request.worker_id);
let summary = api
.runtime
.worker(&worker)
.map_err(|_| fail(DriveApiErrorCode::NotFound))?;
if summary.workspace.workspace_id.as_deref() != Some(workspace_id) {
return Err(fail(DriveApiErrorCode::Denied));
}
let grant = DriveGrantResponse {
grant_id: String::new(),
workspace_id: workspace_id.into(),
runtime_id: request.runtime_id,
worker_id: request.worker_id,
access: request.access,
revoked: false,
created_by: actor.account_id.clone(),
created_at: String::new(),
revoked_by: None,
revoked_at: None,
};
let store = api.store.clone();
tokio::task::spawn_blocking(move || store.create_workspace_drive_grant(&grant))
.await
.map_err(|_| fail(DriveApiErrorCode::OutcomeUnknown))?
.map_err(mutation_error)
}
pub(super) async fn revoke_grant(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace_id: &str,
grant_id: &str,
) -> DriveResult<DriveGrantResponse> {
let actor = browser_actor(context)?;
require_workspace_owner(api, workspace_id, actor, "Drive grants")
.await
.map_err(pre_error)?;
validate_grant_id(grant_id)?;
let (workspace, grant, actor) = (
workspace_id.to_owned(),
grant_id.to_owned(),
actor.account_id.clone(),
);
let store = api.store.clone();
tokio::task::spawn_blocking(move || {
store.revoke_workspace_drive_grant(&workspace, &grant, &actor)
})
.await
.map_err(|_| fail(DriveApiErrorCode::OutcomeUnknown))?
.map_err(mutation_error)?
.ok_or_else(|| fail(DriveApiErrorCode::NotFound))
}
fn validate_grant_id(id: &str) -> DriveResult<()> {
if !id
.parse::<i64>()
.is_ok_and(|n| n > 0 && n.to_string() == id)
{
return Err(fail(DriveApiErrorCode::Invalid));
}
Ok(())
}
pub(super) async fn list_grants(
api: &WorkspaceApi,
context: &ServerRequestContext,
workspace_id: &str,
query: DriveGrantListQuery,
) -> DriveResult<DriveGrantListResponse> {
let actor = browser_actor(context)?;
require_workspace_owner(api, workspace_id, actor, "Drive grants")
.await
.map_err(pre_error)?;
let limit = query.limit.unwrap_or(100) as usize;
if limit == 0 || limit > 200 {
return Err(fail(DriveApiErrorCode::Invalid));
}
if let Some(after) = &query.after {
validate_grant_id(after)?;
}
let workspace = workspace_id.to_owned();
let store = api.store.clone();
let mut grants = tokio::task::spawn_blocking(move || {
store.list_workspace_drive_grants(&workspace, limit + 1, query.after.as_deref())
})
.await
.map_err(|_| fail(DriveApiErrorCode::StorageUnavailable))?
.map_err(pre_error)?;
let more = grants.len() > limit;
grants.truncate(limit);
let next_after = if more {
grants.last().map(|g| g.grant_id.clone())
} else {
None
};
Ok(DriveGrantListResponse { grants, next_after })
}
crates/workspace-server/src/store.rs
+374 -5
View File
@@ -28,7 +28,8 @@ use crate::workspace_deletion::WorkspaceDeletionStore;
use crate::{Error, Result};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 83;
const LATEST_SCHEMA_VERSION: i64 = 84;
const WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME: &str = "durable Workspace Drive grants";
const WORKSPACE_CONFIG_GRANTS_MIGRATION_NAME: &str = "Workspace config grants and logical Workdirs";
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings";
@@ -293,6 +294,11 @@ const MIGRATIONS: &[Migration] = &[
name: "Generic Ticket Worker roles and durable claims",
apply: migrate_ticket_worker_v82_to_v83,
},
Migration {
version: 84,
name: WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME,
apply: migrate_workspace_drive_grants_v83_to_v84,
},
];
#[derive(Clone, Copy)]
@@ -2005,6 +2011,28 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
) -> Result<WorkdirCatalogPage>;
fn delete_workdir_registry(&self, workspace_id: &str, workdir_id: &str) -> Result<bool>;
fn create_workspace_drive_grant(
&self,
grant: &server_api::DriveGrantResponse,
) -> Result<server_api::DriveGrantResponse>;
fn get_workspace_drive_grant(
&self,
workspace_id: &str,
grant_id: &str,
) -> Result<Option<server_api::DriveGrantResponse>>;
fn list_workspace_drive_grants(
&self,
workspace_id: &str,
limit: usize,
after: Option<&str>,
) -> Result<Vec<server_api::DriveGrantResponse>>;
fn revoke_workspace_drive_grant(
&self,
workspace_id: &str,
grant_id: &str,
actor: &str,
) -> Result<Option<server_api::DriveGrantResponse>>;
fn create_workspace_config_grant(
&self,
grant: &server_api::WorkspaceConfigGrantResponse,
@@ -9505,6 +9533,81 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn create_workspace_drive_grant(
&self,
grant: &server_api::DriveGrantResponse,
) -> Result<server_api::DriveGrantResponse> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
require_drive_grant_owner(&tx, &grant.workspace_id, &grant.created_by)?;
let worker = RuntimeWorkerRef::new(&grant.runtime_id, &grant.worker_id);
if !drive_worker_is_live(&tx, false, &grant.workspace_id, &worker)? {
return Err(Error::WorkspacePermissionDenied("Worker is not live in this Workspace".into()));
}
let existing = tx.query_row(&format!("{DRIVE_GRANT_SELECT} WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND revoked=0"),
params![grant.workspace_id, grant.runtime_id, grant.worker_id], read_drive_grant).optional()?;
if let Some(existing) = existing {
if existing.access != grant.access {
return Err(Error::WorkspaceConfigConflict("Drive grant exists; revoke it first".into()));
}
return Ok(existing);
}
let access = match grant.access { server_api::DriveAccess::ReadOnly => "read_only", server_api::DriveAccess::ReadWrite => "read_write" };
tx.execute("INSERT INTO workspace_drive_grants(workspace_id,runtime_id,worker_id,access,revoked,created_by,created_at) VALUES(?1,?2,?3,?4,0,?5,?6)",
params![grant.workspace_id,grant.runtime_id,grant.worker_id,access,grant.created_by,chrono::Utc::now().to_rfc3339()])?;
let stored = tx.query_row(&format!("{DRIVE_GRANT_SELECT} WHERE workspace_id=?1 AND grant_id=?2"),params![grant.workspace_id,tx.last_insert_rowid()],read_drive_grant)?;
tx.commit()?;
Ok(stored)
})
}
fn get_workspace_drive_grant(
&self,
workspace_id: &str,
grant_id: &str,
) -> Result<Option<server_api::DriveGrantResponse>> {
self.with_conn(|conn| {
Ok(conn
.query_row(
&format!("{DRIVE_GRANT_SELECT} WHERE workspace_id=?1 AND grant_id=?2"),
params![workspace_id, grant_id],
read_drive_grant,
)
.optional()?)
})
}
fn list_workspace_drive_grants(
&self,
workspace_id: &str,
limit: usize,
after: Option<&str>,
) -> Result<Vec<server_api::DriveGrantResponse>> {
if limit == 0 || limit > 201 {
return Err(Error::WorkspaceConfigConflict(
"Invalid Drive grant page limit".into(),
));
}
self.with_conn(|conn| {
let mut stmt = conn.prepare(&format!("{DRIVE_GRANT_SELECT} WHERE workspace_id=?1 AND grant_id>?2 ORDER BY grant_id LIMIT ?3"))?;
Ok(stmt.query_map(params![workspace_id,after.unwrap_or("0").parse::<i64>().map_err(|_| Error::WorkspaceConfigConflict("Invalid Drive grant cursor".into()))?,limit],read_drive_grant)?.collect::<rusqlite::Result<Vec<_>>>()?)
})
}
fn revoke_workspace_drive_grant(
&self,
workspace_id: &str,
grant_id: &str,
actor: &str,
) -> Result<Option<server_api::DriveGrantResponse>> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
require_drive_grant_owner(&tx, workspace_id, actor)?;
tx.execute("UPDATE workspace_drive_grants SET revoked=1,revoked_by=?3,revoked_at=?4 WHERE workspace_id=?1 AND grant_id=?2 AND revoked=0",
params![workspace_id,grant_id,actor,chrono::Utc::now().to_rfc3339()])?;
let stored = tx.query_row(&format!("{DRIVE_GRANT_SELECT} WHERE workspace_id=?1 AND grant_id=?2"),params![workspace_id,grant_id],read_drive_grant).optional()?;
tx.commit()?;
Ok(stored)
})
}
fn create_workspace_config_grant(
&self,
grant: &server_api::WorkspaceConfigGrantResponse,
@@ -11906,6 +12009,12 @@ pub(crate) fn commit_worker_catalog_removal(
workspace_id: &str,
worker: &RuntimeWorkerRef,
) -> Result<WorkerRegistryProjectionCommit> {
conn.execute(
"UPDATE workspace_drive_grants SET revoked=1, revoked_by='server:worker-removal',
revoked_at=strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3 AND revoked=0",
params![workspace_id, worker.runtime_id, worker.worker_id],
)?;
let existing = conn
.query_row(
"SELECT 1 FROM worker_registry_projection_removals WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3",
@@ -11925,6 +12034,56 @@ pub(crate) fn commit_worker_catalog_removal(
})
}
const DRIVE_GRANT_SELECT: &str = "SELECT grant_id,workspace_id,runtime_id,worker_id,access,revoked,created_by,created_at,revoked_by,revoked_at FROM workspace_drive_grants";
fn read_drive_grant(row: &rusqlite::Row<'_>) -> rusqlite::Result<server_api::DriveGrantResponse> {
Ok(server_api::DriveGrantResponse {
grant_id: row.get::<_, i64>(0)?.to_string(),
workspace_id: row.get(1)?,
runtime_id: row.get(2)?,
worker_id: row.get(3)?,
access: match row.get::<_, String>(4)?.as_str() {
"read_only" => server_api::DriveAccess::ReadOnly,
"read_write" => server_api::DriveAccess::ReadWrite,
_ => return Err(rusqlite::Error::InvalidQuery),
},
revoked: row.get(5)?,
created_by: row.get(6)?,
created_at: row.get(7)?,
revoked_by: row.get(8)?,
revoked_at: row.get(9)?,
})
}
fn require_drive_grant_owner(conn: &Connection, workspace: &str, actor: &str) -> Result<()> {
let allowed: bool = conn.query_row("SELECT EXISTS(SELECT 1 FROM workspaces WHERE workspace_id=?1 AND owner_account_id=?2 AND state='active')",params![workspace,actor],|r|r.get(0))?;
if !allowed {
return Err(Error::WorkspacePermissionDenied(
"Drive grants require the active Workspace owner".into(),
));
}
Ok(())
}
/// The prefix is selected only by trusted Host code, never request data. Uses
/// the durable Runtime/Worker identity and removal authority, not observations.
pub(crate) fn drive_worker_is_live(
conn: &Connection,
attached: bool,
workspace: &str,
worker: &RuntimeWorkerRef,
) -> rusqlite::Result<bool> {
let a = if attached {
"yoi_workspace_authority."
} else {
""
};
conn.query_row(&format!("SELECT EXISTS(SELECT 1 FROM {a}worker_registry wr JOIN {a}workspaces ws ON ws.workspace_id=wr.workspace_id
WHERE wr.workspace_id=?1 AND wr.runtime_id=?2 AND wr.worker_id=?3 AND ws.state='active'
AND (wr.runtime_id=?4 OR EXISTS(SELECT 1 FROM {a}workspace_runtime_bindings rb WHERE rb.workspace_id=?1 AND rb.runtime_id=?2 AND rb.state='verified' AND rb.revoked_at IS NULL))
AND NOT EXISTS(SELECT 1 FROM {a}worker_registry_projection_removals rm WHERE rm.workspace_id=?1 AND rm.runtime_id=?2 AND rm.worker_id=?3)
AND NOT EXISTS(SELECT 1 FROM {a}worker_tombstones wt WHERE wt.workspace_id=?1 AND wt.runtime_id=?2 AND wt.worker_id=?3)
AND NOT EXISTS(SELECT 1 FROM {a}worker_removal_operations ro WHERE ro.workspace_id=?1 AND ro.runtime_id=?2 AND ro.worker_id=?3 AND ro.state IN ('executing','failed','succeeded')))") ,
params![workspace,worker.runtime_id,worker.worker_id,crate::hosts::EMBEDDED_RUNTIME_ID],|r|r.get(0))
}
fn read_worker_control_grant_record(
row: &rusqlite::Row<'_>,
) -> rusqlite::Result<WorkerControlGrantRecord> {
@@ -12595,6 +12754,22 @@ fn read_workdir_registry_record(
})
}
fn migrate_workspace_drive_grants_v83_to_v84(conn: &Connection) -> Result<()> {
if current_schema_version(conn)? != 83 {
return Err(Error::Store(
"Drive grants migration requires schema 83".into(),
));
}
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
tx.execute_batch(include_str!("workspace_drive_grants.sql"))?;
tx.execute(
"INSERT INTO __yoi_schema_migrations(version,name) VALUES(?1,?2)",
params![84_i64, WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME],
)?;
tx.commit()?;
Ok(())
}
fn migrate_worker_restore_intents_v79_to_v80(conn: &Connection) -> Result<()> {
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
tx.execute_batch("CREATE TABLE worker_restore_intents (
@@ -16816,6 +16991,193 @@ fn migrate_workdir_credential_candidate_snapshots_v58_to_v59(conn: &Connection)
#[cfg(test)]
mod tests {
include!("store_workspace_config_tests.rs");
fn drive_grant_fixture(store: &SqliteWorkspaceStore) -> server_api::DriveGrantResponse {
store.with_conn(|c| {
c.execute_batch("INSERT INTO accounts(account_id,kind,handle,display_name,created_at,updated_at) VALUES('drive-owner','user','drive-owner','Owner','1','1');
INSERT INTO workspaces(workspace_id,owner_account_id,display_name,state,created_at,updated_at) VALUES('drive-ws','drive-owner','Workspace','active','1','1'),('drive-other','drive-owner','Other','active','1','1');")?;
for workspace in ["drive-ws", "drive-other"] {
c.execute("INSERT INTO workspace_signing_identities(workspace_id,key_id,algorithm,private_material_ref,revision,state,created_at,updated_at) VALUES(?1,?1,'ed25519',?1,1,'pending_provisioning','1','1')", [workspace])?;
c.execute("INSERT INTO worker_registry(workspace_id,runtime_id,worker_id,display_name,retention_state,created_at,updated_at) VALUES(?1,?2,'worker','Worker','normal','1','1')", params![workspace, crate::hosts::EMBEDDED_RUNTIME_ID])?;
}
Ok(())
}).unwrap();
server_api::DriveGrantResponse {
grant_id: String::new(),
workspace_id: "drive-ws".into(),
runtime_id: crate::hosts::EMBEDDED_RUNTIME_ID.into(),
worker_id: "worker".into(),
access: server_api::DriveAccess::ReadOnly,
revoked: false,
created_by: "drive-owner".into(),
created_at: String::new(),
revoked_by: None,
revoked_at: None,
}
}
#[test]
fn drive_grants_migrate_and_preserve_audit_through_revoke_worker_delete_and_restart() {
let t = tempfile::tempdir().unwrap();
let path = t.path().join("server.db");
let store = SqliteWorkspaceStore::open(&path).unwrap();
let mut proposed = drive_grant_fixture(&store);
store.with_conn(|c| {
c.execute_batch("DROP TRIGGER workspace_drive_grants_worker_deleted; DROP TABLE workspace_drive_grants; UPDATE __yoi_schema_migrations SET version=83 WHERE version=84;")?;
assert_eq!(current_schema_version(c)?, 83);
Ok(())
}).unwrap();
drop(store);
let store = SqliteWorkspaceStore::open(&path).unwrap();
store
.with_conn(|c| {
assert_eq!(current_schema_version(c)?, 84);
Ok(())
})
.unwrap();
let first = store.create_workspace_drive_grant(&proposed).unwrap();
assert_eq!(first.grant_id, "1");
assert_eq!(first.created_by, "drive-owner");
assert!(!first.created_at.is_empty());
assert_eq!(
store.create_workspace_drive_grant(&proposed).unwrap(),
first
);
proposed.access = server_api::DriveAccess::ReadWrite;
assert!(store.create_workspace_drive_grant(&proposed).is_err());
proposed.workspace_id = "drive-other".into();
let other = store.create_workspace_drive_grant(&proposed).unwrap();
assert!(
store
.get_workspace_drive_grant("drive-other", &first.grant_id)
.unwrap()
.is_none()
);
assert!(
store
.revoke_workspace_drive_grant("drive-other", &first.grant_id, "drive-owner")
.unwrap()
.is_none()
);
assert!(
store
.revoke_workspace_drive_grant("drive-ws", &first.grant_id, "stranger")
.is_err()
);
let revoked = store
.revoke_workspace_drive_grant("drive-ws", &first.grant_id, "drive-owner")
.unwrap()
.unwrap();
assert!(revoked.revoked);
assert_eq!(revoked.revoked_by.as_deref(), Some("drive-owner"));
assert!(revoked.revoked_at.is_some());
assert_eq!(
store
.revoke_workspace_drive_grant("drive-ws", &first.grant_id, "drive-owner")
.unwrap()
.unwrap(),
revoked
);
proposed.workspace_id = "drive-ws".into();
let second = store.create_workspace_drive_grant(&proposed).unwrap();
assert_ne!(second.grant_id, first.grant_id);
assert_eq!(
store
.list_workspace_drive_grants("drive-ws", 1, None)
.unwrap(),
vec![revoked.clone()]
);
assert_eq!(
store
.list_workspace_drive_grants("drive-ws", 1, Some(&first.grant_id))
.unwrap(),
vec![second.clone()]
);
store
.delete_worker_registry(
"drive-ws",
&RuntimeWorkerRef::new(crate::hosts::EMBEDDED_RUNTIME_ID, "worker"),
)
.unwrap();
let retired = store
.get_workspace_drive_grant("drive-ws", &second.grant_id)
.unwrap()
.unwrap();
assert!(retired.revoked);
assert_eq!(retired.revoked_by.as_deref(), Some("server:worker-removal"));
assert_eq!(retired.created_at, second.created_at);
assert!(store.create_workspace_drive_grant(&proposed).is_err());
assert_eq!(
store
.get_workspace_drive_grant("drive-other", &other.grant_id)
.unwrap(),
Some(other)
);
drop(store);
let store = SqliteWorkspaceStore::open(&path).unwrap();
assert_eq!(
store
.get_workspace_drive_grant("drive-ws", &first.grant_id)
.unwrap(),
Some(revoked)
);
assert_eq!(
store
.get_workspace_drive_grant("drive-ws", &second.grant_id)
.unwrap(),
Some(retired)
);
store
.with_conn(|c| {
c.execute("DELETE FROM workspaces WHERE workspace_id='drive-ws'", [])?;
Ok(())
})
.unwrap();
assert!(
store
.list_workspace_drive_grants("drive-ws", 10, None)
.unwrap()
.is_empty()
);
assert_eq!(
store
.list_workspace_drive_grants("drive-other", 10, None)
.unwrap()
.len(),
1
);
}
#[test]
fn drive_grants_baseline_constraints_and_worker_lifetime_fail_closed() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
let proposed = drive_grant_fixture(&store);
let first = store.create_workspace_drive_grant(&proposed).unwrap();
store.with_conn(|c| {
assert_eq!(current_schema_version(c)?, 84);
assert!(c.execute("INSERT INTO workspace_drive_grants(workspace_id,runtime_id,worker_id,access,created_by,created_at) VALUES('drive-ws',?1,'worker','read_write','drive-owner','now')", [crate::hosts::EMBEDDED_RUNTIME_ID]).is_err());
assert!(c.execute("INSERT INTO workspace_drive_grants(workspace_id,runtime_id,worker_id,access,created_by,created_at) VALUES('drive-ws','r','w','command','drive-owner','now')", []).is_err());
assert!(c.execute("UPDATE workspace_drive_grants SET revoked=1 WHERE grant_id=?1", [&first.grant_id]).is_err());
c.execute("INSERT INTO worker_registry_projection_removals(workspace_id,runtime_id,worker_id) VALUES('drive-ws',?1,'worker')", [crate::hosts::EMBEDDED_RUNTIME_ID])?;
assert!(!drive_worker_is_live(c, false, "drive-ws", &RuntimeWorkerRef::new(crate::hosts::EMBEDDED_RUNTIME_ID, "worker"))?);
Ok(())
}).unwrap();
assert!(store.create_workspace_drive_grant(&proposed).is_err());
store
.with_conn(|c| {
c.execute(
"DELETE FROM worker_registry WHERE workspace_id='drive-ws'",
[],
)?;
Ok(())
})
.unwrap();
assert!(
store
.get_workspace_drive_grant("drive-ws", &first.grant_id)
.unwrap()
.unwrap()
.revoked
);
}
use super::*;
#[test]
@@ -18340,6 +18702,8 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
}
fn downgrade_ticket_worker_schema(conn: &Connection, restore_worker_fk: bool) {
// Older-generation fixtures must not retain the new Drive authority.
conn.execute_batch("DROP TRIGGER IF EXISTS workspace_drive_grants_worker_deleted; DROP TABLE IF EXISTS workspace_drive_grants;").unwrap();
// Restore the actual old role and FK contracts, not just the old
// unique index, so migration tests exercise real coder rows.
for table in [
@@ -18428,7 +18792,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.iter()
.map(|s| s.version)
.collect::<Vec<_>>(),
vec![83]
vec![83, 84]
);
let store = SqliteWorkspaceStore::open(&path).unwrap();
let assignment = store
@@ -18454,7 +18818,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
);
}
store.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?,83);
assert_eq!(current_schema_version(conn)?,LATEST_SCHEMA_VERSION);
assert_eq!(conn.query_row("SELECT role FROM ticket_worker_assignments WHERE assignment_id='assignment'", [], |r| r.get::<_,String>(0))?,"worker");
assert_eq!(conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |r| r.get::<_,i64>(0))?,0);
Ok(())
@@ -18739,6 +19103,10 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
version: 83,
name: "Generic Ticket Worker roles and durable claims".to_string()
},
WorkspaceSchemaMigrationStep {
version: 84,
name: WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME.to_string()
},
]
);
@@ -18832,6 +19200,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
(81, "Backend Job immutable resource serialization".to_string()),
(82, "Ticket responsibility and unfinished work separation".to_string()),
(83, "Generic Ticket Worker roles and durable claims".to_string()),
(84, WORKSPACE_DRIVE_GRANTS_MIGRATION_NAME.to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -19183,7 +19552,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
.collect::<Vec<_>>(),
vec![
52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72,
73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83
73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84
]
);
SqliteWorkspaceStore::migrate_database(&path).unwrap();
@@ -19192,7 +19561,7 @@ CREATE UNIQUE INDEX ticket_current_worker_role_idx
current_schema_version(&conn).unwrap(),
LATEST_SCHEMA_VERSION
);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 34);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 35);
assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap());
assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap());
assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap());
crates/workspace-server/src/workspace_deletion.rs
+1
View File
@@ -87,6 +87,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
"worker_workdir_links",
"workspace_config_entries",
"workspace_config_grants",
"workspace_drive_grants",
"workspace_config_tree_revisions",
"workspace_config_trees",
"workspace_create_operations",
crates/workspace-server/src/workspace_drive_grants.sql
+24
View File
@@ -0,0 +1,24 @@
CREATE TABLE workspace_drive_grants (
workspace_id TEXT NOT NULL REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
grant_id INTEGER PRIMARY KEY AUTOINCREMENT,
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
access TEXT NOT NULL CHECK(access IN ('read_only', 'read_write')),
revoked INTEGER NOT NULL DEFAULT 0 CHECK(revoked IN (0, 1)),
created_by TEXT NOT NULL,
created_at TEXT NOT NULL,
revoked_by TEXT,
revoked_at TEXT,
CHECK((revoked=0 AND revoked_by IS NULL AND revoked_at IS NULL)
OR (revoked=1 AND revoked_by IS NOT NULL AND revoked_at IS NOT NULL))
);
CREATE UNIQUE INDEX workspace_drive_grants_active_worker
ON workspace_drive_grants(workspace_id, runtime_id, worker_id) WHERE revoked=0;
-- Retain grant audit across Worker deletion, but never carry its authority into
-- a later lifetime reusing the same Runtime/Worker identity.
CREATE TRIGGER workspace_drive_grants_worker_deleted BEFORE DELETE ON worker_registry BEGIN
UPDATE workspace_drive_grants SET revoked=1, revoked_by='server:worker-removal',
revoked_at=strftime('%Y-%m-%dT%H:%M:%fZ', 'now')
WHERE workspace_id=OLD.workspace_id AND runtime_id=OLD.runtime_id
AND worker_id=OLD.worker_id AND revoked=0;
END;