Merge branch 'develop' into hare/develop

This commit is contained in:
2026-08-27 09:21:58 +09:00
21 changed files with 5243 additions and 407 deletions
Generated
+2
View File
@@ -6652,6 +6652,7 @@ dependencies = [
"workdir", "workdir",
"worker", "worker",
"workspace-api", "workspace-api",
"zeroize",
] ]
[[package]] [[package]]
@@ -6798,6 +6799,7 @@ dependencies = [
"worker", "worker",
"worker-runtime", "worker-runtime",
"workspace-api", "workspace-api",
"zeroize",
] ]
[[package]] [[package]]
+1
View File
@@ -125,4 +125,5 @@ toml = "1.1"
tracing = "0.1" tracing = "0.1"
url = "2.5" url = "2.5"
uuid = "1.23" uuid = "1.23"
zeroize = "1"
webauthn-rs = { version = "0.5.2", features = ["danger-allow-state-serialisation", "danger-credential-internals"] } webauthn-rs = { version = "0.5.2", features = ["danger-allow-state-serialisation", "danger-credential-internals"] }
+3 -1
View File
@@ -28,7 +28,9 @@ pub use fs_operation::{
GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult, GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult,
ReadRequest, ReadResult, StatRequest, StatResult, WriteRequest, WriteResult, ReadRequest, ReadResult, StatRequest, StatResult, WriteRequest, WriteResult,
}; };
pub use local::{LocalWorkdirSession, SymlinkInfo, direct_symlink, first_symlink}; pub use local::{
LocalWorkdirSession, SymlinkInfo, WorkdirSessionResource, direct_symlink, first_symlink,
};
pub use operation::*; pub use operation::*;
/// Persistent, opaque identity of one materialized Workdir. /// Persistent, opaque identity of one materialized Workdir.
+69 -2
View File
@@ -8,7 +8,8 @@
//! `LocalWorkdirSession` is cheap to clone (`Arc` inside). Tool-specific session //! `LocalWorkdirSession` is cheap to clone (`Arc` inside). Tool-specific session
//! state, such as read-before-edit tracking, remains owned by the tool layer. //! state, such as read-before-edit tracking, remains owned by the tool layer.
use std::collections::HashMap; use std::collections::{BTreeMap, HashMap};
use std::fmt::Debug;
#[cfg(test)] #[cfg(test)]
use std::io::Write as _; use std::io::Write as _;
use std::io::{Read as _, Seek as _, SeekFrom}; use std::io::{Read as _, Seek as _, SeekFrom};
@@ -228,6 +229,8 @@ struct LocalWorkdirSessionInner {
next_command_id: AtomicU64, next_command_id: AtomicU64,
commands: Mutex<HashMap<String, LocalCommand>>, commands: Mutex<HashMap<String, LocalCommand>>,
command_telemetry: CommandTelemetry, command_telemetry: CommandTelemetry,
command_environment: BTreeMap<String, String>,
resources: StdMutex<Vec<Arc<dyn WorkdirSessionResource>>>,
} }
impl Drop for LocalWorkdirSessionInner { impl Drop for LocalWorkdirSessionInner {
@@ -242,6 +245,9 @@ impl Drop for LocalWorkdirSessionInner {
} }
} }
pub trait WorkdirSessionResource: Debug + Send + Sync {}
impl<T> WorkdirSessionResource for T where T: Debug + Send + Sync {}
/// Scope-aware filesystem handle. Clone-cheap (`Arc` inside). /// Scope-aware filesystem handle. Clone-cheap (`Arc` inside).
/// ///
/// The wrapped [`SharedScope`] is shared with every clone of this /// The wrapped [`SharedScope`] is shared with every clone of this
@@ -318,6 +324,26 @@ impl LocalWorkdirSession {
cwd: PathBuf, cwd: PathBuf,
scope: SharedScope, scope: SharedScope,
capabilities: WorkdirSessionCapabilities, capabilities: WorkdirSessionCapabilities,
) -> Self {
Self::materialized_bound_with_environment(
workdir,
root,
cwd,
scope,
capabilities,
BTreeMap::new(),
Vec::new(),
)
}
pub fn materialized_bound_with_environment(
workdir: Workdir,
root: PathBuf,
cwd: PathBuf,
scope: SharedScope,
capabilities: WorkdirSessionCapabilities,
command_environment: BTreeMap<String, String>,
resources: Vec<Arc<dyn WorkdirSessionResource>>,
) -> Self { ) -> Self {
Self { Self {
inner: Arc::new(LocalWorkdirSessionInner { inner: Arc::new(LocalWorkdirSessionInner {
@@ -331,6 +357,8 @@ impl LocalWorkdirSession {
next_command_id: AtomicU64::new(1), next_command_id: AtomicU64::new(1),
commands: Mutex::new(HashMap::new()), commands: Mutex::new(HashMap::new()),
command_telemetry: CommandTelemetry::new(), command_telemetry: CommandTelemetry::new(),
command_environment,
resources: StdMutex::new(resources),
}), }),
} }
} }
@@ -669,9 +697,18 @@ impl WorkdirSession for LocalWorkdirSession {
let (completion_tx, completion) = watch::channel(false); let (completion_tx, completion) = watch::channel(false);
let command_id = handle.0.clone(); let command_id = handle.0.clone();
let telemetry = self.inner.command_telemetry.clone(); let telemetry = self.inner.command_telemetry.clone();
let command_environment = self.inner.command_environment.clone();
let (cancel, cancel_rx) = watch::channel(false); let (cancel, cancel_rx) = watch::channel(false);
let task = tokio::spawn(async move { let task = tokio::spawn(async move {
let output = run_command(cwd, request, command_id, telemetry, cancel_rx).await; let output = run_command(
cwd,
request,
command_id,
telemetry,
command_environment,
cancel_rx,
)
.await;
let _ = completion_tx.send(true); let _ = completion_tx.send(true);
output output
}); });
@@ -840,6 +877,9 @@ impl WorkdirSession for LocalWorkdirSession {
LocalCommand::Completed(_) => {} LocalCommand::Completed(_) => {}
} }
} }
if let Ok(mut resources) = self.inner.resources.lock() {
resources.clear();
}
Ok(()) Ok(())
} }
} }
@@ -909,6 +949,7 @@ async fn run_command(
request: CommandRequest, request: CommandRequest,
command_id: String, command_id: String,
telemetry: CommandTelemetry, telemetry: CommandTelemetry,
command_environment: BTreeMap<String, String>,
mut cancel: watch::Receiver<bool>, mut cancel: watch::Receiver<bool>,
) -> Result<CommandOutput, WorkdirError> { ) -> Result<CommandOutput, WorkdirError> {
let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?; let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
@@ -925,6 +966,7 @@ async fn run_command(
.arg("-c") .arg("-c")
.arg(&request.command) .arg(&request.command)
.current_dir(&cwd) .current_dir(&cwd)
.envs(command_environment)
.stdin(Stdio::null()) .stdin(Stdio::null())
.stdout(Stdio::from(stdout_file)) .stdout(Stdio::from(stdout_file))
.stderr(Stdio::from(stderr_file)) .stderr(Stdio::from(stderr_file))
@@ -2319,6 +2361,31 @@ mod tests {
assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None))); assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None)));
} }
#[tokio::test]
async fn closing_session_releases_runtime_resources() {
#[derive(Debug)]
struct Resource(Arc<AtomicBool>);
impl Drop for Resource {
fn drop(&mut self) {
self.0.store(true, Ordering::Release);
}
}
let dir = TempDir::new().unwrap();
let released = Arc::new(AtomicBool::new(false));
let session = LocalWorkdirSession::materialized_bound_with_environment(
Workdir::new("resource-session"),
dir.path().to_path_buf(),
dir.path().to_path_buf(),
SharedScope::new(Scope::writable(dir.path()).unwrap()),
WorkdirSessionCapabilities::ALL,
BTreeMap::from([("SSH_AUTH_SOCK".to_string(), "test-socket".to_string())]),
vec![Arc::new(Resource(released.clone()))],
);
WorkdirSession::close(&session).await.unwrap();
assert!(released.load(Ordering::Acquire));
}
#[tokio::test] #[tokio::test]
async fn provider_cancels_active_command() { async fn provider_cancels_active_command() {
let dir = TempDir::new().unwrap(); let dir = TempDir::new().unwrap();
+23
View File
@@ -30,6 +30,8 @@ impl RuntimeWorkerRef {
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum MaterializerKind { pub enum MaterializerKind {
#[default] #[default]
RuntimeGitCache,
/// Legacy persisted value from the pre-cache local `git worktree` materializer.
LocalGitWorktree, LocalGitWorktree,
} }
@@ -109,6 +111,8 @@ pub struct WorkingDirectoryProvenance {
pub creation_selector: Option<String>, pub creation_selector: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub creation_ref: Option<String>, pub creation_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub creation_tree: Option<String>,
pub materializer_kind: MaterializerKind, pub materializer_kind: MaterializerKind,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub cleanup_target: Option<WorkingDirectoryCleanupTarget>, pub cleanup_target: Option<WorkingDirectoryCleanupTarget>,
@@ -122,6 +126,10 @@ pub struct WorkingDirectoryCurrentObservation {
pub current_selector: Option<String>, pub current_selector: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub current_ref: Option<String>, pub current_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub current_tree: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub observed_at_epoch_seconds: Option<u64>,
pub status: WorkingDirectoryStatusKind, pub status: WorkingDirectoryStatusKind,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub cleanliness: Option<String>, pub cleanliness: Option<String>,
@@ -141,9 +149,15 @@ pub struct WorkingDirectorySummary {
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub creation_ref: Option<String>, pub creation_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub creation_tree: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub current_selector: Option<String>, pub current_selector: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub current_ref: Option<String>, pub current_ref: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub current_tree: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub observed_at_epoch_seconds: Option<u64>,
pub materializer_kind: MaterializerKind, pub materializer_kind: MaterializerKind,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub cleanup_target: Option<WorkingDirectoryCleanupTarget>, pub cleanup_target: Option<WorkingDirectoryCleanupTarget>,
@@ -166,6 +180,7 @@ impl WorkingDirectorySummary {
WorkingDirectoryProvenance { WorkingDirectoryProvenance {
creation_selector: self.creation_selector.clone(), creation_selector: self.creation_selector.clone(),
creation_ref: self.creation_ref.clone(), creation_ref: self.creation_ref.clone(),
creation_tree: self.creation_tree.clone(),
materializer_kind: self.materializer_kind.clone(), materializer_kind: self.materializer_kind.clone(),
cleanup_target: self.cleanup_target.clone(), cleanup_target: self.cleanup_target.clone(),
} }
@@ -175,6 +190,8 @@ impl WorkingDirectorySummary {
WorkingDirectoryCurrentObservation { WorkingDirectoryCurrentObservation {
current_selector: self.current_selector.clone(), current_selector: self.current_selector.clone(),
current_ref: self.current_ref.clone(), current_ref: self.current_ref.clone(),
current_tree: self.current_tree.clone(),
observed_at_epoch_seconds: self.observed_at_epoch_seconds,
status: self.status.clone(), status: self.status.clone(),
cleanliness: self.cleanliness.clone(), cleanliness: self.cleanliness.clone(),
primary_worker_id: self.primary_worker_id.clone(), primary_worker_id: self.primary_worker_id.clone(),
@@ -247,8 +264,11 @@ mod tests {
repository_id: "repo".to_string(), repository_id: "repo".to_string(),
creation_selector: Some("develop".to_string()), creation_selector: Some("develop".to_string()),
creation_ref: Some("abc123".to_string()), creation_ref: Some("abc123".to_string()),
creation_tree: Some("tree123".to_string()),
current_selector: Some("work/ticket".to_string()), current_selector: Some("work/ticket".to_string()),
current_ref: Some("def456".to_string()), current_ref: Some("def456".to_string()),
current_tree: Some("tree456".to_string()),
observed_at_epoch_seconds: Some(1_777_777_777),
materializer_kind: MaterializerKind::LocalGitWorktree, materializer_kind: MaterializerKind::LocalGitWorktree,
cleanup_target: Some(WorkingDirectoryCleanupTarget { cleanup_target: Some(WorkingDirectoryCleanupTarget {
kind: "git_worktree".to_string(), kind: "git_worktree".to_string(),
@@ -269,8 +289,11 @@ mod tests {
repository_id: "repo".to_string(), repository_id: "repo".to_string(),
creation_selector: None, creation_selector: None,
creation_ref: None, creation_ref: None,
creation_tree: None,
current_selector: None, current_selector: None,
current_ref: Some("987fed".to_string()), current_ref: Some("987fed".to_string()),
current_tree: None,
observed_at_epoch_seconds: None,
materializer_kind: MaterializerKind::LocalGitWorktree, materializer_kind: MaterializerKind::LocalGitWorktree,
cleanup_target: None, cleanup_target: None,
status: WorkingDirectoryStatusKind::Active, status: WorkingDirectoryStatusKind::Active,
+1
View File
@@ -43,6 +43,7 @@ tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }
toml.workspace = true toml.workspace = true
url.workspace = true url.workspace = true
uuid = { workspace = true, features = ["v7"] } uuid = { workspace = true, features = ["v7"] }
zeroize.workspace = true
tower = { workspace = true, features = ["util"], optional = true } tower = { workspace = true, features = ["util"], optional = true }
worker.workspace = true worker.workspace = true
workspace-api = { path = "../workspace-api" } workspace-api = { path = "../workspace-api" }
+71
View File
@@ -97,6 +97,74 @@ pub use workdir::workspace::{
WorkingDirectorySummary, WorkingDirectorySummary,
}; };
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SensitiveString(String);
impl SensitiveString {
pub fn new(value: impl Into<String>) -> Self {
Self(value.into())
}
pub fn expose(&self) -> &str {
&self.0
}
}
impl Drop for SensitiveString {
fn drop(&mut self) {
zeroize::Zeroize::zeroize(&mut self.0);
}
}
impl Default for SensitiveString {
fn default() -> Self {
Self(String::new())
}
}
impl std::fmt::Debug for SensitiveString {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("[REDACTED]")
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositorySshMaterializationAccess {
pub credential_id: String,
pub credential_revision: u64,
pub host_trust_id: String,
pub host_trust_revision: u64,
pub access: workspace_api::RepositoryAccessMode,
pub expires_at_epoch_seconds: u64,
pub repository_id: String,
pub repository_source_fingerprint: String,
pub repository_uri: String,
pub secret_resource: crate::resource::BackendResourceHandle,
#[serde(skip, default)]
pub private_key: SensitiveString,
#[serde(skip, default)]
pub known_hosts_entry: SensitiveString,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositoryMaterializationContext {
pub workspace_id: String,
pub runtime_id: String,
pub operation_id: String,
pub config_revision: u64,
pub config_projection_digest: String,
#[serde(default)]
pub cache_generation: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ssh: Option<RepositorySshMaterializationAccess>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryRepositoryAccessRequest {
pub working_directory_id: String,
pub materialization: RepositoryMaterializationContext,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct WorkingDirectoryRequest { pub struct WorkingDirectoryRequest {
pub repository: WorkingDirectoryRepository, pub repository: WorkingDirectoryRepository,
@@ -106,6 +174,9 @@ pub struct WorkingDirectoryRequest {
/// Backend can create canonical registry rows before materialization. /// Backend can create canonical registry rows before materialization.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub backend_workdir_id: Option<String>, pub backend_workdir_id: Option<String>,
/// Backend-authored, operation-scoped repository access and cache identity.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub materialization: Option<RepositoryMaterializationContext>,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+21 -1
View File
@@ -1,4 +1,6 @@
use crate::catalog::{WorkingDirectoryRequest, WorkingDirectoryStatus}; use crate::catalog::{
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
};
use crate::config_bundle::ConfigBundle; use crate::config_bundle::ConfigBundle;
use crate::error::RuntimeError; use crate::error::RuntimeError;
use crate::identity::WorkerRef; use crate::identity::WorkerRef;
@@ -319,6 +321,16 @@ pub trait WorkerExecutionBackend: Send + Sync + 'static {
)) ))
} }
fn authorize_working_directory_repository_access(
&self,
_request: &WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), WorkingDirectoryDiagnostic> {
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_repository_access_unsupported",
"Worker execution backend does not support Repository access authorization",
))
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> { fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
Vec::new() Vec::new()
} }
@@ -454,6 +466,14 @@ impl WorkerExecutionBackendRef {
self.backend.create_working_directory(request) self.backend.create_working_directory(request)
} }
pub(crate) fn authorize_working_directory_repository_access(
&self,
request: &WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), WorkingDirectoryDiagnostic> {
self.backend
.authorize_working_directory_repository_access(request)
}
pub(crate) fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> { pub(crate) fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.backend.list_working_directories() self.backend.list_working_directories()
} }
+53 -2
View File
@@ -12,7 +12,8 @@ use crate::auth::{
}; };
use crate::catalog::{ use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary, ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkspaceApiRef,
}; };
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
use crate::error::RuntimeError; use crate::error::RuntimeError;
@@ -203,6 +204,10 @@ fn runtime_http_router_with_optional_auth(
"/v1/working-directories", "/v1/working-directories",
get(list_working_directories).post(create_working_directory), get(list_working_directories).post(create_working_directory),
) )
.route(
"/v1/working-directories/repository-access",
post(authorize_working_directory_repository_access),
)
.route( .route(
"/v1/working-directories/{working_directory_id}/sessions", "/v1/working-directories/{working_directory_id}/sessions",
post(open_workdir_session), post(open_workdir_session),
@@ -335,6 +340,11 @@ pub struct RuntimeHttpWorkingDirectoriesResponse {
pub working_directories: Vec<WorkingDirectoryStatus>, pub working_directories: Vec<WorkingDirectoryStatus>,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpRepositoryAccessResponse {
pub authorized: bool,
}
/// Working directory response used by create/detail/delete endpoints. /// Working directory response used by create/detail/delete endpoints.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct RuntimeHttpWorkingDirectoryResponse { pub struct RuntimeHttpWorkingDirectoryResponse {
@@ -513,6 +523,29 @@ async fn list_workers(
Ok(Json(RuntimeHttpWorkersResponse { workers })) Ok(Json(RuntimeHttpWorkersResponse { workers }))
} }
async fn authorize_working_directory_repository_access(
State(state): State<RuntimeHttpState>,
Extension(auth): Extension<RuntimeAuthContext>,
body: Result<Json<WorkingDirectoryRepositoryAccessRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpRepositoryAccessResponse> {
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
if request.materialization.workspace_id != auth.workspace_id {
return Err(RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"working_directory_materialization_workspace_mismatch",
"Repository access authority does not match the authenticated Workspace",
));
}
state
.runtime
.authorize_working_directory_repository_access_from_resource(request)
.await
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpRepositoryAccessResponse {
authorized: true,
}))
}
async fn list_working_directories( async fn list_working_directories(
State(state): State<RuntimeHttpState>, State(state): State<RuntimeHttpState>,
) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> { ) -> RestResult<RuntimeHttpWorkingDirectoriesResponse> {
@@ -527,12 +560,23 @@ async fn list_working_directories(
async fn create_working_directory( async fn create_working_directory(
State(state): State<RuntimeHttpState>, State(state): State<RuntimeHttpState>,
Extension(auth): Extension<RuntimeAuthContext>,
body: Result<Json<WorkingDirectoryRequest>, JsonRejection>, body: Result<Json<WorkingDirectoryRequest>, JsonRejection>,
) -> RestResult<RuntimeHttpWorkingDirectoryResponse> { ) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?; let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
if let Some(materialization) = request.materialization.as_ref()
&& materialization.workspace_id != auth.workspace_id
{
return Err(RuntimeHttpRestError::new(
StatusCode::FORBIDDEN,
"working_directory_materialization_workspace_mismatch",
"Repository materialization authority does not match the authenticated Workspace",
));
}
let working_directory = state let working_directory = state
.runtime .runtime
.create_working_directory(request) .create_working_directory_from_resource(request)
.await
.map_err(RuntimeHttpRestError::runtime)?; .map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpWorkingDirectoryResponse { Ok(Json(RuntimeHttpWorkingDirectoryResponse {
working_directory, working_directory,
@@ -1559,6 +1603,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
if path == "/v1/workers" && *method == Method::POST { if path == "/v1/workers" && *method == Method::POST {
return Some("workers:create"); return Some("workers:create");
} }
if path == "/v1/working-directories/repository-access" && *method == Method::POST {
return Some("workdirs:operate");
}
if path.starts_with("/v1/workdir-sessions") if path.starts_with("/v1/workdir-sessions")
|| (path.starts_with("/v1/working-directories/") && path.ends_with("/sessions")) || (path.starts_with("/v1/working-directories/") && path.ends_with("/sessions"))
{ {
@@ -2220,6 +2267,10 @@ mod tests {
#[test] #[test]
fn workdir_routes_require_dedicated_operation_permission() { fn workdir_routes_require_dedicated_operation_permission() {
assert_eq!(
required_runtime_permission(&Method::POST, "/v1/working-directories/repository-access",),
Some("workdirs:operate")
);
assert_eq!( assert_eq!(
required_runtime_permission(&Method::POST, "/v1/working-directories/wd-1/sessions"), required_runtime_permission(&Method::POST, "/v1/working-directories/wd-1/sessions"),
Some("workdirs:operate") Some("workdirs:operate")
+34 -10
View File
@@ -23,10 +23,21 @@ use worker_runtime::http_server::{
RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection, RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection,
}; };
use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend}; use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend};
use worker_runtime::working_directory::LocalGitWorktreeMaterializer; use worker_runtime::working_directory::RuntimeGitCacheMaterializer;
use worker_runtime::{Runtime, RuntimeOptions}; use worker_runtime::{Runtime, RuntimeOptions};
fn main() -> ExitCode { fn main() -> ExitCode {
let mut arguments = std::env::args().skip(1).collect::<Vec<_>>();
if arguments.first().map(String::as_str) == Some("__repository-ssh") {
arguments.remove(0);
return match worker_runtime::working_directory::run_repository_ssh_client(&arguments) {
Ok(status) => ExitCode::from(u8::try_from(status).unwrap_or(1)),
Err(error) => {
eprintln!("{error}");
ExitCode::from(1)
}
};
}
match run() { match run() {
Ok(()) => ExitCode::SUCCESS, Ok(()) => ExitCode::SUCCESS,
Err(error) => { Err(error) => {
@@ -169,6 +180,9 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
factory = factory.with_remote_worker_mutation_identity(identity); factory = factory.with_remote_worker_mutation_identity(identity);
} }
} }
let mut backend_resource_client: Option<
Arc<dyn worker_runtime::resource::BackendResourceClient>,
> = None;
if let Some(endpoint) = config.backend_resource_endpoint.clone() { if let Some(endpoint) = config.backend_resource_endpoint.clone() {
let identity = runtime_auth.identity.as_ref().ok_or_else(|| { let identity = runtime_auth.identity.as_ref().ok_or_else(|| {
ProcessError::Auth( ProcessError::Auth(
@@ -181,26 +195,28 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
.to_owned(), .to_owned(),
)); ));
}; };
factory = factory.with_resource_client(Arc::new( let client = Arc::new(
worker_runtime::resource::HttpBackendResourceClient::new( worker_runtime::resource::HttpBackendResourceClient::new(
endpoint, endpoint,
config.backend_resource_token.clone(), config.backend_resource_token.clone(),
) )
.with_runtime_request_source(identity, trusted_server.server_id.clone()), .with_runtime_request_source(identity, trusted_server.server_id.clone()),
)); );
factory = factory.with_resource_client(client.clone());
backend_resource_client = Some(client);
} }
let backend = Arc::new( let backend = Arc::new(
WorkerRuntimeExecutionBackend::new(factory) WorkerRuntimeExecutionBackend::new(factory)
.map_err(ProcessError::WorkerAdapter)? .map_err(ProcessError::WorkerAdapter)?
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new( .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
fs_paths.workdir_target.clone(), fs_paths.workdir_target.clone(),
)), )),
); );
match &config.http.store { let runtime = match &config.http.store {
RuntimeHttpStoreSelection::Memory => { RuntimeHttpStoreSelection::Memory => {
Runtime::with_execution_backend(runtime_options_from_http(&config.http), backend) Runtime::with_execution_backend(runtime_options_from_http(&config.http), backend)
.map_err(ProcessError::Runtime) .map_err(ProcessError::Runtime)?
} }
RuntimeHttpStoreSelection::Fs { root } => { RuntimeHttpStoreSelection::Fs { root } => {
let mut options = FsRuntimeStoreOptions::new(root.clone()).with_runtime_id( let mut options = FsRuntimeStoreOptions::new(root.clone()).with_runtime_id(
@@ -213,12 +229,20 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
); );
options.display_name = config.http.display_name.clone(); options.display_name = config.http.display_name.clone();
Runtime::with_fs_store_and_execution_backend(options, backend) Runtime::with_fs_store_and_execution_backend(options, backend)
.map_err(ProcessError::Runtime) .map_err(ProcessError::Runtime)?
} }
_ => Err(ProcessError::usage( _ => {
"unsupported Runtime catalog store selection".to_string(), return Err(ProcessError::usage(
)), "unsupported Runtime catalog store selection".to_string(),
));
}
};
if let Some(client) = backend_resource_client {
runtime
.install_backend_resource_client(client)
.map_err(ProcessError::Runtime)?;
} }
Ok(runtime)
} }
fn runtime_options_from_http(config: &RuntimeHttpServerConfig) -> RuntimeOptions { fn runtime_options_from_http(config: &RuntimeHttpServerConfig) -> RuntimeOptions {
+56 -5
View File
@@ -11,18 +11,46 @@ use std::sync::Mutex;
pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str = pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str =
"application/vnd.yoi.profile-source-archive+tar"; "application/vnd.yoi.profile-source-archive+tar";
pub const REPOSITORY_SSH_ACCESS_CONTENT_TYPE: &str =
"application/vnd.yoi.repository-ssh-access+json";
pub const DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES: u64 = 2 * 1024 * 1024; pub const DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES: u64 = 2 * 1024 * 1024;
pub const DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES: u64 = 64 * 1024;
#[derive(Clone, Serialize, Deserialize)]
pub struct RepositorySshAccessSecret {
pub private_key: String,
pub known_hosts_entry: String,
}
impl Drop for RepositorySshAccessSecret {
fn drop(&mut self) {
zeroize::Zeroize::zeroize(&mut self.private_key);
zeroize::Zeroize::zeroize(&mut self.known_hosts_entry);
}
}
impl std::fmt::Debug for RepositorySshAccessSecret {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("RepositorySshAccessSecret")
.field("private_key", &"[REDACTED]")
.field("known_hosts_entry", &"[REDACTED]")
.finish()
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum BackendResourceKind { pub enum BackendResourceKind {
ProfileSourceArchive, ProfileSourceArchive,
RepositorySshAccess,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum BackendResourceOperation { pub enum BackendResourceOperation {
FetchArchive, FetchArchive,
FetchOnce,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -66,7 +94,7 @@ pub struct BackendResourceFetchRequest {
pub audit_correlation_id: String, pub audit_correlation_id: String,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendResourceFetchResponse { pub struct BackendResourceFetchResponse {
pub kind: BackendResourceKind, pub kind: BackendResourceKind,
pub resource_id: String, pub resource_id: String,
@@ -76,6 +104,29 @@ pub struct BackendResourceFetchResponse {
pub audit_correlation_id: String, pub audit_correlation_id: String,
} }
impl std::fmt::Debug for BackendResourceFetchResponse {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("BackendResourceFetchResponse")
.field("kind", &self.kind)
.field("resource_id", &self.resource_id)
.field("digest", &self.digest)
.field("content_type", &self.content_type)
.field(
"bytes",
&format_args!("[REDACTED; {} bytes]", self.bytes.len()),
)
.field("audit_correlation_id", &self.audit_correlation_id)
.finish()
}
}
impl Drop for BackendResourceFetchResponse {
fn drop(&mut self) {
self.bytes.fill(0);
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
#[serde(tag = "code", rename_all = "snake_case")] #[serde(tag = "code", rename_all = "snake_case")]
pub enum BackendResourceError { pub enum BackendResourceError {
@@ -248,7 +299,7 @@ pub fn build_profile_source_archive_fetch_request(
pub fn profile_source_archive_from_response( pub fn profile_source_archive_from_response(
handle: &BackendResourceHandle, handle: &BackendResourceHandle,
response: BackendResourceFetchResponse, mut response: BackendResourceFetchResponse,
) -> Result<ProfileSourceArchive, BackendResourceError> { ) -> Result<ProfileSourceArchive, BackendResourceError> {
if handle.kind != BackendResourceKind::ProfileSourceArchive if handle.kind != BackendResourceKind::ProfileSourceArchive
|| response.kind != BackendResourceKind::ProfileSourceArchive || response.kind != BackendResourceKind::ProfileSourceArchive
@@ -263,7 +314,7 @@ pub fn profile_source_archive_from_response(
if response.content_type != handle.content_type { if response.content_type != handle.content_type {
return Err(BackendResourceError::ContentTypeMismatch { return Err(BackendResourceError::ContentTypeMismatch {
expected: handle.content_type.clone(), expected: handle.content_type.clone(),
actual: response.content_type, actual: response.content_type.clone(),
}); });
} }
let actual_bytes = response.bytes.len() as u64; let actual_bytes = response.bytes.len() as u64;
@@ -278,7 +329,7 @@ pub fn profile_source_archive_from_response(
return Err(BackendResourceError::DigestMismatch { return Err(BackendResourceError::DigestMismatch {
expected: handle.digest.clone(), expected: handle.digest.clone(),
actual: if response.digest != handle.digest { actual: if response.digest != handle.digest {
response.digest response.digest.clone()
} else { } else {
actual_digest actual_digest
}, },
@@ -296,7 +347,7 @@ pub fn profile_source_archive_from_response(
} }
})?, })?,
}, },
content: response.bytes, content: std::mem::take(&mut response.bytes),
}) })
} }
+447 -3
View File
@@ -1,6 +1,6 @@
use crate::catalog::{ use crate::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck, ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck,
WorkerStatus, WorkerSummary, WorkingDirectoryRequest, WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef,
}; };
use crate::config_bundle::{ use crate::config_bundle::{
@@ -26,6 +26,10 @@ use crate::management::{
}; };
#[cfg(feature = "ws-server")] #[cfg(feature = "ws-server")]
use crate::observation::{WorkerObservationCursor, WorkerObservationEvent}; use crate::observation::{WorkerObservationCursor, WorkerObservationEvent};
use crate::resource::{
BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceKind,
REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret,
};
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
use crate::retention::{ use crate::retention::{
FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult,
@@ -172,6 +176,14 @@ impl Runtime {
Ok(runtime) Ok(runtime)
} }
pub fn install_backend_resource_client(
&self,
client: Arc<dyn BackendResourceClient>,
) -> Result<(), RuntimeError> {
self.lock()?.backend_resource_client = Some(BackendResourceClientRef(client));
Ok(())
}
/// Create or restore a filesystem-backed Runtime. /// Create or restore a filesystem-backed Runtime.
/// ///
/// The store is scoped by `options.root`; if the directory already exists, /// The store is scoped by `options.root`; if the directory already exists,
@@ -366,6 +378,103 @@ impl Runtime {
.map_err(RuntimeError::from) .map_err(RuntimeError::from)
} }
pub async fn create_working_directory_from_resource(
&self,
mut request: WorkingDirectoryRequest,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
if let Some(ssh) = request
.materialization
.as_mut()
.and_then(|materialization| materialization.ssh.as_mut())
{
self.resolve_repository_access_resource(ssh).await?;
}
self.create_working_directory(request)
}
pub fn authorize_working_directory_repository_access(
&self,
request: WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory Repository access requires an execution backend"
.to_string(),
}
})?
};
backend
.authorize_working_directory_repository_access(&request)
.map_err(RuntimeError::from)
}
async fn resolve_repository_access_resource(
&self,
ssh: &mut crate::catalog::RepositorySshMaterializationAccess,
) -> Result<(), RuntimeError> {
if !ssh.private_key.expose().is_empty() && !ssh.known_hosts_entry.expose().is_empty() {
return Ok(());
}
let (client, runtime_id) = {
let state = self.lock()?;
let client = state.backend_resource_client.clone().ok_or_else(|| {
RuntimeError::InvalidRequest(
"Backend Repository access resource client is unavailable".to_string(),
)
})?;
let runtime_id = state.runtime_identity.clone().ok_or_else(|| {
RuntimeError::InvalidRequest("Runtime identity is unavailable".to_string())
})?;
(client, runtime_id)
};
let mut response = client
.0
.fetch_resource(BackendResourceFetchRequest {
handle: ssh.secret_resource.clone(),
runtime_id,
worker_id: None,
audit_correlation_id: ssh.secret_resource.audit_correlation_id.clone(),
})
.await
.map_err(repository_resource_error)?;
if response.kind != BackendResourceKind::RepositorySshAccess
|| response.content_type != REPOSITORY_SSH_ACCESS_CONTENT_TYPE
|| response.resource_id != ssh.secret_resource.resource_id
|| response.digest != ssh.secret_resource.digest
|| response.bytes.len() as u64 > ssh.secret_resource.max_bytes
{
return Err(RuntimeError::InvalidRequest(
"Backend Repository SSH access resource response was invalid".to_string(),
));
}
let secret = serde_json::from_slice::<RepositorySshAccessSecret>(&response.bytes);
response.bytes.fill(0);
let mut secret = secret.map_err(|_| {
RuntimeError::InvalidRequest(
"Backend Repository SSH access resource payload was invalid".to_string(),
)
})?;
ssh.private_key =
crate::catalog::SensitiveString::new(std::mem::take(&mut secret.private_key));
ssh.known_hosts_entry =
crate::catalog::SensitiveString::new(std::mem::take(&mut secret.known_hosts_entry));
Ok(())
}
pub async fn authorize_working_directory_repository_access_from_resource(
&self,
mut request: WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), RuntimeError> {
let ssh = request.materialization.ssh.as_mut().ok_or_else(|| {
RuntimeError::InvalidRequest("Repository SSH access metadata is missing".to_string())
})?;
self.resolve_repository_access_resource(ssh).await?;
self.authorize_working_directory_repository_access(request)
}
/// List Runtime-owned working directories through the attached execution backend. /// List Runtime-owned working directories through the attached execution backend.
pub fn list_working_directories( pub fn list_working_directories(
&self, &self,
@@ -566,12 +675,13 @@ impl Runtime {
let worker_id = request.worker_id; let worker_id = request.worker_id;
let worker_ref = WorkerRef::new(worker_id); let worker_ref = WorkerRef::new(worker_id);
let durable_request = durable_create_worker_request(&request);
let record = WorkerRecord { let record = WorkerRecord {
worker_ref: worker_ref.clone(), worker_ref: worker_ref.clone(),
worker_id: worker_id.clone(), worker_id: worker_id.clone(),
status: WorkerStatus::Stopped, status: WorkerStatus::Stopped,
workspace_id: scope.map(|scope| scope.workspace_id.clone()), workspace_id: scope.map(|scope| scope.workspace_id.clone()),
request: request.clone(), request: durable_request,
run_generation: 1, run_generation: 1,
working_directory: None, working_directory: None,
execution_handle: None, execution_handle: None,
@@ -1842,6 +1952,15 @@ struct SubscriptionSink {
lagged: Arc<AtomicBool>, lagged: Arc<AtomicBool>,
} }
#[derive(Clone)]
struct BackendResourceClientRef(Arc<dyn BackendResourceClient>);
impl std::fmt::Debug for BackendResourceClientRef {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str("BackendResourceClientRef(..)")
}
}
#[derive(Debug)] #[derive(Debug)]
struct RuntimeState { struct RuntimeState {
display_name: Option<String>, display_name: Option<String>,
@@ -1853,6 +1972,7 @@ struct RuntimeState {
persistence: RuntimePersistence, persistence: RuntimePersistence,
status: RuntimeStatus, status: RuntimeStatus,
execution_backend: Option<WorkerExecutionBackendRef>, execution_backend: Option<WorkerExecutionBackendRef>,
backend_resource_client: Option<BackendResourceClientRef>,
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
next_diagnostic_id: u64, next_diagnostic_id: u64,
workers: BTreeMap<WorkerId, WorkerRecord>, workers: BTreeMap<WorkerId, WorkerRecord>,
@@ -1880,6 +2000,7 @@ impl RuntimeState {
persistence: RuntimePersistence::Memory, persistence: RuntimePersistence::Memory,
status: RuntimeStatus::Running, status: RuntimeStatus::Running,
execution_backend: None, execution_backend: None,
backend_resource_client: None,
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
next_diagnostic_id: 1, next_diagnostic_id: 1,
workers: BTreeMap::new(), workers: BTreeMap::new(),
@@ -1908,6 +2029,7 @@ impl RuntimeState {
persistence: RuntimePersistence::Fs(store), persistence: RuntimePersistence::Fs(store),
status: RuntimeStatus::Running, status: RuntimeStatus::Running,
execution_backend: None, execution_backend: None,
backend_resource_client: None,
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
next_diagnostic_id: 1, next_diagnostic_id: 1,
workers: BTreeMap::new(), workers: BTreeMap::new(),
@@ -1959,6 +2081,7 @@ impl RuntimeState {
persistence: RuntimePersistence::Fs(store), persistence: RuntimePersistence::Fs(store),
status: persisted.status, status: persisted.status,
execution_backend: None, execution_backend: None,
backend_resource_client: None,
next_diagnostic_id, next_diagnostic_id,
workers, workers,
config_bundles: BTreeMap::new(), config_bundles: BTreeMap::new(),
@@ -2714,6 +2837,33 @@ fn worker_status_from_run_state(run_state: WorkerExecutionRunState) -> WorkerSta
} }
} }
fn repository_resource_error(error: BackendResourceError) -> RuntimeError {
let category = match error {
BackendResourceError::Expired => "expired",
BackendResourceError::Unauthorized { .. } => "unauthorized",
BackendResourceError::UnsupportedKind => "unsupported_kind",
BackendResourceError::MissingResource => "missing_resource",
BackendResourceError::Oversized { .. } => "oversized",
BackendResourceError::DigestMismatch { .. } => "digest_mismatch",
BackendResourceError::ContentTypeMismatch { .. } => "content_type_mismatch",
BackendResourceError::InvalidResponse { .. } => "invalid_response",
BackendResourceError::Transport { .. } => "transport",
};
RuntimeError::InvalidRequest(format!(
"Backend Repository SSH access resource fetch failed: {category}"
))
}
fn durable_create_worker_request(request: &CreateWorkerRequest) -> CreateWorkerRequest {
let mut durable = request.clone();
if let Some(working_directory) = durable.working_directory_request.as_mut()
&& let Some(materialization) = working_directory.materialization.as_mut()
{
materialization.ssh = None;
}
durable
}
fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> { fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> {
request request
.working_directory .working_directory
@@ -2884,7 +3034,9 @@ fn subscription_worker_state(status: WorkerStatus) -> SubscriptionWorkerState {
mod tests { mod tests {
use super::*; use super::*;
use crate::catalog::{ use crate::catalog::{
ConfigBundleRef, ProfileSelector, WorkingDirectoryClaim, WorkspaceApiRef, ConfigBundleRef, MaterializerKind, ProfileSelector, RepositoryMaterializationContext,
RepositorySshMaterializationAccess, SensitiveString, WorkingDirectoryClaim,
WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
}; };
use crate::config_bundle::{ use crate::config_bundle::{
ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration, ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
@@ -2894,6 +3046,8 @@ mod tests {
WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle, WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle,
WorkerExecutionRestoreRequest, WorkerExecutionRunState, WorkerExecutionRestoreRequest, WorkerExecutionRunState,
}; };
use crate::working_directory::WorkingDirectoryDiagnostic;
use async_trait::async_trait;
use std::collections::BTreeMap; use std::collections::BTreeMap;
#[cfg(feature = "fs-store")] #[cfg(feature = "fs-store")]
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
@@ -3115,6 +3269,243 @@ mod tests {
} }
} }
#[test]
fn durable_worker_request_omits_repository_credentials() {
let mut request = task_request("worker-secret-redaction");
request.working_directory_request = Some(WorkingDirectoryRequest {
repository: WorkingDirectoryRepository {
id: "repository-1".to_string(),
provider: "git".to_string(),
source: workspace_api::RepositorySource {
kind: workspace_api::RepositorySourceKind::Ssh,
uri: "ssh://git@example.test/repo.git".to_string(),
},
source_revision: 1,
source_fingerprint: "sha256:source".to_string(),
selector: None,
},
materializer: MaterializerKind::RuntimeGitCache,
backend_workdir_id: Some("working-directory-1".to_string()),
materialization: Some(RepositoryMaterializationContext {
workspace_id: "workspace-1".to_string(),
runtime_id: "runtime-1".to_string(),
operation_id: "operation-1".to_string(),
config_revision: 1,
config_projection_digest: "sha256:projection".to_string(),
cache_generation: 0,
ssh: Some(RepositorySshMaterializationAccess {
credential_id: "credential-1".to_string(),
credential_revision: 1,
host_trust_id: "host-trust-1".to_string(),
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repository-1".to_string(),
repository_source_fingerprint: "sha256:source".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
private_key: SensitiveString::new("private-key-bytes"),
known_hosts_entry: SensitiveString::new("known-hosts-entry"),
}),
}),
});
let durable = durable_create_worker_request(&request);
assert!(
request
.working_directory_request
.as_ref()
.and_then(|working_directory| working_directory.materialization.as_ref())
.and_then(|materialization| materialization.ssh.as_ref())
.is_some()
);
assert!(
durable
.working_directory_request
.as_ref()
.and_then(|working_directory| working_directory.materialization.as_ref())
.and_then(|materialization| materialization.ssh.as_ref())
.is_none()
);
let serialized = serde_json::to_string(&durable).unwrap();
assert!(!serialized.contains("private-key-bytes"));
assert!(!serialized.contains("known-hosts-entry"));
}
fn repository_resource_handle() -> crate::resource::BackendResourceHandle {
crate::resource::BackendResourceHandle {
kind: crate::resource::BackendResourceKind::RepositorySshAccess,
workspace_id: "workspace-1".to_string(),
scope_id: Some("repository-ssh-access".to_string()),
runtime_id: Some("runtime-1".to_string()),
worker_id: None,
resource_id: "repository-access-1".to_string(),
digest: "opaque:repository-access-1".to_string(),
operation: crate::resource::BackendResourceOperation::FetchOnce,
expires_at_unix_seconds: i64::MAX,
nonce: "repository-access-1".to_string(),
revision: "1".to_string(),
generation: None,
max_bytes: crate::resource::DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES,
content_type: crate::resource::REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id: "repository-access-1".to_string(),
profile_source_graph: None,
}
}
#[tokio::test]
async fn repository_access_resource_is_fetched_before_provider_authorization() {
let (runtime, backend) = runtime_and_backend();
backend
.repository_access_available
.store(true, Ordering::SeqCst);
runtime.bind_runtime_identity("runtime-1").unwrap();
let handle = repository_resource_handle();
runtime
.install_backend_resource_client(Arc::new(TestRepositoryResourceClient {
response: Mutex::new(Some(crate::resource::BackendResourceFetchResponse {
kind: crate::resource::BackendResourceKind::RepositorySshAccess,
resource_id: handle.resource_id.clone(),
digest: handle.digest.clone(),
content_type: crate::resource::REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
bytes: serde_json::to_vec(&RepositorySshAccessSecret {
private_key: "private-key-bytes".to_string(),
known_hosts_entry: "known-hosts-entry".to_string(),
})
.unwrap(),
audit_correlation_id: handle.audit_correlation_id.clone(),
})),
}))
.unwrap();
let request = WorkingDirectoryRepositoryAccessRequest {
working_directory_id: "working-directory-1".to_string(),
materialization: RepositoryMaterializationContext {
workspace_id: "workspace-1".to_string(),
runtime_id: "runtime-1".to_string(),
operation_id: "operation-1".to_string(),
config_revision: 1,
config_projection_digest: "sha256:projection".to_string(),
cache_generation: 0,
ssh: Some(RepositorySshMaterializationAccess {
credential_id: "credential-1".to_string(),
credential_revision: 1,
host_trust_id: "host-trust-1".to_string(),
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repository-1".to_string(),
repository_source_fingerprint: "sha256:source".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: handle,
private_key: SensitiveString::default(),
known_hosts_entry: SensitiveString::default(),
}),
},
};
let replay = request.clone();
runtime
.authorize_working_directory_repository_access_from_resource(request)
.await
.unwrap();
assert!(
runtime
.authorize_working_directory_repository_access_from_resource(replay)
.await
.is_err()
);
let accesses = backend.repository_accesses.lock().unwrap();
assert_eq!(accesses.len(), 1);
let access = accesses[0].materialization.ssh.as_ref().unwrap();
assert_eq!(access.private_key.expose(), "private-key-bytes");
assert_eq!(access.known_hosts_entry.expose(), "known-hosts-entry");
}
#[tokio::test]
async fn working_directory_create_fetches_repository_access_before_provider_call() {
let (runtime, backend) = runtime_and_backend();
backend
.repository_access_available
.store(true, Ordering::SeqCst);
runtime.bind_runtime_identity("runtime-1").unwrap();
let handle = repository_resource_handle();
runtime
.install_backend_resource_client(Arc::new(TestRepositoryResourceClient {
response: Mutex::new(Some(crate::resource::BackendResourceFetchResponse {
kind: crate::resource::BackendResourceKind::RepositorySshAccess,
resource_id: handle.resource_id.clone(),
digest: handle.digest.clone(),
content_type: crate::resource::REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
bytes: serde_json::to_vec(&RepositorySshAccessSecret {
private_key: "create-private-key-bytes".to_string(),
known_hosts_entry: "create-known-hosts-entry".to_string(),
})
.unwrap(),
audit_correlation_id: handle.audit_correlation_id.clone(),
})),
}))
.unwrap();
let request = WorkingDirectoryRequest {
repository: WorkingDirectoryRepository {
id: "repository-1".to_string(),
provider: "git".to_string(),
source: workspace_api::RepositorySource {
kind: workspace_api::RepositorySourceKind::Ssh,
uri: "ssh://git@example.test/repo.git".to_string(),
},
source_revision: 1,
source_fingerprint: "sha256:source".to_string(),
selector: None,
},
materializer: MaterializerKind::RuntimeGitCache,
backend_workdir_id: Some("working-directory-1".to_string()),
materialization: Some(RepositoryMaterializationContext {
workspace_id: "workspace-1".to_string(),
runtime_id: "runtime-1".to_string(),
operation_id: "operation-create".to_string(),
config_revision: 1,
config_projection_digest: "sha256:projection".to_string(),
cache_generation: 0,
ssh: Some(RepositorySshMaterializationAccess {
credential_id: "credential-1".to_string(),
credential_revision: 1,
host_trust_id: "host-trust-1".to_string(),
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repository-1".to_string(),
repository_source_fingerprint: "sha256:source".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: handle,
private_key: SensitiveString::default(),
known_hosts_entry: SensitiveString::default(),
}),
}),
};
assert!(
runtime
.create_working_directory_from_resource(request)
.await
.is_err()
);
let requests = backend.working_directory_requests.lock().unwrap();
let access = requests[0]
.materialization
.as_ref()
.and_then(|materialization| materialization.ssh.as_ref())
.unwrap();
assert_eq!(access.private_key.expose(), "create-private-key-bytes");
assert_eq!(
access.known_hosts_entry.expose(),
"create-known-hosts-entry"
);
}
fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest { fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest {
let mut request = task_request(objective); let mut request = task_request(objective);
request.workspace_api = Some(WorkspaceApiRef { request.workspace_api = Some(WorkspaceApiRef {
@@ -3196,6 +3587,9 @@ mod tests {
config_bundles: Mutex<Vec<Option<ConfigBundle>>>, config_bundles: Mutex<Vec<Option<ConfigBundle>>>,
contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>, contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
dispatched_inputs: Mutex<Vec<WorkerInput>>, dispatched_inputs: Mutex<Vec<WorkerInput>>,
repository_accesses: Mutex<Vec<WorkingDirectoryRepositoryAccessRequest>>,
repository_access_available: AtomicBool,
working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
preserve_commit_ack_submission_id: AtomicBool, preserve_commit_ack_submission_id: AtomicBool,
#[cfg(feature = "ws-server")] #[cfg(feature = "ws-server")]
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>, snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
@@ -3236,6 +3630,38 @@ mod tests {
"test-execution-backend" "test-execution-backend"
} }
fn create_working_directory(
&self,
request: &WorkingDirectoryRequest,
) -> Result<CatalogWorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
self.working_directory_requests
.lock()
.unwrap()
.push(request.clone());
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_unsupported",
"Worker execution backend does not support working directory materialization",
))
}
fn authorize_working_directory_repository_access(
&self,
request: &WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), WorkingDirectoryDiagnostic> {
self.repository_accesses
.lock()
.unwrap()
.push(request.clone());
if self.repository_access_available.load(Ordering::SeqCst) {
Ok(())
} else {
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_repository_access_unsupported",
"Worker execution backend does not support Repository access authorization",
))
}
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult { fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
self.run_generations self.run_generations
.lock() .lock()
@@ -3343,6 +3769,24 @@ mod tests {
} }
} }
struct TestRepositoryResourceClient {
response: Mutex<Option<crate::resource::BackendResourceFetchResponse>>,
}
#[async_trait]
impl BackendResourceClient for TestRepositoryResourceClient {
async fn fetch_resource(
&self,
_request: BackendResourceFetchRequest,
) -> Result<crate::resource::BackendResourceFetchResponse, BackendResourceError> {
self.response
.lock()
.unwrap()
.take()
.ok_or(BackendResourceError::MissingResource)
}
}
fn runtime_with_backend() -> Runtime { fn runtime_with_backend() -> Runtime {
let runtime = Runtime::with_execution_backend( let runtime = Runtime::with_execution_backend(
RuntimeOptions::default(), RuntimeOptions::default(),
+45 -11
View File
@@ -20,7 +20,7 @@ use crate::auth::{
}; };
use crate::catalog::{ use crate::catalog::{
CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource, CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
}; };
use crate::execution::{ use crate::execution::{
WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation,
@@ -706,13 +706,17 @@ fn runtime_local_workdir_session(
root: &Path, root: &Path,
cwd: &Path, cwd: &Path,
scope: manifest::SharedScope, scope: manifest::SharedScope,
command_environment: std::collections::BTreeMap<String, String>,
resources: Vec<Arc<dyn workdir::WorkdirSessionResource>>,
) -> WorkdirSessionHandle { ) -> WorkdirSessionHandle {
Arc::new(LocalWorkdirSession::materialized_bound( Arc::new(LocalWorkdirSession::materialized_bound_with_environment(
Workdir::new(workdir_id), Workdir::new(workdir_id),
root.to_path_buf(), root.to_path_buf(),
cwd.to_path_buf(), cwd.to_path_buf(),
scope, scope,
WorkdirSessionCapabilities::ALL, WorkdirSessionCapabilities::ALL,
command_environment,
resources,
)) ))
} }
@@ -893,6 +897,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
binding.root(), binding.root(),
binding.cwd(), binding.cwd(),
worker.scope().clone(), worker.scope().clone(),
binding.command_environment(),
binding.session_resources(),
))); )));
} else { } else {
worker.bind_workdir_session(None); worker.bind_workdir_session(None);
@@ -1071,6 +1077,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
binding.root(), binding.root(),
binding.cwd(), binding.cwd(),
worker.scope().clone(), worker.scope().clone(),
binding.command_environment(),
binding.session_resources(),
))); )));
} else { } else {
worker.bind_workdir_session(None); worker.bind_workdir_session(None);
@@ -1582,6 +1590,19 @@ where
Ok(materializer.create(request)?.status()) Ok(materializer.create(request)?.status())
} }
fn authorize_working_directory_repository_access(
&self,
request: &WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), WorkingDirectoryDiagnostic> {
let materializer = self.working_directory_materializer.as_ref().ok_or_else(|| {
WorkingDirectoryDiagnostic::rejected(
"working_directory_materializer_unavailable",
"working directory Repository access requested, but no materializer is configured for this runtime backend",
)
})?;
materializer.authorize_repository_access(request)
}
fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> { fn list_working_directories(&self) -> Vec<WorkingDirectoryStatus> {
self.working_directory_materializer self.working_directory_materializer
.as_ref() .as_ref()
@@ -1624,6 +1645,8 @@ where
binding.root(), binding.root(),
binding.cwd(), binding.cwd(),
manifest::SharedScope::new(scope), manifest::SharedScope::new(scope),
binding.command_environment(),
binding.session_resources(),
)) ))
} }
@@ -2142,7 +2165,7 @@ mod tests {
use crate::identity::WorkerRef; use crate::identity::WorkerRef;
use crate::management::RuntimeOptions; use crate::management::RuntimeOptions;
use crate::observation::WorkerObservationCursor; use crate::observation::WorkerObservationCursor;
use crate::working_directory::LocalGitWorktreeMaterializer; use crate::working_directory::RuntimeGitCacheMaterializer;
use agen::Engine; use agen::Engine;
use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent}; use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
use agen::llm_client::{ClientError, LlmClient, Request}; use agen::llm_client::{ClientError, LlmClient, Request};
@@ -2728,8 +2751,9 @@ mod tests {
source_fingerprint: "sha256:test".to_string(), source_fingerprint: "sha256:test".to_string(),
selector: Some(RepositorySelector::from("HEAD")), selector: Some(RepositorySelector::from("HEAD")),
}, },
materializer: MaterializerKind::LocalGitWorktree, materializer: MaterializerKind::RuntimeGitCache,
backend_workdir_id: None, backend_workdir_id: None,
materialization: None,
} }
} }
@@ -2853,12 +2877,16 @@ mod tests {
root.path(), root.path(),
root.path(), root.path(),
manifest::SharedScope::new(Scope::writable(root.path()).unwrap()), manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
Default::default(),
Vec::new(),
); );
let restored = runtime_local_workdir_session( let restored = runtime_local_workdir_session(
"working-directory-42", "working-directory-42",
root.path(), root.path(),
root.path(), root.path(),
manifest::SharedScope::new(Scope::writable(root.path()).unwrap()), manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
Default::default(),
Vec::new(),
); );
assert_eq!(spawned.workdir().id().as_str(), "working-directory-42"); assert_eq!(spawned.workdir().id().as_str(), "working-directory-42");
@@ -3330,7 +3358,7 @@ mod tests {
}; };
let backend = WorkerRuntimeExecutionBackend::new(factory) let backend = WorkerRuntimeExecutionBackend::new(factory)
.unwrap() .unwrap()
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new( .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
runtime_base.path(), runtime_base.path(),
)); ));
let runtime = let runtime =
@@ -3485,7 +3513,7 @@ mod tests {
}; };
let backend = WorkerRuntimeExecutionBackend::new(factory) let backend = WorkerRuntimeExecutionBackend::new(factory)
.unwrap() .unwrap()
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new( .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
runtime_base.path(), runtime_base.path(),
)); ));
let runtime = let runtime =
@@ -3524,7 +3552,7 @@ mod tests {
let repo = create_clean_repo(); let repo = create_clean_repo();
let backend = WorkerRuntimeExecutionBackend::new(FailingFactory) let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
.unwrap() .unwrap()
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new( .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
runtime_base.path(), runtime_base.path(),
)); ));
let runtime = let runtime =
@@ -3560,7 +3588,7 @@ mod tests {
let repo = create_clean_repo(); let repo = create_clean_repo();
let backend = WorkerRuntimeExecutionBackend::new(FailingFactory) let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
.unwrap() .unwrap()
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new( .with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
runtime_base.path(), runtime_base.path(),
)); ));
let runtime = let runtime =
@@ -3574,9 +3602,15 @@ mod tests {
assert!(format!("{error:?}").contains("spawn failed")); assert!(format!("{error:?}").contains("spawn failed"));
let working_directories_root = runtime_base.path(); let working_directories_root = runtime_base.path();
let remaining_entries = fs::read_dir(working_directories_root) let remaining_workdirs = fs::read_dir(working_directories_root)
.map(|entries| entries.count()) .map(|entries| {
entries
.flatten()
.filter(|entry| !entry.file_name().to_string_lossy().starts_with('.'))
.count()
})
.unwrap_or(0); .unwrap_or(0);
assert_eq!(remaining_entries, 0); assert_eq!(remaining_workdirs, 0);
assert!(working_directories_root.join(".repository-cache").is_dir());
} }
} }
File diff suppressed because it is too large Load Diff
+1
View File
@@ -48,6 +48,7 @@ tracing.workspace = true
ts-rs = { version = "12.0.1", optional = true } ts-rs = { version = "12.0.1", optional = true }
url.workspace = true url.workspace = true
uuid = { workspace = true, features = ["v7"] } uuid = { workspace = true, features = ["v7"] }
zeroize.workspace = true
webauthn-rs = { workspace = true } webauthn-rs = { workspace = true }
[dev-dependencies] [dev-dependencies]
+48 -8
View File
@@ -25,8 +25,9 @@ use worker_runtime::auth::{CapabilityTokenSigner, capability_claims};
use worker_runtime::catalog::{ use worker_runtime::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef, ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef,
ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail, ProfileSourceArchiveSource, WorkerDetail as EmbeddedWorkerDetail,
WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRequest, WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim,
WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectorySummary, WorkspaceApiRef,
}; };
use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary}; use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
#[cfg(test)] #[cfg(test)]
@@ -39,11 +40,11 @@ use worker_runtime::execution::WorkerExecutionRunState;
use worker_runtime::fs_store::FsRuntimeStoreOptions; use worker_runtime::fs_store::FsRuntimeStoreOptions;
use worker_runtime::http_server::{ use worker_runtime::http_server::{
RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest, RuntimeHttpConfigBundleAvailabilityResponse, RuntimeHttpConfigBundleSyncRequest,
RuntimeHttpErrorResponse, RuntimeHttpSummaryResponse, RuntimeHttpWorkerCompletionsRequest, RuntimeHttpErrorResponse, RuntimeHttpRepositoryAccessResponse, RuntimeHttpSummaryResponse,
RuntimeHttpWorkerCompletionsResponse, RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerCompletionsRequest, RuntimeHttpWorkerCompletionsResponse,
RuntimeHttpWorkerInputResponse, RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerDeleteResponse, RuntimeHttpWorkerInputResponse,
RuntimeHttpWorkerLifecycleResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkerLifecycleRequest, RuntimeHttpWorkerLifecycleResponse,
RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse, RuntimeHttpWorkerResponse, RuntimeHttpWorkerWorkspaceApiRequest, RuntimeHttpWorkersResponse,
RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse, RuntimeHttpWorkingDirectoriesResponse, RuntimeHttpWorkingDirectoryResponse,
RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse, RuntimeHttpWorkspacePromptProjectionRequest, RuntimeHttpWorkspacePromptProjectionResponse,
}; };
@@ -818,6 +819,16 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
} }
} }
fn authorize_working_directory_repository_access(
&self,
_request: WorkingDirectoryRepositoryAccessRequest,
) -> std::result::Result<(), Error> {
Err(Error::InvalidInput(
"Runtime does not support working directory Repository access authorization"
.to_string(),
))
}
fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> { fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
RuntimeList::new(Vec::new(), Vec::new()) RuntimeList::new(Vec::new(), Vec::new())
} }
@@ -1391,6 +1402,23 @@ impl RuntimeRegistry {
Ok(runtime.create_working_directory(request)) Ok(runtime.create_working_directory(request))
} }
pub fn authorize_working_directory_repository_access(
&self,
runtime_id: &str,
request: WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), RuntimeRegistryError> {
validate_backend_identifier("runtime_id", runtime_id)?;
validate_backend_identifier("working_directory_id", &request.working_directory_id)?;
let runtime = self.runtime(runtime_id)?;
runtime
.authorize_working_directory_repository_access(request)
.map_err(|error| RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code: "working_directory_repository_access_failed".to_string(),
message: error.to_string(),
})
}
pub fn list_working_directories( pub fn list_working_directories(
&self, &self,
runtime_id: &str, runtime_id: &str,
@@ -3179,6 +3207,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
} }
} }
fn authorize_working_directory_repository_access(
&self,
request: WorkingDirectoryRepositoryAccessRequest,
) -> std::result::Result<(), Error> {
self.post_json::<_, RuntimeHttpRepositoryAccessResponse>(
"/v1/working-directories/repository-access",
&request,
)
.map(|_| ())
.map_err(|diagnostic| Error::RegistryInconsistency(diagnostic.message))
}
fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> { fn list_working_directories(&self) -> RuntimeList<WorkingDirectoryStatus> {
match self.get_json::<RuntimeHttpWorkingDirectoriesResponse>("/v1/working-directories") { match self.get_json::<RuntimeHttpWorkingDirectoriesResponse>("/v1/working-directories") {
Ok(response) => RuntimeList::new(response.working_directories, Vec::new()), Ok(response) => RuntimeList::new(response.working_directories, Vec::new()),
@@ -4386,7 +4426,7 @@ mod tests {
let handle = bundle.profile_source_archive_handle.as_ref().unwrap(); let handle = bundle.profile_source_archive_handle.as_ref().unwrap();
assert!(bundle.profile_source_archive.is_none()); assert!(bundle.profile_source_archive.is_none());
let response = broker let response = broker
.fetch_profile_source_archive(worker_runtime::resource::BackendResourceFetchRequest { .fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
handle: handle.clone(), handle: handle.clone(),
runtime_id: runtime_id.to_string(), runtime_id: runtime_id.to_string(),
worker_id: None, worker_id: None,
@@ -11,7 +11,7 @@ use ring::rand::{SecureRandom, SystemRandom};
use rusqlite::{OptionalExtension, TransactionBehavior, params}; use rusqlite::{OptionalExtension, TransactionBehavior, params};
use serde::Deserialize; use serde::Deserialize;
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use ssh_key::{Algorithm, HashAlg, PrivateKey, PublicKey}; use ssh_key::{Algorithm, HashAlg, LineEnding, PrivateKey, PublicKey};
use workspace_api::{ use workspace_api::{
CreateRepositorySshCredentialRequest, DeleteRepositorySshCredentialRequest, CreateRepositorySshCredentialRequest, DeleteRepositorySshCredentialRequest,
DeleteRepositorySshHostTrustRequest, PutRepositorySshHostTrustRequest, RepositoryAccessMode, DeleteRepositorySshHostTrustRequest, PutRepositorySshHostTrustRequest, RepositoryAccessMode,
@@ -59,7 +59,6 @@ impl WorkspaceConfigSchemaProvider for RepositoryAccessConfigSchemaProvider {
} }
#[derive(Debug, Default, Deserialize)] #[derive(Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct VirtualWorkspaceConfig { struct VirtualWorkspaceConfig {
#[serde(default)] #[serde(default)]
repository_access: BTreeMap<String, VirtualRepositoryAccess>, repository_access: BTreeMap<String, VirtualRepositoryAccess>,
@@ -219,6 +218,16 @@ fn project_repository_access_evaluation(
}) })
} }
#[derive(Clone)]
pub struct LeasedRepositorySshAccess {
pub credential_id: String,
pub credential_revision: u64,
pub host_trust_id: String,
pub host_trust_revision: u64,
pub private_key: zeroize::Zeroizing<String>,
pub known_hosts_entry: String,
}
#[derive(Clone)] #[derive(Clone)]
pub struct RepositorySecretService { pub struct RepositorySecretService {
store: Arc<SqliteWorkspaceStore>, store: Arc<SqliteWorkspaceStore>,
@@ -829,6 +838,184 @@ impl RepositorySecretService {
}) })
} }
pub fn lease_ssh_materialization_access(
&self,
workspace_id: &str,
binding: &RepositorySshAccessBinding,
) -> Result<LeasedRepositorySshAccess> {
let credential = self
.get_credential(workspace_id, &binding.credential_id, &[])?
.ok_or_else(|| {
Error::InvalidInput(format!(
"unknown Repository SSH credential `{}`",
binding.credential_id
))
})?;
if credential.status != "active" {
return Err(Error::InvalidInput(format!(
"Repository SSH credential `{}` is not active",
binding.credential_id
)));
}
let host_trust = self
.get_host_trust(workspace_id, &binding.host_trust_id, &[])?
.ok_or_else(|| {
Error::InvalidInput(format!(
"unknown Repository SSH host trust `{}`",
binding.host_trust_id
))
})?;
self.lease_ssh_materialization_access_revision(
workspace_id,
&binding.credential_id,
credential.current_revision,
&binding.host_trust_id,
host_trust.current_revision,
)
}
pub fn lease_ssh_materialization_access_revision(
&self,
workspace_id: &str,
credential_id: &str,
credential_revision: u64,
host_trust_id: &str,
host_trust_revision: u64,
) -> Result<LeasedRepositorySshAccess> {
let (private_key, passphrase, hostname, port, host_key) = self.store.with_conn(|conn| {
let private_key = read_sealed_secret(
conn,
workspace_id,
credential_id,
credential_revision,
"private_key",
)?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Repository SSH credential `{credential_id}` revision {credential_revision} is unavailable"
))
})?;
let passphrase = read_sealed_secret(
conn,
workspace_id,
credential_id,
credential_revision,
"passphrase",
)?;
let (hostname, port, host_key) = conn
.query_row(
r#"SELECT h.hostname, h.port, v.host_key
FROM repository_ssh_host_trusts h
JOIN repository_ssh_host_trust_revisions v
ON v.workspace_id = h.workspace_id
AND v.host_trust_id = h.host_trust_id
WHERE h.workspace_id = ?1 AND h.host_trust_id = ?2
AND v.revision = ?3"#,
params![workspace_id, host_trust_id, host_trust_revision as i64],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, i64>(1)? as u16,
row.get::<_, String>(2)?,
))
},
)
.optional()?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Repository SSH host trust `{host_trust_id}` revision {host_trust_revision} is unavailable"
))
})?;
Ok((private_key, passphrase, hostname, port, host_key))
})?;
let private_key = self.unseal(
workspace_id,
credential_id,
credential_revision,
"private_key",
private_key,
)?;
let passphrase = passphrase
.map(|secret| {
self.unseal(
workspace_id,
credential_id,
credential_revision,
"passphrase",
secret,
)
})
.transpose()?;
let private_key =
zeroize::Zeroizing::new(String::from_utf8(private_key).map_err(|_| {
Error::Store("Repository SSH private key plaintext is invalid".to_string())
})?);
let passphrase = passphrase
.map(|value| {
String::from_utf8(value)
.map(zeroize::Zeroizing::new)
.map_err(|_| {
Error::Store("Repository SSH passphrase plaintext is invalid".to_string())
})
})
.transpose()?;
let key = PrivateKey::from_openssh(private_key.as_str()).map_err(|_| {
Error::Store("Repository SSH private key plaintext is invalid".to_string())
})?;
let key = if key.is_encrypted() {
key.decrypt(passphrase.as_deref().ok_or_else(|| {
Error::Store("Repository SSH passphrase revision is unavailable".to_string())
})?)
.map_err(|_| Error::Store("Repository SSH private key decryption failed".to_string()))?
} else {
key
};
let private_key = key
.to_openssh(LineEnding::LF)
.map_err(|_| Error::Store("Repository SSH private key encoding failed".to_string()))?;
let host = if port == 22 {
hostname
} else {
format!("[{hostname}]:{port}")
};
Ok(LeasedRepositorySshAccess {
credential_id: credential_id.to_string(),
credential_revision,
host_trust_id: host_trust_id.to_string(),
host_trust_revision,
private_key,
known_hosts_entry: format!("{host} {host_key}\n"),
})
}
fn unseal(
&self,
workspace_id: &str,
credential_id: &str,
revision: u64,
purpose: &str,
secret: SealedSecret,
) -> Result<Vec<u8>> {
let master_key = self.master_key.as_ref().ok_or_else(|| {
Error::Store("Repository secret encryption authority is unavailable".to_string())
})?;
let unbound = UnboundKey::new(&AES_256_GCM, master_key.as_slice())
.map_err(|_| Error::Store("Repository secret encryption key is invalid".to_string()))?;
let key = LessSafeKey::new(unbound);
let mut plaintext = secret.ciphertext;
let aad = secret_aad(workspace_id, credential_id, revision, purpose);
let plaintext_len = key
.open_in_place(
Nonce::assume_unique_for_key(secret.nonce),
Aad::from(aad.as_bytes()),
&mut plaintext,
)
.map_err(|_| Error::Store("Repository secret decryption failed".to_string()))?
.len();
plaintext.truncate(plaintext_len);
Ok(plaintext)
}
fn seal( fn seal(
&self, &self,
workspace_id: &str, workspace_id: &str,
@@ -968,6 +1155,45 @@ fn insert_secret(
Ok(()) Ok(())
} }
fn read_sealed_secret(
conn: &rusqlite::Connection,
workspace_id: &str,
credential_id: &str,
revision: u64,
purpose: &str,
) -> Result<Option<SealedSecret>> {
let row = conn
.query_row(
r#"SELECT encryption_algorithm, nonce, ciphertext
FROM server_secret_versions
WHERE workspace_id = ?1 AND secret_id = ?2
AND revision = ?3 AND purpose = ?4"#,
params![workspace_id, credential_id, revision, purpose],
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, Vec<u8>>(1)?,
row.get::<_, Vec<u8>>(2)?,
))
},
)
.optional()?;
let Some((algorithm, nonce, ciphertext)) = row else {
return Ok(None);
};
if algorithm != "aes-256-gcm-v1" || nonce.len() != NONCE_BYTES {
return Err(Error::RegistryInconsistency(
"Repository secret envelope is invalid".to_string(),
));
}
let mut nonce_bytes = [0u8; NONCE_BYTES];
nonce_bytes.copy_from_slice(&nonce);
Ok(Some(SealedSecret {
nonce: nonce_bytes,
ciphertext,
}))
}
fn replay_credential_operation( fn replay_credential_operation(
tx: &rusqlite::Transaction<'_>, tx: &rusqlite::Transaction<'_>,
workspace_id: &str, workspace_id: &str,
@@ -1691,6 +1917,29 @@ mod tests {
projection.bindings[0].access, projection.bindings[0].access,
RepositoryAccessMode::ReadOnly RepositoryAccessMode::ReadOnly
); );
let lease = service
.lease_ssh_materialization_access("workspace-a", &projection.bindings[0])
.unwrap();
assert_eq!(lease.credential_revision, 1);
assert_eq!(lease.host_trust_revision, 1);
assert!(lease.private_key.contains("BEGIN OPENSSH PRIVATE KEY"));
assert!(
lease
.known_hosts_entry
.starts_with("example.test ssh-ed25519 ")
);
let exact = service
.lease_ssh_materialization_access_revision(
"workspace-a",
"deploy",
lease.credential_revision,
"example",
lease.host_trust_revision,
)
.unwrap();
assert_eq!(exact.credential_revision, lease.credential_revision);
assert_eq!(exact.host_trust_revision, lease.host_trust_revision);
assert_eq!(exact.known_hosts_entry, lease.known_hosts_entry);
let unknown = config_state( let unknown = config_state(
r#"{ r#"{
+199 -33
View File
@@ -9,7 +9,8 @@ use worker_runtime::resource::{
BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceClient, BackendResourceError, BackendResourceFetchRequest,
BackendResourceFetchResponse, BackendResourceHandle, BackendResourceKind, BackendResourceFetchResponse, BackendResourceHandle, BackendResourceKind,
BackendResourceOperation, DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES, BackendResourceOperation, DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE, ResourceRedactionPolicy, DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE,
REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret, ResourceRedactionPolicy,
}; };
#[derive(Clone, Default)] #[derive(Clone, Default)]
@@ -29,7 +30,34 @@ struct StoredResource {
runtime_id: Option<String>, runtime_id: Option<String>,
worker: Option<RuntimeWorkerRef>, worker: Option<RuntimeWorkerRef>,
handle: BackendResourceHandle, handle: BackendResourceHandle,
archive: ProfileSourceArchive, bytes: Vec<u8>,
archive: Option<ProfileSourceArchive>,
one_shot: bool,
}
impl StoredResource {
fn byte_len(&self) -> usize {
self.archive
.as_ref()
.map(|archive| archive.content.len())
.unwrap_or_else(|| self.bytes.len())
}
fn take_bytes(&mut self) -> Vec<u8> {
self.archive
.as_mut()
.map(|archive| std::mem::take(&mut archive.content))
.unwrap_or_else(|| std::mem::take(&mut self.bytes))
}
}
impl Drop for StoredResource {
fn drop(&mut self) {
self.bytes.fill(0);
if let Some(archive) = self.archive.as_mut() {
archive.content.fill(0);
}
}
} }
impl BackendResourceBroker { impl BackendResourceBroker {
@@ -73,7 +101,9 @@ impl BackendResourceBroker {
runtime_id, runtime_id,
worker, worker,
handle: handle.clone(), handle: handle.clone(),
archive, bytes: Vec::new(),
archive: Some(archive),
one_shot: false,
}; };
if let Ok(mut resources) = self.resources.lock() { if let Ok(mut resources) = self.resources.lock() {
resources.insert(nonce, stored); resources.insert(nonce, stored);
@@ -81,6 +111,84 @@ impl BackendResourceBroker {
handle handle
} }
pub fn issue_repository_ssh_access_handle(
&self,
workspace_id: impl Into<String>,
runtime_id: &str,
resource_id: impl Into<String>,
revision: impl Into<String>,
expires_at_unix_seconds: i64,
secret: RepositorySshAccessSecret,
) -> Result<BackendResourceHandle, BackendResourceError> {
let bytes =
serde_json::to_vec(&secret).map_err(|error| BackendResourceError::InvalidResponse {
message: error.to_string(),
})?;
if bytes.len() as u64 > DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES {
return Err(BackendResourceError::Oversized {
max_bytes: DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES,
actual_bytes: bytes.len() as u64,
});
}
let workspace_id = workspace_id.into();
let resource_id = resource_id.into();
let revision = revision.into();
let nonce = Uuid::now_v7().to_string();
let handle = BackendResourceHandle {
kind: BackendResourceKind::RepositorySshAccess,
workspace_id,
scope_id: Some("repository-ssh-access".to_string()),
runtime_id: Some(runtime_id.to_string()),
worker_id: None,
resource_id,
digest: format!("opaque:{nonce}"),
operation: BackendResourceOperation::FetchOnce,
expires_at_unix_seconds,
nonce: nonce.clone(),
revision,
generation: None,
max_bytes: DEFAULT_REPOSITORY_SSH_ACCESS_MAX_BYTES,
content_type: REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
redaction: ResourceRedactionPolicy::RuntimeInternalOnly,
audit_correlation_id: format!("repository-ssh-access-{nonce}"),
profile_source_graph: None,
};
let stored = StoredResource {
runtime_id: Some(runtime_id.to_string()),
worker: None,
handle: handle.clone(),
bytes,
archive: None,
one_shot: true,
};
let resource_key = nonce.clone();
self.resources
.lock()
.map_err(|_| BackendResourceError::Transport {
message: "resource broker lock poisoned".to_string(),
})?
.insert(resource_key.clone(), stored);
if expires_at_unix_seconds != i64::MAX {
let resources = self.resources.clone();
std::thread::spawn(move || {
let now = Utc::now().timestamp();
if expires_at_unix_seconds > now {
std::thread::sleep(std::time::Duration::from_secs(
(expires_at_unix_seconds - now) as u64,
));
}
if let Ok(mut resources) = resources.lock()
&& resources
.get(&resource_key)
.is_some_and(|stored| stored.handle.nonce == resource_key)
{
resources.remove(&resource_key);
}
});
}
Ok(handle)
}
pub fn profile_source_archive( pub fn profile_source_archive(
&self, &self,
digest: &str, digest: &str,
@@ -90,20 +198,21 @@ impl BackendResourceBroker {
.ok()? .ok()?
.values() .values()
.find(|resource| resource.handle.digest == digest) .find(|resource| resource.handle.digest == digest)
.map(|resource| resource.archive.clone()) .and_then(|resource| resource.archive.clone())
} }
pub fn fetch_profile_source_archive( pub fn fetch_resource(
&self, &self,
request: BackendResourceFetchRequest, request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> { ) -> Result<BackendResourceFetchResponse, BackendResourceError> {
verify_handle_shape(&request.handle)?; verify_handle_shape(&request.handle)?;
let stored = self let mut resources = self
.resources .resources
.lock() .lock()
.map_err(|_| BackendResourceError::Transport { .map_err(|_| BackendResourceError::Transport {
message: "resource broker lock poisoned".to_string(), message: "resource broker lock poisoned".to_string(),
})? })?;
let mut stored = resources
.get(&request.handle.nonce) .get(&request.handle.nonce)
.cloned() .cloned()
.ok_or(BackendResourceError::MissingResource)?; .ok_or(BackendResourceError::MissingResource)?;
@@ -111,7 +220,7 @@ impl BackendResourceBroker {
if stored.handle.expires_at_unix_seconds < Utc::now().timestamp() { if stored.handle.expires_at_unix_seconds < Utc::now().timestamp() {
return Err(BackendResourceError::Expired); return Err(BackendResourceError::Expired);
} }
let actual_bytes = stored.archive.content.len() as u64; let actual_bytes = stored.byte_len() as u64;
if actual_bytes > stored.handle.max_bytes { if actual_bytes > stored.handle.max_bytes {
return Err(BackendResourceError::Oversized { return Err(BackendResourceError::Oversized {
max_bytes: stored.handle.max_bytes, max_bytes: stored.handle.max_bytes,
@@ -139,12 +248,15 @@ impl BackendResourceBroker {
}); });
} }
} }
if stored.one_shot {
resources.remove(&request.handle.nonce);
}
Ok(BackendResourceFetchResponse { Ok(BackendResourceFetchResponse {
kind: BackendResourceKind::ProfileSourceArchive, kind: stored.handle.kind.clone(),
resource_id: stored.archive.reference.id, resource_id: stored.handle.resource_id.clone(),
digest: stored.archive.reference.digest, digest: stored.handle.digest.clone(),
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), content_type: stored.handle.content_type.clone(),
bytes: stored.archive.content, bytes: stored.take_bytes(),
audit_correlation_id: request.audit_correlation_id, audit_correlation_id: request.audit_correlation_id,
}) })
} }
@@ -156,24 +268,38 @@ impl BackendResourceClient for BackendResourceBroker {
&self, &self,
request: BackendResourceFetchRequest, request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> { ) -> Result<BackendResourceFetchResponse, BackendResourceError> {
self.fetch_profile_source_archive(request) self.fetch_resource(request)
} }
} }
fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendResourceError> { fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendResourceError> {
if handle.kind != BackendResourceKind::ProfileSourceArchive { match handle.kind {
return Err(BackendResourceError::UnsupportedKind); BackendResourceKind::ProfileSourceArchive => {
} if handle.operation != BackendResourceOperation::FetchArchive {
if handle.operation != BackendResourceOperation::FetchArchive { return Err(BackendResourceError::Unauthorized {
return Err(BackendResourceError::Unauthorized { message: "resource handle operation is not fetch_archive".to_string(),
message: "resource handle operation is not fetch_archive".to_string(), });
}); }
} if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE {
if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE { return Err(BackendResourceError::ContentTypeMismatch {
return Err(BackendResourceError::ContentTypeMismatch { expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(), actual: handle.content_type.clone(),
actual: handle.content_type.clone(), });
}); }
}
BackendResourceKind::RepositorySshAccess => {
if handle.operation != BackendResourceOperation::FetchOnce {
return Err(BackendResourceError::Unauthorized {
message: "resource handle operation is not fetch_once".to_string(),
});
}
if handle.content_type != REPOSITORY_SSH_ACCESS_CONTENT_TYPE {
return Err(BackendResourceError::ContentTypeMismatch {
expected: REPOSITORY_SSH_ACCESS_CONTENT_TYPE.to_string(),
actual: handle.content_type.clone(),
});
}
}
} }
Ok(()) Ok(())
} }
@@ -249,7 +375,7 @@ mod tests {
archive(), archive(),
); );
let response = broker let response = broker
.fetch_profile_source_archive(BackendResourceFetchRequest { .fetch_resource(BackendResourceFetchRequest {
handle: handle.clone(), handle: handle.clone(),
runtime_id: runtime_id.to_string(), runtime_id: runtime_id.to_string(),
worker_id: None, worker_id: None,
@@ -260,6 +386,46 @@ mod tests {
assert_eq!(response.content_type, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE); assert_eq!(response.content_type, PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE);
} }
#[test]
fn repository_ssh_access_resource_is_runtime_bound_and_one_shot() {
let broker = BackendResourceBroker::default();
let handle = broker
.issue_repository_ssh_access_handle(
"workspace-test",
"runtime-test",
"repository-access-test",
"1",
i64::MAX,
RepositorySshAccessSecret {
private_key: "private-key-bytes".to_string(),
known_hosts_entry: "known-hosts-entry".to_string(),
},
)
.unwrap();
let unauthorized = broker
.fetch_resource(request(handle.clone(), "runtime-other", None))
.unwrap_err();
assert!(matches!(
unauthorized,
BackendResourceError::Unauthorized { .. }
));
let response = broker
.fetch_resource(request(handle.clone(), "runtime-test", None))
.unwrap();
assert_eq!(response.kind, BackendResourceKind::RepositorySshAccess);
assert_eq!(response.content_type, REPOSITORY_SSH_ACCESS_CONTENT_TYPE);
let debug = format!("{response:?}");
assert!(!debug.contains("private-key-bytes"));
assert!(debug.contains("REDACTED"));
let secret: RepositorySshAccessSecret = serde_json::from_slice(&response.bytes).unwrap();
assert_eq!(secret.private_key, "private-key-bytes");
assert!(matches!(
broker.fetch_resource(request(handle, "runtime-test", None)),
Err(BackendResourceError::MissingResource)
));
}
#[test] #[test]
fn broker_rejects_runtime_mismatch() { fn broker_rejects_runtime_mismatch() {
let broker = BackendResourceBroker::default(); let broker = BackendResourceBroker::default();
@@ -270,7 +436,7 @@ mod tests {
archive(), archive(),
); );
let err = broker let err = broker
.fetch_profile_source_archive(request(handle, "runtime-b", None)) .fetch_resource(request(handle, "runtime-b", None))
.unwrap_err(); .unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. })); assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
} }
@@ -287,7 +453,7 @@ mod tests {
archive(), archive(),
); );
let err = broker let err = broker
.fetch_profile_source_archive(request(handle, runtime_id, Some(&worker_b.worker_id))) .fetch_resource(request(handle, runtime_id, Some(&worker_b.worker_id)))
.unwrap_err(); .unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. })); assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
} }
@@ -312,7 +478,7 @@ mod tests {
let mut extended = handle; let mut extended = handle;
extended.expires_at_unix_seconds = 4_102_444_800; extended.expires_at_unix_seconds = 4_102_444_800;
let err = broker let err = broker
.fetch_profile_source_archive(request(extended, &runtime_id, None)) .fetch_resource(request(extended, &runtime_id, None))
.unwrap_err(); .unwrap_err();
assert!(matches!(err, BackendResourceError::Expired)); assert!(matches!(err, BackendResourceError::Expired));
} }
@@ -328,7 +494,7 @@ mod tests {
); );
handle.scope_id = Some("tampered-scope".to_string()); handle.scope_id = Some("tampered-scope".to_string());
let err = broker let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, None)) .fetch_resource(request(handle, &runtime_id, None))
.unwrap_err(); .unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. })); assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
} }
@@ -345,7 +511,7 @@ mod tests {
); );
handle.max_bytes = DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1024; handle.max_bytes = DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1024;
let err = broker let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, None)) .fetch_resource(request(handle, &runtime_id, None))
.unwrap_err(); .unwrap_err();
assert!(matches!(err, BackendResourceError::Oversized { .. })); assert!(matches!(err, BackendResourceError::Oversized { .. }));
} }
File diff suppressed because it is too large Load Diff
+161 -32
View File
@@ -257,6 +257,11 @@ const MIGRATIONS: &[Migration] = &[
name: "create Workspace Repository SSH secret authority", name: "create Workspace Repository SSH secret authority",
apply: create_repository_ssh_secret_authority, apply: create_repository_ssh_secret_authority,
}, },
Migration {
version: 47,
name: "bind Workdir create repository access evidence",
apply: bind_workdir_create_repository_access_evidence,
},
]; ];
struct Migration { struct Migration {
@@ -590,6 +595,16 @@ pub struct WorkdirCreateOperationRecord {
pub resolved_runtime_id: String, pub resolved_runtime_id: String,
pub config_revision: u64, pub config_revision: u64,
pub config_projection_digest: String, pub config_projection_digest: String,
pub source_kind: Option<String>,
pub source_uri: Option<String>,
pub source_revision: Option<u64>,
pub source_fingerprint: Option<String>,
pub credential_id: Option<String>,
pub credential_revision: Option<u64>,
pub host_trust_id: Option<String>,
pub host_trust_revision: Option<u64>,
pub repository_access_mode: Option<String>,
pub cache_generation: u64,
pub working_directory_id: String, pub working_directory_id: String,
pub state: String, pub state: String,
pub failure: Option<String>, pub failure: Option<String>,
@@ -605,8 +620,11 @@ pub struct WorkdirRegistryRecord {
pub repository_id: String, pub repository_id: String,
pub creation_selector: Option<String>, pub creation_selector: Option<String>,
pub creation_ref: Option<String>, pub creation_ref: Option<String>,
pub creation_tree: Option<String>,
pub current_selector: Option<String>, pub current_selector: Option<String>,
pub current_ref: Option<String>, pub current_ref: Option<String>,
pub current_tree: Option<String>,
pub observed_at_epoch_seconds: Option<u64>,
pub materialization_status: String, pub materialization_status: String,
pub cleanliness: String, pub cleanliness: String,
pub created_at: String, pub created_at: String,
@@ -4639,16 +4657,20 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
conn.execute( conn.execute(
r#"INSERT INTO workdir_registry ( r#"INSERT INTO workdir_registry (
workspace_id, workdir_id, runtime_id, repository_id, workspace_id, workdir_id, runtime_id, repository_id,
creation_selector, creation_ref, current_selector, current_ref, creation_selector, creation_ref, creation_tree,
current_selector, current_ref, current_tree, observed_at_epoch_seconds,
materialization_status, cleanliness, created_at, updated_at materialization_status, cleanliness, created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)
ON CONFLICT(workspace_id, workdir_id) DO UPDATE SET ON CONFLICT(workspace_id, workdir_id) DO UPDATE SET
runtime_id = excluded.runtime_id, runtime_id = excluded.runtime_id,
repository_id = excluded.repository_id, repository_id = excluded.repository_id,
creation_selector = excluded.creation_selector, creation_selector = excluded.creation_selector,
creation_ref = excluded.creation_ref, creation_ref = excluded.creation_ref,
creation_tree = excluded.creation_tree,
current_selector = excluded.current_selector, current_selector = excluded.current_selector,
current_ref = excluded.current_ref, current_ref = excluded.current_ref,
current_tree = excluded.current_tree,
observed_at_epoch_seconds = excluded.observed_at_epoch_seconds,
materialization_status = excluded.materialization_status, materialization_status = excluded.materialization_status,
cleanliness = excluded.cleanliness, cleanliness = excluded.cleanliness,
updated_at = excluded.updated_at"#, updated_at = excluded.updated_at"#,
@@ -4659,8 +4681,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.repository_id, record.repository_id,
record.creation_selector, record.creation_selector,
record.creation_ref, record.creation_ref,
record.creation_tree,
record.current_selector, record.current_selector,
record.current_ref, record.current_ref,
record.current_tree,
record.observed_at_epoch_seconds.map(|value| value as i64),
record.materialization_status, record.materialization_status,
record.cleanliness, record.cleanliness,
record.created_at, record.created_at,
@@ -5880,7 +5905,8 @@ fn require_expected_ticket_assignment(
fn workdir_registry_select_sql(where_clause: &str) -> String { fn workdir_registry_select_sql(where_clause: &str) -> String {
format!( format!(
"SELECT workspace_id, workdir_id, runtime_id, repository_id, \ "SELECT workspace_id, workdir_id, runtime_id, repository_id, \
creation_selector, creation_ref, current_selector, current_ref, \ creation_selector, creation_ref, creation_tree, \
current_selector, current_ref, current_tree, observed_at_epoch_seconds, \
materialization_status, cleanliness, created_at, updated_at \ materialization_status, cleanliness, created_at, updated_at \
FROM workdir_registry {where_clause}" FROM workdir_registry {where_clause}"
) )
@@ -5896,12 +5922,15 @@ fn read_workdir_registry_record(
repository_id: row.get(3)?, repository_id: row.get(3)?,
creation_selector: row.get(4)?, creation_selector: row.get(4)?,
creation_ref: row.get(5)?, creation_ref: row.get(5)?,
current_selector: row.get(6)?, creation_tree: row.get(6)?,
current_ref: row.get(7)?, current_selector: row.get(7)?,
materialization_status: row.get(8)?, current_ref: row.get(8)?,
cleanliness: row.get(9)?, current_tree: row.get(9)?,
created_at: row.get(10)?, observed_at_epoch_seconds: row.get::<_, Option<i64>>(10)?.map(|value| value as u64),
updated_at: row.get(11)?, materialization_status: row.get(11)?,
cleanliness: row.get(12)?,
created_at: row.get(13)?,
updated_at: row.get(14)?,
}) })
} }
@@ -6839,6 +6868,28 @@ fn create_repository_ssh_secret_authority(conn: &Connection) -> Result<()> {
Ok(()) Ok(())
} }
fn bind_workdir_create_repository_access_evidence(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
ALTER TABLE workdir_registry ADD COLUMN creation_tree TEXT;
ALTER TABLE workdir_registry ADD COLUMN current_tree TEXT;
ALTER TABLE workdir_registry ADD COLUMN observed_at_epoch_seconds INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN source_kind TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN source_uri TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN source_revision INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN source_fingerprint TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN credential_id TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN credential_revision INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN host_trust_id TEXT;
ALTER TABLE workdir_create_operations ADD COLUMN host_trust_revision INTEGER;
ALTER TABLE workdir_create_operations ADD COLUMN repository_access_mode TEXT;
ALTER TABLE workdir_create_operations
ADD COLUMN cache_generation INTEGER NOT NULL DEFAULT 0;
"#,
)?;
Ok(())
}
fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> { fn create_workspace_catalog_operations(conn: &Connection) -> Result<()> {
conn.execute_batch( conn.execute_batch(
r#" r#"
@@ -9724,7 +9775,7 @@ mod tests {
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
let remote = conn let remote = conn
.query_row( .query_row(
"SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \ "SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \
@@ -9802,7 +9853,7 @@ mod tests {
let before = std::fs::read(&path).unwrap(); let before = std::fs::read(&path).unwrap();
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap(); let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
assert_eq!(plan.current_schema_version, 36); assert_eq!(plan.current_schema_version, 36);
assert_eq!(plan.target_schema_version, 46); assert_eq!(plan.target_schema_version, 47);
assert!(plan.migration_required); assert!(plan.migration_required);
assert_eq!(plan.worker_count, 1); assert_eq!(plan.worker_count, 1);
assert_eq!(plan.mappings[0].legacy_worker_id, 7); assert_eq!(plan.mappings[0].legacy_worker_id, 7);
@@ -9816,7 +9867,7 @@ mod tests {
store store
.with_conn(|conn| { .with_conn(|conn| {
assert!(table_exists(conn, "worker_diagnostics_archives")?); assert!(table_exists(conn, "worker_diagnostics_archives")?);
assert_eq!(current_schema_version(conn)?, 46); assert_eq!(current_schema_version(conn)?, 47);
Ok(()) Ok(())
}) })
.unwrap(); .unwrap();
@@ -9952,7 +10003,7 @@ mod tests {
), ),
] ]
); );
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
let foreign_key_error: Option<String> = conn let foreign_key_error: Option<String> = conn
.query_row("PRAGMA foreign_key_check", [], |row| row.get(0)) .query_row("PRAGMA foreign_key_check", [], |row| row.get(0))
.optional() .optional()
@@ -10081,7 +10132,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap()); assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
let controller_worker_id: String = conn let controller_worker_id: String = conn
.query_row( .query_row(
@@ -10199,7 +10250,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap()); assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
} }
@@ -10217,7 +10268,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
let settings = conn let settings = conn
.query_row( .query_row(
"SELECT settings_revision, language FROM workspace_memory_settings \ "SELECT settings_revision, language FROM workspace_memory_settings \
@@ -10258,7 +10309,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(table_exists(&conn, "flow_sources").unwrap()); assert!(table_exists(&conn, "flow_sources").unwrap());
assert!(table_exists(&conn, "flow_source_revisions").unwrap()); assert!(table_exists(&conn, "flow_source_revisions").unwrap());
assert!(!table_exists(&conn, "flow_instances").unwrap()); assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -10325,7 +10376,7 @@ INSERT INTO worker_workdir_attachment_reservations (
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
let repositories_sql: String = conn let repositories_sql: String = conn
.query_row( .query_row(
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'", "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -10508,7 +10559,7 @@ INSERT INTO workdir_registry (
let db = dir.path().join("control-plane.sqlite"); let db = dir.path().join("control-plane.sqlite");
let store = SqliteWorkspaceStore::open(&db).unwrap(); let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46); assert_eq!(store.schema_version().await.unwrap(), 47);
assert!( assert!(
!store !store
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials")) .with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -10525,7 +10576,7 @@ INSERT INTO workdir_registry (
store.upsert_workspace(&record).await.unwrap(); store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(reopened.schema_version().await.unwrap(), 46); assert_eq!(reopened.schema_version().await.unwrap(), 47);
assert_eq!( assert_eq!(
reopened.get_workspace("local-dev").await.unwrap(), reopened.get_workspace("local-dev").await.unwrap(),
Some(record) Some(record)
@@ -11290,7 +11341,7 @@ INSERT INTO worker_registry (
let migrated = SqliteWorkspaceStore::open(&db_path).unwrap(); let migrated = SqliteWorkspaceStore::open(&db_path).unwrap();
migrated migrated
.with_conn(|conn| { .with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 46); assert_eq!(current_schema_version(conn)?, 47);
assert_eq!( assert_eq!(
conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?, conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?,
1, 1,
@@ -11647,13 +11698,16 @@ INSERT INTO worker_registry (
DROP TABLE repository_ssh_host_trust_revisions; DROP TABLE repository_ssh_host_trust_revisions;
DROP TABLE repository_ssh_host_trusts; DROP TABLE repository_ssh_host_trusts;
DROP TABLE workdir_create_operations; DROP TABLE workdir_create_operations;
DELETE FROM __yoi_schema_migrations WHERE version IN (45, 46);", ALTER TABLE workdir_registry DROP COLUMN creation_tree;
ALTER TABLE workdir_registry DROP COLUMN current_tree;
ALTER TABLE workdir_registry DROP COLUMN observed_at_epoch_seconds;
DELETE FROM __yoi_schema_migrations WHERE version IN (45, 46, 47);",
) )
.unwrap(); .unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 44); assert_eq!(current_schema_version(&conn).unwrap(), 44);
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
assert!(table_exists(&conn, "workdir_create_operations").unwrap()); assert!(table_exists(&conn, "workdir_create_operations").unwrap());
let columns = table_columns(&conn, "workdir_create_operations").unwrap(); let columns = table_columns(&conn, "workdir_create_operations").unwrap();
for required in [ for required in [
@@ -11685,13 +11739,26 @@ INSERT INTO worker_registry (
DROP TABLE repository_ssh_credentials; DROP TABLE repository_ssh_credentials;
DROP TABLE repository_ssh_host_trust_revisions; DROP TABLE repository_ssh_host_trust_revisions;
DROP TABLE repository_ssh_host_trusts; DROP TABLE repository_ssh_host_trusts;
DELETE FROM __yoi_schema_migrations WHERE version = 46;", ALTER TABLE workdir_registry DROP COLUMN creation_tree;
ALTER TABLE workdir_registry DROP COLUMN current_tree;
ALTER TABLE workdir_registry DROP COLUMN observed_at_epoch_seconds;
ALTER TABLE workdir_create_operations DROP COLUMN source_kind;
ALTER TABLE workdir_create_operations DROP COLUMN source_uri;
ALTER TABLE workdir_create_operations DROP COLUMN source_revision;
ALTER TABLE workdir_create_operations DROP COLUMN source_fingerprint;
ALTER TABLE workdir_create_operations DROP COLUMN credential_id;
ALTER TABLE workdir_create_operations DROP COLUMN credential_revision;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_id;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_revision;
ALTER TABLE workdir_create_operations DROP COLUMN repository_access_mode;
ALTER TABLE workdir_create_operations DROP COLUMN cache_generation;
DELETE FROM __yoi_schema_migrations WHERE version IN (46, 47);",
) )
.unwrap(); .unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 45); assert_eq!(current_schema_version(&conn).unwrap(), 45);
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
for table in [ for table in [
"repository_ssh_credentials", "repository_ssh_credentials",
"repository_ssh_credential_revisions", "repository_ssh_credential_revisions",
@@ -11710,19 +11777,72 @@ INSERT INTO worker_registry (
assert!(foreign_key_error.is_none()); assert!(foreign_key_error.is_none());
} }
#[test]
fn schema_v47_binds_workdir_create_repository_access_evidence() {
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations(&conn).unwrap();
conn.execute_batch(
"ALTER TABLE workdir_registry DROP COLUMN creation_tree;
ALTER TABLE workdir_registry DROP COLUMN current_tree;
ALTER TABLE workdir_registry DROP COLUMN observed_at_epoch_seconds;
ALTER TABLE workdir_create_operations DROP COLUMN source_kind;
ALTER TABLE workdir_create_operations DROP COLUMN source_uri;
ALTER TABLE workdir_create_operations DROP COLUMN source_revision;
ALTER TABLE workdir_create_operations DROP COLUMN source_fingerprint;
ALTER TABLE workdir_create_operations DROP COLUMN credential_id;
ALTER TABLE workdir_create_operations DROP COLUMN credential_revision;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_id;
ALTER TABLE workdir_create_operations DROP COLUMN host_trust_revision;
ALTER TABLE workdir_create_operations DROP COLUMN repository_access_mode;
ALTER TABLE workdir_create_operations DROP COLUMN cache_generation;
DELETE FROM __yoi_schema_migrations WHERE version = 47;",
)
.unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 47);
let columns = table_columns(&conn, "workdir_create_operations").unwrap();
for required in [
"source_kind",
"source_uri",
"source_revision",
"source_fingerprint",
"credential_id",
"credential_revision",
"host_trust_id",
"host_trust_revision",
"repository_access_mode",
"cache_generation",
] {
assert!(
columns.iter().any(|column| column == required),
"missing column {required}"
);
}
let workdir_columns = table_columns(&conn, "workdir_registry").unwrap();
for required in ["creation_tree", "current_tree", "observed_at_epoch_seconds"] {
assert!(
workdir_columns.iter().any(|column| column == required),
"missing column {required}"
);
}
}
#[test] #[test]
fn server_refuses_a_database_from_a_newer_schema_generation() { fn server_refuses_a_database_from_a_newer_schema_generation() {
let conn = Connection::open_in_memory().unwrap(); let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap(); configure_sqlite(&conn).unwrap();
apply_migrations(&conn).unwrap(); apply_migrations(&conn).unwrap();
conn.execute( conn.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (47, 'future')", "INSERT INTO __yoi_schema_migrations (version, name) VALUES (48, 'future')",
[], [],
) )
.unwrap(); .unwrap();
let error = apply_migrations(&conn).unwrap_err().to_string(); let error = apply_migrations(&conn).unwrap_err().to_string();
assert!(error.contains("schema version 47 is newer"), "{error}"); assert!(error.contains("schema version 48 is newer"), "{error}");
assert!(error.contains("refusing to serve"), "{error}"); assert!(error.contains("refusing to serve"), "{error}");
} }
@@ -11943,7 +12063,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
apply_migrations(&mut conn).unwrap(); apply_migrations(&mut conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 46); assert_eq!(current_schema_version(&conn).unwrap(), 47);
let workspace_id: Option<String> = conn let workspace_id: Option<String> = conn
.query_row( .query_row(
"SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'", "SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'",
@@ -12488,6 +12608,9 @@ WHERE workspace_id = 'workspace-a'
"updated_at", "updated_at",
"current_selector", "current_selector",
"current_ref", "current_ref",
"creation_tree",
"current_tree",
"observed_at_epoch_seconds",
], ],
); );
assert_columns( assert_columns(
@@ -12566,7 +12689,7 @@ WHERE workspace_id = 'workspace-a'
.unwrap(); .unwrap();
let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46); assert_eq!(store.schema_version().await.unwrap(), 47);
store store
.with_conn(|conn| { .with_conn(|conn| {
@@ -12755,7 +12878,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test] #[tokio::test]
async fn repository_records_round_trip() { async fn repository_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46); assert_eq!(store.schema_version().await.unwrap(), 47);
let workspace = WorkspaceRecord { let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
owner_account_id: None, owner_account_id: None,
@@ -12833,7 +12956,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test] #[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() { async fn memory_authority_records_round_trip_and_close_staging() {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46); assert_eq!(store.schema_version().await.unwrap(), 47);
let workspace = WorkspaceRecord { let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(), workspace_id: "local-dev".to_string(),
owner_account_id: None, owner_account_id: None,
@@ -12979,8 +13102,11 @@ CREATE TABLE ticket_assignment_operations (
repository_id: "repo".to_string(), repository_id: "repo".to_string(),
creation_selector: Some("develop".to_string()), creation_selector: Some("develop".to_string()),
creation_ref: Some("abcdef".to_string()), creation_ref: Some("abcdef".to_string()),
creation_tree: Some("tree-creation".to_string()),
current_selector: None, current_selector: None,
current_ref: Some("abcdef".to_string()), current_ref: Some("abcdef".to_string()),
current_tree: Some("tree-current".to_string()),
observed_at_epoch_seconds: Some(1_777_777_777),
materialization_status: "not_found".to_string(), materialization_status: "not_found".to_string(),
cleanliness: "clean".to_string(), cleanliness: "clean".to_string(),
created_at: "2".to_string(), created_at: "2".to_string(),
@@ -12994,8 +13120,11 @@ CREATE TABLE ticket_assignment_operations (
repository_id: "repo".to_string(), repository_id: "repo".to_string(),
creation_selector: Some("feature".to_string()), creation_selector: Some("feature".to_string()),
creation_ref: Some("123456".to_string()), creation_ref: Some("123456".to_string()),
creation_tree: None,
current_selector: Some("feature".to_string()), current_selector: Some("feature".to_string()),
current_ref: Some("123456".to_string()), current_ref: Some("123456".to_string()),
current_tree: None,
observed_at_epoch_seconds: None,
materialization_status: "present".to_string(), materialization_status: "present".to_string(),
cleanliness: "unknown".to_string(), cleanliness: "unknown".to_string(),
created_at: "3".to_string(), created_at: "3".to_string(),
@@ -13240,7 +13369,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test] #[tokio::test]
async fn account_and_login_records_round_trip() { async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 46); assert_eq!(store.schema_version().await.unwrap(), 47);
let now = "2026-07-22T00:00:00Z".to_string(); let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord { let account = AccountRecord {
account_id: "acct-user-alice".to_string(), account_id: "acct-user-alice".to_string(),
@@ -4,13 +4,31 @@ use sha2::{Digest, Sha256};
use crate::store::WorkdirCreateOperationRecord; use crate::store::WorkdirCreateOperationRecord;
use crate::{Error, Result, SqliteWorkspaceStore}; use crate::{Error, Result, SqliteWorkspaceStore};
pub fn selector_for_retry(
explicit_selector: Option<&str>,
persisted_selector: Option<&str>,
current_default_selector: Option<&str>,
) -> Option<String> {
explicit_selector
.or(persisted_selector)
.or(current_default_selector)
.map(str::to_string)
}
pub fn request_fingerprint( pub fn request_fingerprint(
repository_id: &str, repository_id: &str,
selector: Option<&str>, selector: Option<&str>,
requested_runtime_id: Option<&str>, requested_runtime_id: Option<&str>,
repository_source_fingerprint: &str,
repository_source_revision: u64,
) -> String { ) -> String {
let mut hasher = Sha256::new(); let mut hasher = Sha256::new();
for value in [Some(repository_id), selector, requested_runtime_id] { for value in [
Some(repository_id),
selector,
requested_runtime_id,
Some(repository_source_fingerprint),
] {
match value { match value {
Some(value) => { Some(value) => {
hasher.update([1]); hasher.update([1]);
@@ -20,6 +38,7 @@ pub fn request_fingerprint(
None => hasher.update([0]), None => hasher.update([0]),
} }
} }
hasher.update(repository_source_revision.to_be_bytes());
let digest = hasher.finalize(); let digest = hasher.finalize();
let mut encoded = String::with_capacity(digest.len() * 2); let mut encoded = String::with_capacity(digest.len() * 2);
for byte in digest { for byte in digest {
@@ -40,9 +59,10 @@ impl SqliteWorkspaceStore {
r#"INSERT OR IGNORE INTO workdir_create_operations ( r#"INSERT OR IGNORE INTO workdir_create_operations (
workspace_id, operation_id, request_fingerprint, repository_id, selector, workspace_id, operation_id, request_fingerprint, repository_id, selector,
requested_runtime_id, resolved_runtime_id, config_revision, requested_runtime_id, resolved_runtime_id, config_revision,
config_projection_digest, working_directory_id, state, failure, config_projection_digest, source_kind, source_uri, source_revision,
source_fingerprint, working_directory_id, state, failure,
created_at, updated_at created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#, ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17, ?18)"#,
params![ params![
record.workspace_id, record.workspace_id,
record.operation_id, record.operation_id,
@@ -53,6 +73,10 @@ impl SqliteWorkspaceStore {
record.resolved_runtime_id, record.resolved_runtime_id,
record.config_revision as i64, record.config_revision as i64,
record.config_projection_digest, record.config_projection_digest,
record.source_kind,
record.source_uri,
record.source_revision.map(|revision| revision as i64),
record.source_fingerprint,
record.working_directory_id, record.working_directory_id,
record.state, record.state,
record.failure, record.failure,
@@ -79,6 +103,81 @@ impl SqliteWorkspaceStore {
}) })
} }
pub fn bind_workdir_create_repository_access(
&self,
workspace_id: &str,
operation_id: &str,
request_fingerprint: &str,
credential_id: &str,
credential_revision: u64,
host_trust_id: &str,
host_trust_revision: u64,
repository_access_mode: &str,
cache_generation: u64,
now: &str,
) -> Result<WorkdirCreateOperationRecord> {
self.with_conn_mut(|conn| {
let operation = read_workdir_create_operation(conn, workspace_id, operation_id)?
.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Workdir create operation `{operation_id}` disappeared before Repository access binding"
))
})?;
if operation.request_fingerprint != request_fingerprint {
return Err(Error::InvalidInput(format!(
"Workdir create operation `{operation_id}` was reused with different input"
)));
}
if let Some(existing) = operation.credential_id.as_deref() {
if existing != credential_id
|| operation.credential_revision != Some(credential_revision)
|| operation.host_trust_id.as_deref() != Some(host_trust_id)
|| operation.host_trust_revision != Some(host_trust_revision)
|| operation.repository_access_mode.as_deref()
!= Some(repository_access_mode)
|| operation.cache_generation != cache_generation
{
return Err(Error::InvalidInput(format!(
"Workdir create operation `{operation_id}` Repository access evidence changed"
)));
}
return Ok(operation);
}
conn.execute(
r#"UPDATE workdir_create_operations
SET credential_id = ?4, credential_revision = ?5,
host_trust_id = ?6, host_trust_revision = ?7,
repository_access_mode = ?8, cache_generation = ?9,
updated_at = ?10
WHERE workspace_id = ?1 AND operation_id = ?2
AND request_fingerprint = ?3 AND credential_id IS NULL"#,
params![
workspace_id,
operation_id,
request_fingerprint,
credential_id,
i64::try_from(credential_revision).map_err(|_| Error::InvalidInput(
"credential revision is out of range".to_string()
))?,
host_trust_id,
i64::try_from(host_trust_revision).map_err(|_| Error::InvalidInput(
"host-trust revision is out of range".to_string()
))?,
repository_access_mode,
i64::try_from(cache_generation).map_err(|_| Error::InvalidInput(
"cache generation is out of range".to_string()
))?,
now,
],
)?;
read_workdir_create_operation(conn, workspace_id, operation_id)?.ok_or_else(|| {
Error::RegistryInconsistency(format!(
"Workdir create operation `{operation_id}` disappeared after Repository access binding"
))
})
})
}
pub fn finish_workdir_create_operation( pub fn finish_workdir_create_operation(
&self, &self,
workspace_id: &str, workspace_id: &str,
@@ -133,7 +232,10 @@ fn read_workdir_create_operation(
conn.query_row( conn.query_row(
r#"SELECT workspace_id, operation_id, request_fingerprint, repository_id, selector, r#"SELECT workspace_id, operation_id, request_fingerprint, repository_id, selector,
requested_runtime_id, resolved_runtime_id, config_revision, requested_runtime_id, resolved_runtime_id, config_revision,
config_projection_digest, working_directory_id, state, failure, config_projection_digest, source_kind, source_uri, source_revision,
source_fingerprint, credential_id, credential_revision,
host_trust_id, host_trust_revision, repository_access_mode,
cache_generation, working_directory_id, state, failure,
created_at, updated_at created_at, updated_at
FROM workdir_create_operations FROM workdir_create_operations
WHERE workspace_id = ?1 AND operation_id = ?2"#, WHERE workspace_id = ?1 AND operation_id = ?2"#,
@@ -149,11 +251,21 @@ fn read_workdir_create_operation(
resolved_runtime_id: row.get(6)?, resolved_runtime_id: row.get(6)?,
config_revision: row.get::<_, i64>(7)? as u64, config_revision: row.get::<_, i64>(7)? as u64,
config_projection_digest: row.get(8)?, config_projection_digest: row.get(8)?,
working_directory_id: row.get(9)?, source_kind: row.get(9)?,
state: row.get(10)?, source_uri: row.get(10)?,
failure: row.get(11)?, source_revision: row.get::<_, Option<i64>>(11)?.map(|value| value as u64),
created_at: row.get(12)?, source_fingerprint: row.get(12)?,
updated_at: row.get(13)?, credential_id: row.get(13)?,
credential_revision: row.get::<_, Option<i64>>(14)?.map(|value| value as u64),
host_trust_id: row.get(15)?,
host_trust_revision: row.get::<_, Option<i64>>(16)?.map(|value| value as u64),
repository_access_mode: row.get(17)?,
cache_generation: row.get::<_, i64>(18)? as u64,
working_directory_id: row.get(19)?,
state: row.get(20)?,
failure: row.get(21)?,
created_at: row.get(22)?,
updated_at: row.get(23)?,
}) })
}, },
) )
@@ -166,6 +278,22 @@ mod tests {
use super::*; use super::*;
use crate::store::{ControlPlaneStore, RepositoryRecord, WorkspaceRecord}; use crate::store::{ControlPlaneStore, RepositoryRecord, WorkspaceRecord};
#[test]
fn retry_selector_keeps_persisted_default_but_honors_explicit_input() {
assert_eq!(
selector_for_retry(None, Some("develop"), Some("main")),
Some("develop".to_string())
);
assert_eq!(
selector_for_retry(Some("release"), Some("develop"), Some("main")),
Some("release".to_string())
);
assert_eq!(
selector_for_retry(None, None, Some("main")),
Some("main".to_string())
);
}
#[test] #[test]
fn retry_keeps_resolved_config_evidence_and_rejects_changed_input() { fn retry_keeps_resolved_config_evidence_and_rejects_changed_input() {
let store = SqliteWorkspaceStore::in_memory().unwrap(); let store = SqliteWorkspaceStore::in_memory().unwrap();
@@ -201,13 +329,29 @@ mod tests {
let record = WorkdirCreateOperationRecord { let record = WorkdirCreateOperationRecord {
workspace_id: "workspace".to_string(), workspace_id: "workspace".to_string(),
operation_id: "call-1".to_string(), operation_id: "call-1".to_string(),
request_fingerprint: request_fingerprint("main", Some("develop"), None), request_fingerprint: request_fingerprint(
"main",
Some("develop"),
None,
"sha256:test",
1,
),
repository_id: "main".to_string(), repository_id: "main".to_string(),
selector: Some("develop".to_string()), selector: Some("develop".to_string()),
requested_runtime_id: None, requested_runtime_id: None,
resolved_runtime_id: "arcadia".to_string(), resolved_runtime_id: "arcadia".to_string(),
config_revision: 7, config_revision: 7,
config_projection_digest: "sha256:projection".to_string(), config_projection_digest: "sha256:projection".to_string(),
source_kind: Some("local_path".to_string()),
source_uri: Some("/tmp/repo".to_string()),
source_revision: Some(1),
source_fingerprint: Some("sha256:source".to_string()),
credential_id: None,
credential_revision: None,
host_trust_id: None,
host_trust_revision: None,
repository_access_mode: None,
cache_generation: 0,
working_directory_id: "wd-1".to_string(), working_directory_id: "wd-1".to_string(),
state: "pending".to_string(), state: "pending".to_string(),
failure: None, failure: None,
@@ -218,23 +362,59 @@ mod tests {
store.reserve_workdir_create_operation(&record).unwrap(), store.reserve_workdir_create_operation(&record).unwrap(),
record record
); );
let bound = store
.bind_workdir_create_repository_access(
"workspace",
"call-1",
&record.request_fingerprint,
"credential-1",
3,
"trust-1",
5,
"read_only",
2,
"2026-08-24T00:00:01Z",
)
.unwrap();
assert_eq!(bound.credential_id.as_deref(), Some("credential-1"));
assert_eq!(bound.credential_revision, Some(3));
assert_eq!(bound.host_trust_revision, Some(5));
assert_eq!(bound.cache_generation, 2);
assert!(
store
.bind_workdir_create_repository_access(
"workspace",
"call-1",
&record.request_fingerprint,
"credential-1",
4,
"trust-1",
5,
"read_only",
2,
"2026-08-24T00:00:02Z",
)
.is_err()
);
let mut changed_resolution = record.clone(); let mut changed_resolution = record.clone();
changed_resolution.resolved_runtime_id = "other".to_string(); changed_resolution.resolved_runtime_id = "other".to_string();
changed_resolution.config_revision = 8; changed_resolution.config_revision = 8;
assert_eq!( changed_resolution.source_uri = Some("ssh://git@other.test/repo.git".to_string());
store changed_resolution.source_revision = Some(9);
.reserve_workdir_create_operation(&changed_resolution) let replayed = store
.unwrap(), .reserve_workdir_create_operation(&changed_resolution)
record .unwrap();
); assert_eq!(replayed, bound);
assert_eq!(replayed.source_uri.as_deref(), Some("/tmp/repo"));
assert_eq!( assert_eq!(
store store
.load_workdir_create_operation("workspace", "call-1") .load_workdir_create_operation("workspace", "call-1")
.unwrap(), .unwrap(),
Some(record.clone()) Some(bound.clone())
); );
let mut changed_input = record.clone(); let mut changed_input = record.clone();
changed_input.request_fingerprint = request_fingerprint("main", Some("main"), None); changed_input.request_fingerprint =
request_fingerprint("main", Some("main"), None, "sha256:test", 1);
assert!( assert!(
store store
.reserve_workdir_create_operation(&changed_input) .reserve_workdir_create_operation(&changed_input)