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

fix: bind repository SSH secrets to one-shot resources

Browse Source
This commit is contained in:
Hare
2026-08-26 18:28:22 +09:00
parent b644971d45
commit 4b132a21e9
9 changed files with 699 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+12
View File
@@ -116,6 +116,12 @@ impl Drop for SensitiveString {
}
}
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]")
@@ -130,7 +136,13 @@ pub struct RepositorySshMaterializationAccess {
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,
}
crates/worker-runtime/src/http_server.rs
+2 -1
View File
@@ -538,7 +538,8 @@ async fn authorize_working_directory_repository_access(
}
state
.runtime
.authorize_working_directory_repository_access(request)
.authorize_working_directory_repository_access_from_resource(request)
.await
.map_err(RuntimeHttpRestError::runtime)?;
Ok(Json(RuntimeHttpRepositoryAccessResponse {
authorized: true,
crates/worker-runtime/src/main.rs
+21 -8
View File
@@ -182,6 +182,9 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
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() {
let identity = runtime_auth.identity.as_ref().ok_or_else(|| {
ProcessError::Auth(
@@ -194,13 +197,15 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
.to_owned(),
));
};
factory = factory.with_resource_client(Arc::new(
let client = Arc::new(
worker_runtime::resource::HttpBackendResourceClient::new(
endpoint,
config.backend_resource_token.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(
WorkerRuntimeExecutionBackend::new(factory)
@@ -210,10 +215,10 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
)),
);
match &config.http.store {
let runtime = match &config.http.store {
RuntimeHttpStoreSelection::Memory => {
Runtime::with_execution_backend(runtime_options_from_http(&config.http), backend)
.map_err(ProcessError::Runtime)
.map_err(ProcessError::Runtime)?
}
RuntimeHttpStoreSelection::Fs { root } => {
let mut options = FsRuntimeStoreOptions::new(root.clone()).with_runtime_id(
@@ -226,12 +231,20 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
);
options.display_name = config.http.display_name.clone();
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 {
crates/worker-runtime/src/resource.rs
+56 -5
View File
@@ -11,18 +11,46 @@ use std::sync::Mutex;
pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str =
"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_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)]
#[serde(rename_all = "snake_case")]
pub enum BackendResourceKind {
ProfileSourceArchive,
RepositorySshAccess,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BackendResourceOperation {
FetchArchive,
FetchOnce,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -66,7 +94,7 @@ pub struct BackendResourceFetchRequest {
pub audit_correlation_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct BackendResourceFetchResponse {
pub kind: BackendResourceKind,
pub resource_id: String,
@@ -76,6 +104,29 @@ pub struct BackendResourceFetchResponse {
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)]
#[serde(tag = "code", rename_all = "snake_case")]
pub enum BackendResourceError {
@@ -248,7 +299,7 @@ pub fn build_profile_source_archive_fetch_request(
pub fn profile_source_archive_from_response(
handle: &BackendResourceHandle,
response: BackendResourceFetchResponse,
mut response: BackendResourceFetchResponse,
) -> Result<ProfileSourceArchive, BackendResourceError> {
if handle.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 {
return Err(BackendResourceError::ContentTypeMismatch {
expected: handle.content_type.clone(),
actual: response.content_type,
actual: response.content_type.clone(),
});
}
let actual_bytes = response.bytes.len() as u64;
@@ -278,7 +329,7 @@ pub fn profile_source_archive_from_response(
return Err(BackendResourceError::DigestMismatch {
expected: handle.digest.clone(),
actual: if response.digest != handle.digest {
response.digest
response.digest.clone()
} else {
actual_digest
},
@@ -296,7 +347,7 @@ pub fn profile_source_archive_from_response(
}
})?,
},
content: response.bytes,
content: std::mem::take(&mut response.bytes),
})
}
crates/worker-runtime/src/runtime.rs
+221
View File
@@ -26,6 +26,10 @@ use crate::management::{
};
#[cfg(feature = "ws-server")]
use crate::observation::{WorkerObservationCursor, WorkerObservationEvent};
use crate::resource::{
BackendResourceClient, BackendResourceError, BackendResourceFetchRequest, BackendResourceKind,
REPOSITORY_SSH_ACCESS_CONTENT_TYPE, RepositorySshAccessSecret,
};
#[cfg(feature = "fs-store")]
use crate::retention::{
FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult,
@@ -172,6 +176,14 @@ impl 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.
///
/// The store is scoped by `options.root`; if the directory already exists,
@@ -385,6 +397,61 @@ impl Runtime {
.map_err(RuntimeError::from)
}
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())
})?;
if ssh.private_key.expose().is_empty() || ssh.known_hosts_entry.expose().is_empty() {
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));
}
self.authorize_working_directory_repository_access(request)
}
/// List Runtime-owned working directories through the attached execution backend.
pub fn list_working_directories(
&self,
@@ -1862,6 +1929,15 @@ struct SubscriptionSink {
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)]
struct RuntimeState {
display_name: Option<String>,
@@ -1873,6 +1949,7 @@ struct RuntimeState {
persistence: RuntimePersistence,
status: RuntimeStatus,
execution_backend: Option<WorkerExecutionBackendRef>,
backend_resource_client: Option<BackendResourceClientRef>,
#[cfg(feature = "fs-store")]
next_diagnostic_id: u64,
workers: BTreeMap<WorkerId, WorkerRecord>,
@@ -1900,6 +1977,7 @@ impl RuntimeState {
persistence: RuntimePersistence::Memory,
status: RuntimeStatus::Running,
execution_backend: None,
backend_resource_client: None,
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
@@ -1928,6 +2006,7 @@ impl RuntimeState {
persistence: RuntimePersistence::Fs(store),
status: RuntimeStatus::Running,
execution_backend: None,
backend_resource_client: None,
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
@@ -1979,6 +2058,7 @@ impl RuntimeState {
persistence: RuntimePersistence::Fs(store),
status: persisted.status,
execution_backend: None,
backend_resource_client: None,
next_diagnostic_id,
workers,
config_bundles: BTreeMap::new(),
@@ -2734,6 +2814,23 @@ 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()
@@ -2926,6 +3023,8 @@ mod tests {
WorkerExecutionBackend, WorkerExecutionContext, WorkerExecutionHandle,
WorkerExecutionRestoreRequest, WorkerExecutionRunState,
};
use crate::working_directory::WorkingDirectoryDiagnostic;
use async_trait::async_trait;
use std::collections::BTreeMap;
#[cfg(feature = "fs-store")]
use std::sync::atomic::{AtomicU64, Ordering};
@@ -3178,6 +3277,10 @@ mod tests {
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"),
}),
@@ -3207,6 +3310,94 @@ mod tests {
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();
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");
}
fn scoped_task_request(objective: &str, workspace_id: &str) -> CreateWorkerRequest {
let mut request = task_request(objective);
request.workspace_api = Some(WorkspaceApiRef {
@@ -3288,6 +3479,7 @@ mod tests {
config_bundles: Mutex<Vec<Option<ConfigBundle>>>,
contexts: Mutex<BTreeMap<WorkerId, WorkerExecutionContext>>,
dispatched_inputs: Mutex<Vec<WorkerInput>>,
repository_accesses: Mutex<Vec<WorkingDirectoryRepositoryAccessRequest>>,
preserve_commit_ack_submission_id: AtomicBool,
#[cfg(feature = "ws-server")]
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
@@ -3328,6 +3520,17 @@ mod tests {
"test-execution-backend"
}
fn authorize_working_directory_repository_access(
&self,
request: &WorkingDirectoryRepositoryAccessRequest,
) -> Result<(), WorkingDirectoryDiagnostic> {
self.repository_accesses
.lock()
.unwrap()
.push(request.clone());
Ok(())
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
self.run_generations
.lock()
@@ -3435,6 +3638,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 {
let runtime = Runtime::with_execution_backend(
RuntimeOptions::default(),
crates/worker-runtime/src/working_directory.rs
+98
View File
@@ -729,6 +729,12 @@ impl RuntimeGitCacheMaterializer {
)
})?;
validate_ssh_materialization_access(&access)?;
validate_repository_access_binding(
&access,
&request.repository.id,
&request.repository.source_fingerprint,
Some(request.repository.source.uri.as_str()),
)?;
request
.materialization
.as_mut()
@@ -912,6 +918,17 @@ impl WorkingDirectoryMaterializer for RuntimeGitCacheMaterializer {
"Workdir is not backed by an SSH Repository",
));
}
validate_repository_access_binding(
ssh,
&binding.working_directory.evidence.repository_id,
binding
.working_directory
.evidence
.repository_source_fingerprint
.as_deref()
.unwrap_or_default(),
None,
)?;
binding.working_directory.evidence.operation_id =
Some(request.materialization.operation_id.clone());
binding.working_directory.evidence.credential_revision = Some(ssh.credential_revision);
@@ -1730,6 +1747,24 @@ impl Drop for RepositoryCommandAccess {
}
}
fn validate_repository_access_binding(
access: &RepositorySshMaterializationAccess,
repository_id: &str,
repository_source_fingerprint: &str,
repository_uri: Option<&str>,
) -> Result<(), WorkingDirectoryDiagnostic> {
if access.repository_id != repository_id
|| access.repository_source_fingerprint != repository_source_fingerprint
|| repository_uri.is_some_and(|repository_uri| access.repository_uri != repository_uri)
{
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_repository_access_binding_mismatch",
"Repository SSH access authority does not match the requested Repository source",
));
}
Ok(())
}
fn validate_ssh_materialization_access(
access: &RepositorySshMaterializationAccess,
) -> Result<(), WorkingDirectoryDiagnostic> {
@@ -2286,6 +2321,28 @@ mod tests {
use crate::catalog::{RepositorySelector, WorkingDirectoryRepository};
use crate::identity::{WorkerId, WorkerRef};
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,
}
}
fn git(path: &Path, args: &[&str]) {
let status = Command::new("git")
.arg("-C")
@@ -2523,6 +2580,10 @@ mod tests {
host_trust_revision: 4,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repo-main".to_string(),
repository_source_fingerprint: "sha256:test".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
private_key: crate::catalog::SensitiveString::new("PRIVATE KEY secret bytes"),
known_hosts_entry: crate::catalog::SensitiveString::new("host key secret bytes"),
};
@@ -2560,6 +2621,10 @@ mod tests {
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadWrite,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repo-main".to_string(),
repository_source_fingerprint: "sha256:test".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
private_key: crate::catalog::SensitiveString::new(
fs::read_to_string(&key_path).unwrap(),
),
@@ -2609,6 +2674,31 @@ mod tests {
.ssh
.is_some()
);
let mut mismatched_source_request = authorized_ssh_request.clone();
mismatched_source_request.repository.source.uri =
"ssh://git@example.test/other.git".to_string();
assert_eq!(
materializer
.request_with_authorized_repository_access(
&working_directory_id,
&mismatched_source_request,
)
.unwrap_err()
.code,
"working_directory_repository_access_binding_mismatch"
);
let mut mismatched_repository_request = authorized_ssh_request.clone();
mismatched_repository_request.repository.id = "repo-other".to_string();
assert_eq!(
materializer
.request_with_authorized_repository_access(
&working_directory_id,
&mismatched_repository_request,
)
.unwrap_err()
.code,
"working_directory_repository_access_binding_mismatch"
);
let created = materializer.create(&request).unwrap();
let id = created.working_directory.id;
assert_eq!(id, working_directory_id);
@@ -2794,6 +2884,10 @@ mod tests {
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repo-main".to_string(),
repository_source_fingerprint: "sha256:test".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
private_key: crate::catalog::SensitiveString::new("PRIVATE KEY placeholder"),
known_hosts_entry: crate::catalog::SensitiveString::new(
"example.test ssh-ed25519 placeholder",
@@ -2884,6 +2978,10 @@ mod tests {
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
repository_id: "repo-main".to_string(),
repository_source_fingerprint: "sha256:test".to_string(),
repository_uri: "ssh://git@example.test/repo.git".to_string(),
secret_resource: repository_resource_handle(),
private_key: crate::catalog::SensitiveString::new("PRIVATE KEY placeholder"),
known_hosts_entry: crate::catalog::SensitiveString::new(
"other.test ssh-ed25519 placeholder",
crates/workspace-server/src/hosts.rs
+1 -1
View File
@@ -4500,7 +4500,7 @@ mod tests {
let handle = bundle.profile_source_archive_handle.as_ref().unwrap();
assert!(bundle.profile_source_archive.is_none());
let response = broker
.fetch_profile_source_archive(worker_runtime::resource::BackendResourceFetchRequest {
.fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
handle: handle.clone(),
runtime_id: runtime_id.to_string(),
worker_id: None,
crates/workspace-server/src/resource_broker.rs
+199 -33
View File
@@ -9,7 +9,8 @@ use worker_runtime::resource::{
BackendResourceClient, BackendResourceError, BackendResourceFetchRequest,
BackendResourceFetchResponse, BackendResourceHandle, BackendResourceKind,
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)]
@@ -29,7 +30,34 @@ struct StoredResource {
runtime_id: Option<String>,
worker: Option<RuntimeWorkerRef>,
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 {
@@ -73,7 +101,9 @@ impl BackendResourceBroker {
runtime_id,
worker,
handle: handle.clone(),
archive,
bytes: Vec::new(),
archive: Some(archive),
one_shot: false,
};
if let Ok(mut resources) = self.resources.lock() {
resources.insert(nonce, stored);
@@ -81,6 +111,84 @@ impl BackendResourceBroker {
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(
&self,
digest: &str,
@@ -90,20 +198,21 @@ impl BackendResourceBroker {
.ok()?
.values()
.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,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
verify_handle_shape(&request.handle)?;
let stored = self
let mut resources = self
.resources
.lock()
.map_err(|_| BackendResourceError::Transport {
message: "resource broker lock poisoned".to_string(),
})?
})?;
let mut stored = resources
.get(&request.handle.nonce)
.cloned()
.ok_or(BackendResourceError::MissingResource)?;
@@ -111,7 +220,7 @@ impl BackendResourceBroker {
if stored.handle.expires_at_unix_seconds < Utc::now().timestamp() {
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 {
return Err(BackendResourceError::Oversized {
max_bytes: stored.handle.max_bytes,
@@ -139,12 +248,15 @@ impl BackendResourceBroker {
});
}
}
if stored.one_shot {
resources.remove(&request.handle.nonce);
}
Ok(BackendResourceFetchResponse {
kind: BackendResourceKind::ProfileSourceArchive,
resource_id: stored.archive.reference.id,
digest: stored.archive.reference.digest,
content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
bytes: stored.archive.content,
kind: stored.handle.kind.clone(),
resource_id: stored.handle.resource_id.clone(),
digest: stored.handle.digest.clone(),
content_type: stored.handle.content_type.clone(),
bytes: stored.take_bytes(),
audit_correlation_id: request.audit_correlation_id,
})
}
@@ -156,24 +268,38 @@ impl BackendResourceClient for BackendResourceBroker {
&self,
request: BackendResourceFetchRequest,
) -> Result<BackendResourceFetchResponse, BackendResourceError> {
self.fetch_profile_source_archive(request)
self.fetch_resource(request)
}
}
fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendResourceError> {
if handle.kind != BackendResourceKind::ProfileSourceArchive {
return Err(BackendResourceError::UnsupportedKind);
}
if handle.operation != BackendResourceOperation::FetchArchive {
return Err(BackendResourceError::Unauthorized {
message: "resource handle operation is not fetch_archive".to_string(),
});
}
if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE {
return Err(BackendResourceError::ContentTypeMismatch {
expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
actual: handle.content_type.clone(),
});
match handle.kind {
BackendResourceKind::ProfileSourceArchive => {
if handle.operation != BackendResourceOperation::FetchArchive {
return Err(BackendResourceError::Unauthorized {
message: "resource handle operation is not fetch_archive".to_string(),
});
}
if handle.content_type != PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE {
return Err(BackendResourceError::ContentTypeMismatch {
expected: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
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(())
}
@@ -249,7 +375,7 @@ mod tests {
archive(),
);
let response = broker
.fetch_profile_source_archive(BackendResourceFetchRequest {
.fetch_resource(BackendResourceFetchRequest {
handle: handle.clone(),
runtime_id: runtime_id.to_string(),
worker_id: None,
@@ -260,6 +386,46 @@ mod tests {
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]
fn broker_rejects_runtime_mismatch() {
let broker = BackendResourceBroker::default();
@@ -270,7 +436,7 @@ mod tests {
archive(),
);
let err = broker
.fetch_profile_source_archive(request(handle, "runtime-b", None))
.fetch_resource(request(handle, "runtime-b", None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
@@ -287,7 +453,7 @@ mod tests {
archive(),
);
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();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
@@ -312,7 +478,7 @@ mod tests {
let mut extended = handle;
extended.expires_at_unix_seconds = 4_102_444_800;
let err = broker
.fetch_profile_source_archive(request(extended, &runtime_id, None))
.fetch_resource(request(extended, &runtime_id, None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Expired));
}
@@ -328,7 +494,7 @@ mod tests {
);
handle.scope_id = Some("tampered-scope".to_string());
let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, None))
.fetch_resource(request(handle, &runtime_id, None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
}
@@ -345,7 +511,7 @@ mod tests {
);
handle.max_bytes = DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1024;
let err = broker
.fetch_profile_source_archive(request(handle, &runtime_id, None))
.fetch_resource(request(handle, &runtime_id, None))
.unwrap_err();
assert!(matches!(err, BackendResourceError::Oversized { .. }));
}
crates/workspace-server/src/server.rs
+89 -13
View File
@@ -11504,7 +11504,7 @@ async fn scoped_post_internal_runtime_resource_fetch(
));
}
api.resource_broker
.fetch_profile_source_archive(request)
.fetch_resource(request)
.map(Json)
.map_err(|error| (backend_resource_error_status(&error), Json(error)))
}
@@ -14122,6 +14122,29 @@ fn authorize_repository_materialization_operation(
));
}
};
let expires_at_epoch_seconds = repository_access_expiry();
let secret_resource = api
.resource_broker
.issue_repository_ssh_access_handle(
&api.config.workspace_id,
&operation.resolved_runtime_id,
format!("repository-ssh-access:{}", operation.operation_id),
format!(
"credential:{}:host-trust:{}",
lease.credential_revision, lease.host_trust_revision
),
i64::try_from(expires_at_epoch_seconds).unwrap_or(i64::MAX),
worker_runtime::resource::RepositorySshAccessSecret {
private_key: lease.private_key.as_str().to_string(),
known_hosts_entry: lease.known_hosts_entry.clone(),
},
)
.map_err(|_| {
settings_bad_request(
"working_directory_repository_access_resource_failed",
"Repository SSH access resource could not be issued",
)
})?;
RepositoryMaterializationContext {
workspace_id: api.config.workspace_id.clone(),
runtime_id: operation.resolved_runtime_id.clone(),
@@ -14135,9 +14158,13 @@ fn authorize_repository_materialization_operation(
host_trust_id: lease.host_trust_id,
host_trust_revision: lease.host_trust_revision,
access,
expires_at_epoch_seconds: repository_access_expiry(),
private_key: SensitiveString::new(lease.private_key.as_str()),
known_hosts_entry: SensitiveString::new(lease.known_hosts_entry),
expires_at_epoch_seconds,
repository_id: request.repository.id.clone(),
repository_source_fingerprint: request.repository.source_fingerprint.clone(),
repository_uri: request.repository.source.uri.clone(),
secret_resource,
private_key: SensitiveString::default(),
known_hosts_entry: SensitiveString::default(),
}),
}
} else {
@@ -14230,15 +14257,42 @@ fn authorize_repository_materialization(
let lease = api
.repository_secrets
.lease_ssh_materialization_access(&api.config.workspace_id, binding)?;
let expires_at_epoch_seconds = repository_access_expiry();
let secret_resource = api
.resource_broker
.issue_repository_ssh_access_handle(
&api.config.workspace_id,
runtime_id,
format!("repository-ssh-access:{operation_id}"),
format!(
"credential:{}:host-trust:{}",
lease.credential_revision, lease.host_trust_revision
),
i64::try_from(expires_at_epoch_seconds).unwrap_or(i64::MAX),
worker_runtime::resource::RepositorySshAccessSecret {
private_key: lease.private_key.as_str().to_string(),
known_hosts_entry: lease.known_hosts_entry.clone(),
},
)
.map_err(|_| {
settings_bad_request(
"working_directory_repository_access_resource_failed",
"Repository SSH access resource could not be issued",
)
})?;
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,
expires_at_epoch_seconds: repository_access_expiry(),
private_key: SensitiveString::new(lease.private_key.as_str()),
known_hosts_entry: SensitiveString::new(lease.known_hosts_entry),
expires_at_epoch_seconds,
repository_id: request.repository.id.clone(),
repository_source_fingerprint: request.repository.source_fingerprint.clone(),
repository_uri: request.repository.source.uri.clone(),
secret_resource,
private_key: SensitiveString::default(),
known_hosts_entry: SensitiveString::default(),
})
} else {
None
@@ -15423,6 +15477,20 @@ mod tests {
);
let mut repository_access_launch = workdir_flow_launch;
let secret_resource = api
.resource_broker
.issue_repository_ssh_access_handle(
&api.config.workspace_id,
"runtime-1",
"repository-access-test",
"1",
i64::MAX,
worker_runtime::resource::RepositorySshAccessSecret {
private_key: "private-key-bytes".to_string(),
known_hosts_entry: "known-hosts-entry".to_string(),
},
)
.unwrap();
let working_directory = repository_access_launch
.resolved_working_directory_request
.as_mut()
@@ -15444,12 +15512,15 @@ mod tests {
host_trust_revision: 1,
access: workspace_api::RepositoryAccessMode::ReadOnly,
expires_at_epoch_seconds: u64::MAX,
private_key: worker_runtime::catalog::SensitiveString::new(
"private-key-bytes",
),
known_hosts_entry: worker_runtime::catalog::SensitiveString::new(
"known-hosts-entry",
),
repository_id: working_directory.repository.id.clone(),
repository_source_fingerprint: working_directory
.repository
.source_fingerprint
.clone(),
repository_uri: working_directory.repository.source.uri.clone(),
secret_resource,
private_key: worker_runtime::catalog::SensitiveString::default(),
known_hosts_entry: worker_runtime::catalog::SensitiveString::default(),
},
),
});
@@ -15460,6 +15531,11 @@ mod tests {
assert_eq!(access.working_directory_id, "working-directory-1");
assert!(access.materialization.ssh.is_some());
let serialized_access = serde_json::to_string(&access).unwrap();
assert!(!serialized_access.contains("private-key-bytes"));
assert!(!serialized_access.contains("known-hosts-entry"));
assert!(!serialized_access.contains("private_key"));
assert!(!serialized_access.contains("known_hosts_entry"));
assert!(
repository_access_launch
.resolved_working_directory_request