feat: materialize repositories through runtime Git cache
This commit is contained in:
Generated
+2
@@ -6652,6 +6652,7 @@ dependencies = [
|
||||
"workdir",
|
||||
"worker",
|
||||
"workspace-api",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
@@ -6798,6 +6799,7 @@ dependencies = [
|
||||
"worker",
|
||||
"worker-runtime",
|
||||
"workspace-api",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -125,4 +125,5 @@ toml = "1.1"
|
||||
tracing = "0.1"
|
||||
url = "2.5"
|
||||
uuid = "1.23"
|
||||
zeroize = "1"
|
||||
webauthn-rs = { version = "0.5.2", features = ["danger-allow-state-serialisation", "danger-credential-internals"] }
|
||||
|
||||
@@ -28,7 +28,9 @@ pub use fs_operation::{
|
||||
GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult,
|
||||
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::*;
|
||||
|
||||
/// Persistent, opaque identity of one materialized Workdir.
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
//! `LocalWorkdirSession` is cheap to clone (`Arc` inside). Tool-specific session
|
||||
//! 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)]
|
||||
use std::io::Write as _;
|
||||
use std::io::{Read as _, Seek as _, SeekFrom};
|
||||
@@ -228,6 +229,8 @@ struct LocalWorkdirSessionInner {
|
||||
next_command_id: AtomicU64,
|
||||
commands: Mutex<HashMap<String, LocalCommand>>,
|
||||
command_telemetry: CommandTelemetry,
|
||||
command_environment: BTreeMap<String, String>,
|
||||
resources: StdMutex<Vec<Arc<dyn WorkdirSessionResource>>>,
|
||||
}
|
||||
|
||||
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).
|
||||
///
|
||||
/// The wrapped [`SharedScope`] is shared with every clone of this
|
||||
@@ -318,6 +324,26 @@ impl LocalWorkdirSession {
|
||||
cwd: PathBuf,
|
||||
scope: SharedScope,
|
||||
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 {
|
||||
inner: Arc::new(LocalWorkdirSessionInner {
|
||||
@@ -331,6 +357,8 @@ impl LocalWorkdirSession {
|
||||
next_command_id: AtomicU64::new(1),
|
||||
commands: Mutex::new(HashMap::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 command_id = handle.0.clone();
|
||||
let telemetry = self.inner.command_telemetry.clone();
|
||||
let command_environment = self.inner.command_environment.clone();
|
||||
let (cancel, cancel_rx) = watch::channel(false);
|
||||
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);
|
||||
output
|
||||
});
|
||||
@@ -840,6 +877,9 @@ impl WorkdirSession for LocalWorkdirSession {
|
||||
LocalCommand::Completed(_) => {}
|
||||
}
|
||||
}
|
||||
if let Ok(mut resources) = self.inner.resources.lock() {
|
||||
resources.clear();
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -909,6 +949,7 @@ async fn run_command(
|
||||
request: CommandRequest,
|
||||
command_id: String,
|
||||
telemetry: CommandTelemetry,
|
||||
command_environment: BTreeMap<String, String>,
|
||||
mut cancel: watch::Receiver<bool>,
|
||||
) -> Result<CommandOutput, WorkdirError> {
|
||||
let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
|
||||
@@ -925,6 +966,7 @@ async fn run_command(
|
||||
.arg("-c")
|
||||
.arg(&request.command)
|
||||
.current_dir(&cwd)
|
||||
.envs(command_environment)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::from(stdout_file))
|
||||
.stderr(Stdio::from(stderr_file))
|
||||
@@ -2319,6 +2361,31 @@ mod tests {
|
||||
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]
|
||||
async fn provider_cancels_active_command() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
|
||||
@@ -30,6 +30,8 @@ impl RuntimeWorkerRef {
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum MaterializerKind {
|
||||
#[default]
|
||||
RuntimeGitCache,
|
||||
/// Legacy persisted value from the pre-cache local `git worktree` materializer.
|
||||
LocalGitWorktree,
|
||||
}
|
||||
|
||||
|
||||
@@ -43,6 +43,7 @@ tokio = { workspace = true, features = ["net", "rt", "sync", "time"] }
|
||||
toml.workspace = true
|
||||
url.workspace = true
|
||||
uuid = { workspace = true, features = ["v7"] }
|
||||
zeroize.workspace = true
|
||||
tower = { workspace = true, features = ["util"], optional = true }
|
||||
worker.workspace = true
|
||||
workspace-api = { path = "../workspace-api" }
|
||||
|
||||
@@ -97,6 +97,55 @@ pub use workdir::workspace::{
|
||||
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 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 private_key: SensitiveString,
|
||||
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 WorkingDirectoryRequest {
|
||||
pub repository: WorkingDirectoryRepository,
|
||||
@@ -106,6 +155,9 @@ pub struct WorkingDirectoryRequest {
|
||||
/// Backend can create canonical registry rows before materialization.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
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)]
|
||||
|
||||
@@ -527,9 +527,19 @@ async fn list_working_directories(
|
||||
|
||||
async fn create_working_directory(
|
||||
State(state): State<RuntimeHttpState>,
|
||||
Extension(auth): Extension<RuntimeAuthContext>,
|
||||
body: Result<Json<WorkingDirectoryRequest>, JsonRejection>,
|
||||
) -> RestResult<RuntimeHttpWorkingDirectoryResponse> {
|
||||
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
|
||||
.runtime
|
||||
.create_working_directory(request)
|
||||
|
||||
@@ -23,7 +23,7 @@ use worker_runtime::http_server::{
|
||||
RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection,
|
||||
};
|
||||
use worker_runtime::worker_backend::{ProfileRuntimeWorkerFactory, WorkerRuntimeExecutionBackend};
|
||||
use worker_runtime::working_directory::LocalGitWorktreeMaterializer;
|
||||
use worker_runtime::working_directory::RuntimeGitCacheMaterializer;
|
||||
use worker_runtime::{Runtime, RuntimeOptions};
|
||||
|
||||
fn main() -> ExitCode {
|
||||
@@ -192,7 +192,7 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
|
||||
let backend = Arc::new(
|
||||
WorkerRuntimeExecutionBackend::new(factory)
|
||||
.map_err(ProcessError::WorkerAdapter)?
|
||||
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
|
||||
.with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
|
||||
fs_paths.workdir_target.clone(),
|
||||
)),
|
||||
);
|
||||
|
||||
@@ -706,13 +706,17 @@ fn runtime_local_workdir_session(
|
||||
root: &Path,
|
||||
cwd: &Path,
|
||||
scope: manifest::SharedScope,
|
||||
command_environment: std::collections::BTreeMap<String, String>,
|
||||
resources: Vec<Arc<dyn workdir::WorkdirSessionResource>>,
|
||||
) -> WorkdirSessionHandle {
|
||||
Arc::new(LocalWorkdirSession::materialized_bound(
|
||||
Arc::new(LocalWorkdirSession::materialized_bound_with_environment(
|
||||
Workdir::new(workdir_id),
|
||||
root.to_path_buf(),
|
||||
cwd.to_path_buf(),
|
||||
scope,
|
||||
WorkdirSessionCapabilities::ALL,
|
||||
command_environment,
|
||||
resources,
|
||||
))
|
||||
}
|
||||
|
||||
@@ -893,6 +897,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
binding.root(),
|
||||
binding.cwd(),
|
||||
worker.scope().clone(),
|
||||
binding.command_environment(),
|
||||
binding.session_resources(),
|
||||
)));
|
||||
} else {
|
||||
worker.bind_workdir_session(None);
|
||||
@@ -1071,6 +1077,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
|
||||
binding.root(),
|
||||
binding.cwd(),
|
||||
worker.scope().clone(),
|
||||
binding.command_environment(),
|
||||
binding.session_resources(),
|
||||
)));
|
||||
} else {
|
||||
worker.bind_workdir_session(None);
|
||||
@@ -1624,6 +1632,8 @@ where
|
||||
binding.root(),
|
||||
binding.cwd(),
|
||||
manifest::SharedScope::new(scope),
|
||||
binding.command_environment(),
|
||||
binding.session_resources(),
|
||||
))
|
||||
}
|
||||
|
||||
@@ -2142,7 +2152,7 @@ mod tests {
|
||||
use crate::identity::WorkerRef;
|
||||
use crate::management::RuntimeOptions;
|
||||
use crate::observation::WorkerObservationCursor;
|
||||
use crate::working_directory::LocalGitWorktreeMaterializer;
|
||||
use crate::working_directory::RuntimeGitCacheMaterializer;
|
||||
use agen::Engine;
|
||||
use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
|
||||
use agen::llm_client::{ClientError, LlmClient, Request};
|
||||
@@ -2728,8 +2738,9 @@ mod tests {
|
||||
source_fingerprint: "sha256:test".to_string(),
|
||||
selector: Some(RepositorySelector::from("HEAD")),
|
||||
},
|
||||
materializer: MaterializerKind::LocalGitWorktree,
|
||||
materializer: MaterializerKind::RuntimeGitCache,
|
||||
backend_workdir_id: None,
|
||||
materialization: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2853,12 +2864,16 @@ mod tests {
|
||||
root.path(),
|
||||
root.path(),
|
||||
manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
|
||||
Default::default(),
|
||||
Vec::new(),
|
||||
);
|
||||
let restored = runtime_local_workdir_session(
|
||||
"working-directory-42",
|
||||
root.path(),
|
||||
root.path(),
|
||||
manifest::SharedScope::new(Scope::writable(root.path()).unwrap()),
|
||||
Default::default(),
|
||||
Vec::new(),
|
||||
);
|
||||
|
||||
assert_eq!(spawned.workdir().id().as_str(), "working-directory-42");
|
||||
@@ -3330,7 +3345,7 @@ mod tests {
|
||||
};
|
||||
let backend = WorkerRuntimeExecutionBackend::new(factory)
|
||||
.unwrap()
|
||||
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
|
||||
.with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
|
||||
runtime_base.path(),
|
||||
));
|
||||
let runtime =
|
||||
@@ -3485,7 +3500,7 @@ mod tests {
|
||||
};
|
||||
let backend = WorkerRuntimeExecutionBackend::new(factory)
|
||||
.unwrap()
|
||||
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
|
||||
.with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
|
||||
runtime_base.path(),
|
||||
));
|
||||
let runtime =
|
||||
@@ -3524,7 +3539,7 @@ mod tests {
|
||||
let repo = create_clean_repo();
|
||||
let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
|
||||
.unwrap()
|
||||
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
|
||||
.with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
|
||||
runtime_base.path(),
|
||||
));
|
||||
let runtime =
|
||||
@@ -3560,7 +3575,7 @@ mod tests {
|
||||
let repo = create_clean_repo();
|
||||
let backend = WorkerRuntimeExecutionBackend::new(FailingFactory)
|
||||
.unwrap()
|
||||
.with_working_directory_materializer(LocalGitWorktreeMaterializer::new(
|
||||
.with_working_directory_materializer(RuntimeGitCacheMaterializer::new(
|
||||
runtime_base.path(),
|
||||
));
|
||||
let runtime =
|
||||
@@ -3574,9 +3589,15 @@ mod tests {
|
||||
|
||||
assert!(format!("{error:?}").contains("spawn failed"));
|
||||
let working_directories_root = runtime_base.path();
|
||||
let remaining_entries = fs::read_dir(working_directories_root)
|
||||
.map(|entries| entries.count())
|
||||
let remaining_workdirs = fs::read_dir(working_directories_root)
|
||||
.map(|entries| {
|
||||
entries
|
||||
.flatten()
|
||||
.filter(|entry| !entry.file_name().to_string_lossy().starts_with('.'))
|
||||
.count()
|
||||
})
|
||||
.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
@@ -48,6 +48,7 @@ tracing.workspace = true
|
||||
ts-rs = { version = "12.0.1", optional = true }
|
||||
url.workspace = true
|
||||
uuid = { workspace = true, features = ["v7"] }
|
||||
zeroize.workspace = true
|
||||
webauthn-rs = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
|
||||
@@ -11,7 +11,7 @@ use ring::rand::{SecureRandom, SystemRandom};
|
||||
use rusqlite::{OptionalExtension, TransactionBehavior, params};
|
||||
use serde::Deserialize;
|
||||
use sha2::{Digest, Sha256};
|
||||
use ssh_key::{Algorithm, HashAlg, PrivateKey, PublicKey};
|
||||
use ssh_key::{Algorithm, HashAlg, LineEnding, PrivateKey, PublicKey};
|
||||
use workspace_api::{
|
||||
CreateRepositorySshCredentialRequest, DeleteRepositorySshCredentialRequest,
|
||||
DeleteRepositorySshHostTrustRequest, PutRepositorySshHostTrustRequest, RepositoryAccessMode,
|
||||
@@ -59,7 +59,6 @@ impl WorkspaceConfigSchemaProvider for RepositoryAccessConfigSchemaProvider {
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct VirtualWorkspaceConfig {
|
||||
#[serde(default)]
|
||||
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)]
|
||||
pub struct RepositorySecretService {
|
||||
store: Arc<SqliteWorkspaceStore>,
|
||||
@@ -829,6 +838,144 @@ 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
|
||||
))
|
||||
})?;
|
||||
let (private_key, passphrase) = self.store.with_conn(|conn| {
|
||||
let private_key = read_sealed_secret(
|
||||
conn,
|
||||
workspace_id,
|
||||
&binding.credential_id,
|
||||
credential.current_revision,
|
||||
"private_key",
|
||||
)?
|
||||
.ok_or_else(|| {
|
||||
Error::RegistryInconsistency(format!(
|
||||
"Repository SSH credential `{}` is missing its private-key revision",
|
||||
binding.credential_id
|
||||
))
|
||||
})?;
|
||||
let passphrase = read_sealed_secret(
|
||||
conn,
|
||||
workspace_id,
|
||||
&binding.credential_id,
|
||||
credential.current_revision,
|
||||
"passphrase",
|
||||
)?;
|
||||
Ok((private_key, passphrase))
|
||||
})?;
|
||||
let private_key = self.unseal(
|
||||
workspace_id,
|
||||
&binding.credential_id,
|
||||
credential.current_revision,
|
||||
"private_key",
|
||||
private_key,
|
||||
)?;
|
||||
let passphrase = passphrase
|
||||
.map(|secret| {
|
||||
self.unseal(
|
||||
workspace_id,
|
||||
&binding.credential_id,
|
||||
credential.current_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 host_trust.port == 22 {
|
||||
host_trust.hostname.clone()
|
||||
} else {
|
||||
format!("[{}]:{}", host_trust.hostname, host_trust.port)
|
||||
};
|
||||
Ok(LeasedRepositorySshAccess {
|
||||
credential_id: binding.credential_id.clone(),
|
||||
credential_revision: credential.current_revision,
|
||||
host_trust_id: binding.host_trust_id.clone(),
|
||||
host_trust_revision: host_trust.current_revision,
|
||||
private_key,
|
||||
known_hosts_entry: format!("{host} {}\n", host_trust.host_key),
|
||||
})
|
||||
}
|
||||
|
||||
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(
|
||||
&self,
|
||||
workspace_id: &str,
|
||||
@@ -968,6 +1115,45 @@ fn insert_secret(
|
||||
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(
|
||||
tx: &rusqlite::Transaction<'_>,
|
||||
workspace_id: &str,
|
||||
@@ -1691,6 +1877,17 @@ mod tests {
|
||||
projection.bindings[0].access,
|
||||
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 unknown = config_state(
|
||||
r#"{
|
||||
|
||||
@@ -132,8 +132,10 @@ use crate::store::{
|
||||
use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest};
|
||||
use crate::{Error, Result};
|
||||
use worker_runtime::catalog::{
|
||||
ConfigBundleRef, ProfileSelector, RepositorySelector as RuntimeRepositorySelector,
|
||||
WorkingDirectoryClaim, WorkingDirectoryRepository, WorkingDirectoryRequest, WorkspaceApiRef,
|
||||
ConfigBundleRef, ProfileSelector, RepositoryMaterializationContext,
|
||||
RepositorySelector as RuntimeRepositorySelector, RepositorySshMaterializationAccess,
|
||||
SensitiveString, WorkingDirectoryClaim, WorkingDirectoryRepository, WorkingDirectoryRequest,
|
||||
WorkspaceApiRef,
|
||||
};
|
||||
use worker_runtime::config_bundle::ConfigBundle;
|
||||
use worker_runtime::http_server::{
|
||||
@@ -8453,6 +8455,8 @@ async fn create_workspace_working_directory(
|
||||
&request.repository_id,
|
||||
selector.as_deref(),
|
||||
requested_runtime_id.as_deref(),
|
||||
&working_directory_request.repository.source_fingerprint,
|
||||
working_directory_request.repository.source_revision,
|
||||
);
|
||||
let reserved = if let Some(existing) = api
|
||||
.config_store
|
||||
@@ -8596,6 +8600,24 @@ async fn create_workspace_working_directory(
|
||||
));
|
||||
}
|
||||
|
||||
let repository_access_projection = active_repository_access_projection(api, workspace_id)?;
|
||||
if let Err(error) = authorize_repository_materialization(
|
||||
api,
|
||||
&reserved.resolved_runtime_id,
|
||||
&operation_id,
|
||||
&repository_access_projection,
|
||||
&mut working_directory_request,
|
||||
) {
|
||||
api.config_store.finish_workdir_create_operation(
|
||||
workspace_id,
|
||||
&operation_id,
|
||||
&request_fingerprint,
|
||||
false,
|
||||
Some("working_directory_remote_repository_access_required"),
|
||||
&now_registry_timestamp(),
|
||||
)?;
|
||||
return Err(error);
|
||||
}
|
||||
working_directory_request.backend_workdir_id = Some(reserved.working_directory_id.clone());
|
||||
let existing = match api.runtime.working_directory(
|
||||
&reserved.resolved_runtime_id,
|
||||
@@ -10834,8 +10856,9 @@ fn working_directory_request_from_repository(
|
||||
})
|
||||
.or_else(|| Some(RuntimeRepositorySelector::from("HEAD"))),
|
||||
},
|
||||
materializer: MaterializerKind::LocalGitWorktree,
|
||||
materializer: MaterializerKind::RuntimeGitCache,
|
||||
backend_workdir_id: None,
|
||||
materialization: None,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -13917,6 +13940,51 @@ fn validate_working_directory_claim_for_browser(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn authorize_repository_materialization(
|
||||
api: &WorkspaceApi,
|
||||
runtime_id: &str,
|
||||
operation_id: &str,
|
||||
projection: &RepositoryAccessProjection,
|
||||
request: &mut WorkingDirectoryRequest,
|
||||
) -> ApiResult<()> {
|
||||
let ssh = if request.repository.source.kind == workspace_api::RepositorySourceKind::Ssh {
|
||||
let binding = projection
|
||||
.bindings
|
||||
.iter()
|
||||
.find(|binding| binding.repository_id == request.repository.id)
|
||||
.ok_or_else(|| {
|
||||
settings_bad_request(
|
||||
"working_directory_remote_repository_access_required",
|
||||
"SSH Repository has no active Workspace credential and host-trust binding",
|
||||
)
|
||||
})?;
|
||||
let lease = api
|
||||
.repository_secrets
|
||||
.lease_ssh_materialization_access(&api.config.workspace_id, binding)?;
|
||||
Some(RepositorySshMaterializationAccess {
|
||||
credential_id: lease.credential_id,
|
||||
credential_revision: lease.credential_revision,
|
||||
host_trust_id: lease.host_trust_id,
|
||||
host_trust_revision: lease.host_trust_revision,
|
||||
access: binding.access,
|
||||
private_key: SensitiveString::new(lease.private_key.as_str()),
|
||||
known_hosts_entry: SensitiveString::new(lease.known_hosts_entry),
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
request.materialization = Some(RepositoryMaterializationContext {
|
||||
workspace_id: api.config.workspace_id.clone(),
|
||||
runtime_id: runtime_id.to_string(),
|
||||
operation_id: operation_id.to_string(),
|
||||
config_revision: projection.config_revision,
|
||||
config_projection_digest: projection.projection_digest.clone(),
|
||||
cache_generation: 0,
|
||||
ssh,
|
||||
});
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn working_directory_request_for_browser(
|
||||
api: &WorkspaceApi,
|
||||
request: BrowserWorkingDirectoryCreateRequest,
|
||||
@@ -13935,8 +14003,9 @@ fn working_directory_request_for_browser(
|
||||
source_fingerprint: repository.source_fingerprint.clone(),
|
||||
selector: selector.map(RuntimeRepositorySelector),
|
||||
},
|
||||
materializer: MaterializerKind::LocalGitWorktree,
|
||||
materializer: MaterializerKind::RuntimeGitCache,
|
||||
backend_workdir_id: None,
|
||||
materialization: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -16015,7 +16084,7 @@ mod tests {
|
||||
worker_runtime::execution::WorkerExecutionContext,
|
||||
>,
|
||||
>,
|
||||
materializer: worker_runtime::working_directory::LocalGitWorktreeMaterializer,
|
||||
materializer: worker_runtime::working_directory::RuntimeGitCacheMaterializer,
|
||||
spawn_failure: std::sync::Mutex<Option<String>>,
|
||||
input_failure: std::sync::Mutex<Option<String>>,
|
||||
inputs: std::sync::Mutex<Vec<(worker_runtime::identity::WorkerRef, String)>>,
|
||||
@@ -16035,7 +16104,7 @@ mod tests {
|
||||
);
|
||||
Self {
|
||||
contexts: std::sync::Mutex::new(std::collections::HashMap::new()),
|
||||
materializer: worker_runtime::working_directory::LocalGitWorktreeMaterializer::new(
|
||||
materializer: worker_runtime::working_directory::RuntimeGitCacheMaterializer::new(
|
||||
std::env::temp_dir().join(unique),
|
||||
),
|
||||
spawn_failure: std::sync::Mutex::new(None),
|
||||
@@ -21083,10 +21152,15 @@ mod tests {
|
||||
init_clean_git_workspace(dir.path());
|
||||
let api = test_api(dir.path()).await;
|
||||
let operation_id = "provider-rejection-classification";
|
||||
let repository = api
|
||||
.require_configured_workspace_repository(TEST_REPOSITORY_ID)
|
||||
.unwrap();
|
||||
let request_fingerprint = crate::workdir_create_operations::request_fingerprint(
|
||||
TEST_REPOSITORY_ID,
|
||||
Some("HEAD"),
|
||||
Some(EMBEDDED_WORKER_RUNTIME_ID),
|
||||
&repository.source_fingerprint,
|
||||
repository.source_revision,
|
||||
);
|
||||
api.config_store
|
||||
.reserve_workdir_create_operation(&WorkdirCreateOperationRecord {
|
||||
|
||||
@@ -8,9 +8,16 @@ pub fn request_fingerprint(
|
||||
repository_id: &str,
|
||||
selector: Option<&str>,
|
||||
requested_runtime_id: Option<&str>,
|
||||
repository_source_fingerprint: &str,
|
||||
repository_source_revision: u64,
|
||||
) -> String {
|
||||
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 {
|
||||
Some(value) => {
|
||||
hasher.update([1]);
|
||||
@@ -20,6 +27,7 @@ pub fn request_fingerprint(
|
||||
None => hasher.update([0]),
|
||||
}
|
||||
}
|
||||
hasher.update(repository_source_revision.to_be_bytes());
|
||||
let digest = hasher.finalize();
|
||||
let mut encoded = String::with_capacity(digest.len() * 2);
|
||||
for byte in digest {
|
||||
@@ -201,7 +209,13 @@ mod tests {
|
||||
let record = WorkdirCreateOperationRecord {
|
||||
workspace_id: "workspace".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(),
|
||||
selector: Some("develop".to_string()),
|
||||
requested_runtime_id: None,
|
||||
@@ -234,7 +248,8 @@ mod tests {
|
||||
Some(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!(
|
||||
store
|
||||
.reserve_workdir_create_operation(&changed_input)
|
||||
|
||||
Reference in New Issue
Block a user