From 3c2664c3ce5a8571b0c27df3a23ecd0f1aa1aa3b Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 24 Aug 2026 13:32:49 +0900 Subject: [PATCH] feat: sign Runtime workspace requests --- crates/worker-runtime/src/auth.rs | 259 ++++++++++++++++++++ crates/worker-runtime/src/main.rs | 24 +- crates/worker-runtime/src/resource.rs | 57 ++++- crates/worker-runtime/src/worker_backend.rs | 121 +++++++-- crates/worker-runtime/src/worker_source.rs | 64 ++++- 5 files changed, 499 insertions(+), 26 deletions(-) diff --git a/crates/worker-runtime/src/auth.rs b/crates/worker-runtime/src/auth.rs index 15dc0fb2..bf02b35f 100644 --- a/crates/worker-runtime/src/auth.rs +++ b/crates/worker-runtime/src/auth.rs @@ -3,6 +3,7 @@ use base64::engine::general_purpose::URL_SAFE_NO_PAD; use ring::rand::{SecureRandom, SystemRandom}; use ring::signature::{ED25519, Ed25519KeyPair, KeyPair, UnparsedPublicKey}; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::fmt; use std::time::{SystemTime, UNIX_EPOCH}; @@ -14,6 +15,11 @@ pub const WORKER_MUTATION_SOURCE_PROOF_HEADER: &str = "x-yoi-worker-mutation-pro const WORKER_MUTATION_SOURCE_PROOF_PREFIX: &str = "yoi-worker-source-v1"; const WORKER_MUTATION_SOURCE_SIGNING_INPUT_PREFIX: &str = "yoi-worker-source-v1."; pub const WORKER_REMOVE_PERMISSION: &str = "workspace:worker-remove"; +pub const RUNTIME_REQUEST_SOURCE_PROOF_HEADER: &str = "x-yoi-runtime-request-proof"; +pub const WORKSPACE_REQUEST_PERMISSION: &str = "workspace:request"; +pub const BACKEND_RESOURCE_FETCH_PERMISSION: &str = "workspace:resource-fetch"; +const RUNTIME_REQUEST_SOURCE_PROOF_PREFIX: &str = "yoi-runtime-request-v1"; +const RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX: &str = "yoi-runtime-request-v1."; #[derive(Debug, thiserror::Error)] pub enum RuntimeAuthError { @@ -33,6 +39,10 @@ pub enum RuntimeAuthError { InvalidTokenFormat, #[error("malformed capability token claims: {0}")] MalformedClaims(#[from] serde_json::Error), + #[error("runtime request proof contains an invalid `{0}` claim")] + InvalidClaim(&'static str), + #[error("runtime request proof does not match the HTTP request")] + ClaimMismatch, #[error("unknown token issuer `{0}`")] UnknownIssuer(String), #[error("invalid token signature")] @@ -224,6 +234,162 @@ pub fn verify_capability_token( }) } +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeRequestSourceClaims { + pub iss: String, + pub aud: String, + pub workspace_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub worker_id: Option, + pub permission: String, + pub method: String, + pub path: String, + pub body_digest: String, + pub iat: i64, + pub exp: i64, + pub jti: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RuntimeRequestSourceSigner { + identity_id: String, + private_key: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RuntimeRequestSourceExpectation<'a> { + pub identity_id: &'a str, + pub audience: &'a str, + pub workspace_id: &'a str, + pub worker_id: Option<&'a str>, + pub permission: &'a str, + pub method: &'a str, + pub path: &'a str, + pub body_digest: &'a str, + pub now_unix: i64, +} + +pub fn request_body_digest(body: &[u8]) -> String { + URL_SAFE_NO_PAD.encode(Sha256::digest(body)) +} + +impl RuntimeRequestSourceSigner { + pub fn from_identity(identity: &RuntimeIdentityMaterial) -> Self { + Self { + identity_id: identity.identity_id.clone(), + private_key: identity.private_key.clone(), + } + } + + #[allow(clippy::too_many_arguments)] + pub fn issue( + &self, + audience: &str, + workspace_id: &str, + worker_id: Option<&str>, + permission: &str, + method: &str, + path: &str, + body: &[u8], + now_unix: i64, + ttl_seconds: u64, + ) -> Result { + for (name, value) in [ + ("audience", audience), + ("workspace_id", workspace_id), + ("permission", permission), + ("method", method), + ("path", path), + ] { + if value.trim().is_empty() { + return Err(RuntimeAuthError::InvalidClaim(name)); + } + } + if worker_id.is_some_and(str::is_empty) { + return Err(RuntimeAuthError::InvalidClaim("worker_id")); + } + let ttl_seconds = i64::try_from(ttl_seconds).unwrap_or(i64::MAX); + let claims = RuntimeRequestSourceClaims { + iss: self.identity_id.clone(), + aud: audience.to_owned(), + workspace_id: workspace_id.to_owned(), + worker_id: worker_id.map(str::to_owned), + permission: permission.to_owned(), + method: method.to_owned(), + path: path.to_owned(), + body_digest: request_body_digest(body), + iat: now_unix, + exp: now_unix.saturating_add(ttl_seconds), + jti: new_token_id()?, + }; + let payload = serde_json::to_vec(&claims)?; + let payload = URL_SAFE_NO_PAD.encode(payload); + let signing_input = format!("{RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX}{payload}"); + let private = decode_private_key(&self.private_key)?; + let key_pair = Ed25519KeyPair::from_pkcs8(&private) + .map_err(|_| RuntimeAuthError::InvalidPrivateKey)?; + let signature = URL_SAFE_NO_PAD.encode(key_pair.sign(signing_input.as_bytes()).as_ref()); + Ok(format!( + "{RUNTIME_REQUEST_SOURCE_PROOF_PREFIX}.{payload}.{signature}" + )) + } +} + +pub fn decode_runtime_request_source_claims( + proof: &str, +) -> Result { + let (prefix, payload, _signature) = split_runtime_request_source_proof(proof)?; + if prefix != RUNTIME_REQUEST_SOURCE_PROOF_PREFIX { + return Err(RuntimeAuthError::InvalidTokenFormat); + } + let payload = URL_SAFE_NO_PAD.decode(payload)?; + serde_json::from_slice(&payload).map_err(RuntimeAuthError::from) +} + +pub fn verify_runtime_request_source( + proof: &str, + public_key: &str, + expected: &RuntimeRequestSourceExpectation<'_>, +) -> Result { + let (prefix, payload, signature) = split_runtime_request_source_proof(proof)?; + if prefix != RUNTIME_REQUEST_SOURCE_PROOF_PREFIX { + return Err(RuntimeAuthError::InvalidTokenFormat); + } + let signature = URL_SAFE_NO_PAD.decode(signature)?; + let signing_input = format!("{RUNTIME_REQUEST_SOURCE_SIGNING_INPUT_PREFIX}{payload}"); + let public_key = decode_public_key(public_key)?; + UnparsedPublicKey::new(&ED25519, public_key) + .verify(signing_input.as_bytes(), &signature) + .map_err(|_| RuntimeAuthError::InvalidSignature)?; + let claims = decode_runtime_request_source_claims(proof)?; + if claims.iss != expected.identity_id + || claims.aud != expected.audience + || claims.workspace_id != expected.workspace_id + || claims.worker_id.as_deref() != expected.worker_id + || claims.permission != expected.permission + || claims.method != expected.method + || claims.path != expected.path + || claims.body_digest != expected.body_digest + { + return Err(RuntimeAuthError::ClaimMismatch); + } + if claims.iat > expected.now_unix || claims.exp < expected.now_unix { + return Err(RuntimeAuthError::Expired); + } + Ok(claims) +} + +fn split_runtime_request_source_proof(proof: &str) -> Result<(&str, &str, &str), RuntimeAuthError> { + let mut parts = proof.split('.'); + let prefix = parts.next().unwrap_or_default(); + let payload = parts.next().unwrap_or_default(); + let signature = parts.next().unwrap_or_default(); + if prefix.is_empty() || payload.is_empty() || signature.is_empty() || parts.next().is_some() { + return Err(RuntimeAuthError::InvalidTokenFormat); + } + Ok((prefix, payload, signature)) +} + #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct WorkerMutationSourceClaims { pub iss: String, @@ -592,6 +758,99 @@ mod tests { )); } + #[test] + fn runtime_request_source_proof_binds_request_and_rejects_spoofed_signature() { + let trusted = RuntimeIdentityMaterial::generate("runtime-main").unwrap(); + let signer = RuntimeRequestSourceSigner::from_identity(&trusted); + let body = br#"{"ticket":"T-1"}"#; + let proof = signer + .issue( + "server-main", + "workspace-a", + Some("worker-7"), + WORKSPACE_REQUEST_PERMISSION, + "POST", + "/api/w/workspace-a/tickets/comment", + body, + 90, + 10, + ) + .unwrap(); + let expected = RuntimeRequestSourceExpectation { + identity_id: "runtime-main", + audience: "server-main", + workspace_id: "workspace-a", + worker_id: Some("worker-7"), + permission: WORKSPACE_REQUEST_PERMISSION, + method: "POST", + path: "/api/w/workspace-a/tickets/comment", + body_digest: &request_body_digest(body), + now_unix: 99, + }; + let claims = verify_runtime_request_source(&proof, &trusted.public_key, &expected).unwrap(); + assert_eq!(claims.iss, "runtime-main"); + let changed_body = RuntimeRequestSourceExpectation { + body_digest: &request_body_digest(br#"{"ticket":"T-2"}"#), + ..expected.clone() + }; + assert!(matches!( + verify_runtime_request_source(&proof, &trusted.public_key, &changed_body), + Err(RuntimeAuthError::ClaimMismatch) + )); + let spoofed = RuntimeIdentityMaterial::generate("runtime-main").unwrap(); + assert!(matches!( + verify_runtime_request_source(&proof, &spoofed.public_key, &expected), + Err(RuntimeAuthError::InvalidSignature) + )); + } + + #[test] + fn runtime_request_source_proof_rejects_wrong_scope_and_expiry() { + let runtime = RuntimeIdentityMaterial::generate("runtime-main").unwrap(); + let proof = RuntimeRequestSourceSigner::from_identity(&runtime) + .issue( + "server-main", + "workspace-a", + None, + BACKEND_RESOURCE_FETCH_PERMISSION, + "POST", + "/api/runtime/v1/workspaces/workspace-a/resources/fetch", + b"{}", + 90, + 10, + ) + .unwrap(); + let digest = request_body_digest(b"{}"); + let expected = RuntimeRequestSourceExpectation { + identity_id: "runtime-main", + audience: "server-main", + workspace_id: "workspace-a", + worker_id: None, + permission: BACKEND_RESOURCE_FETCH_PERMISSION, + method: "POST", + path: "/api/runtime/v1/workspaces/workspace-a/resources/fetch", + body_digest: &digest, + now_unix: 99, + }; + assert!(verify_runtime_request_source(&proof, &runtime.public_key, &expected).is_ok()); + let wrong_workspace = RuntimeRequestSourceExpectation { + workspace_id: "workspace-b", + ..expected.clone() + }; + assert!(matches!( + verify_runtime_request_source(&proof, &runtime.public_key, &wrong_workspace), + Err(RuntimeAuthError::ClaimMismatch) + )); + let expired = RuntimeRequestSourceExpectation { + now_unix: 101, + ..expected + }; + assert!(matches!( + verify_runtime_request_source(&proof, &runtime.public_key, &expired), + Err(RuntimeAuthError::Expired) + )); + } + #[test] fn capability_token_verifies_signature_audience_expiry_and_permission() { let server = RuntimeIdentityMaterial::generate("server-main").unwrap(); diff --git a/crates/worker-runtime/src/main.rs b/crates/worker-runtime/src/main.rs index a8cb5fb9..3cd733f2 100644 --- a/crates/worker-runtime/src/main.rs +++ b/crates/worker-runtime/src/main.rs @@ -160,15 +160,33 @@ fn build_runtime(config: &ProcessConfig) -> Result { }; let mut factory = ProfileRuntimeWorkerFactory::new(fs_paths.worker_dir.join("worker-root")) .with_runtime_store_dir(runtime_store_dir); - if let Some(identity) = read_runtime_auth_file(&runtime_auth_path(config))?.identity { - factory = factory.with_remote_worker_mutation_identity(identity); + let runtime_auth = read_runtime_auth_file(&runtime_auth_path(config))?; + if let Some(identity) = runtime_auth.identity.clone() { + if let [trusted_server] = runtime_auth.trusted_servers.as_slice() { + factory = + factory.with_runtime_request_identity(identity, trusted_server.server_id.clone()); + } else { + factory = factory.with_remote_worker_mutation_identity(identity); + } } if let Some(endpoint) = config.backend_resource_endpoint.clone() { + let identity = runtime_auth.identity.as_ref().ok_or_else(|| { + ProcessError::Auth( + "--backend-resource-endpoint requires a configured Runtime identity".to_owned(), + ) + })?; + let [trusted_server] = runtime_auth.trusted_servers.as_slice() else { + return Err(ProcessError::Auth( + "--backend-resource-endpoint requires exactly one trusted Server identity" + .to_owned(), + )); + }; factory = factory.with_resource_client(Arc::new( worker_runtime::resource::HttpBackendResourceClient::new( endpoint, config.backend_resource_token.clone(), - ), + ) + .with_runtime_request_source(identity, trusted_server.server_id.clone()), )); } let backend = Arc::new( diff --git a/crates/worker-runtime/src/resource.rs b/crates/worker-runtime/src/resource.rs index 2813fe1a..60046fab 100644 --- a/crates/worker-runtime/src/resource.rs +++ b/crates/worker-runtime/src/resource.rs @@ -1,3 +1,7 @@ +use crate::auth::{ + BACKEND_RESOURCE_FETCH_PERMISSION, RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + RuntimeIdentityMaterial, RuntimeRequestSourceSigner, unix_now_seconds, +}; use crate::identity::WorkerId; use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex}; use async_trait::async_trait; @@ -108,6 +112,8 @@ pub trait BackendResourceClient: Send + Sync + 'static { pub struct HttpBackendResourceClient { endpoint: String, bearer_token: Option, + request_source_signer: Option, + request_source_audience: Option, client: reqwest::Client, } @@ -117,9 +123,21 @@ impl HttpBackendResourceClient { Self { endpoint: endpoint.into(), bearer_token, + request_source_signer: None, + request_source_audience: None, client: reqwest::Client::new(), } } + + pub fn with_runtime_request_source( + mut self, + identity: &RuntimeIdentityMaterial, + audience: impl Into, + ) -> Self { + self.request_source_signer = Some(RuntimeRequestSourceSigner::from_identity(identity)); + self.request_source_audience = Some(audience.into()); + self + } } #[cfg(feature = "http-server")] @@ -129,7 +147,44 @@ impl BackendResourceClient for HttpBackendResourceClient { &self, request: BackendResourceFetchRequest, ) -> Result { - let builder = self.client.post(&self.endpoint).json(&request); + let body = serde_json::to_vec(&request).map_err(|error| { + BackendResourceError::InvalidResponse { + message: error.to_string(), + } + })?; + let endpoint = reqwest::Url::parse(&self.endpoint).map_err(|error| { + BackendResourceError::Transport { + message: error.to_string(), + } + })?; + let mut builder = self + .client + .post(endpoint.clone()) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(body.clone()); + if let Some(signer) = self.request_source_signer.as_ref() { + let audience = self.request_source_audience.as_deref().ok_or_else(|| { + BackendResourceError::Unauthorized { + message: "Runtime request proof audience is unavailable".to_owned(), + } + })?; + let proof = signer + .issue( + audience, + &request.handle.workspace_id, + None, + BACKEND_RESOURCE_FETCH_PERMISSION, + "POST", + endpoint.path(), + &body, + i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX), + 30, + ) + .map_err(|error| BackendResourceError::Unauthorized { + message: error.to_string(), + })?; + builder = builder.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof); + } let builder = if let Some(token) = self.bearer_token.as_deref() { builder.bearer_auth(token) } else { diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index ed1538f4..aa38778f 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -14,7 +14,10 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; -use crate::auth::RuntimeIdentityMaterial; +use crate::auth::{ + BACKEND_RESOURCE_FETCH_PERMISSION, RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + RuntimeIdentityMaterial, RuntimeRequestSourceSigner, unix_now_seconds, +}; use crate::catalog::{ CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource, WorkingDirectoryRequest, WorkingDirectoryStatus, @@ -295,6 +298,7 @@ pub struct ProfileRuntimeWorkerFactory { prompt_projection_cache: Arc, runtime_id: Option, worker_mutation_identity: Option, + runtime_request_audience: Option, embedded_worker_mutation_dispatcher: Option>, controller_transport: WorkerControllerTransport, } @@ -311,6 +315,7 @@ impl ProfileRuntimeWorkerFactory { prompt_projection_cache: Arc::new(WorkspacePromptProjectionCache::default()), runtime_id: None, worker_mutation_identity: None, + runtime_request_audience: None, embedded_worker_mutation_dispatcher: None, controller_transport: WorkerControllerTransport::UnixSocket, } @@ -331,6 +336,17 @@ impl ProfileRuntimeWorkerFactory { self } + pub fn with_runtime_request_identity( + mut self, + identity: RuntimeIdentityMaterial, + audience: impl Into, + ) -> Self { + self.runtime_id = Some(identity.identity_id.clone()); + self.worker_mutation_identity = Some(identity); + self.runtime_request_audience = Some(audience.into()); + self + } + pub fn with_embedded_worker_mutation_dispatcher( mut self, runtime_id: impl Into, @@ -457,13 +473,15 @@ impl ProfileRuntimeWorkerFactory { async fn resolve_profile_source_archive( &self, source: &ProfileSourceArchiveSource, + request_audience: Option<&str>, ) -> Result { match source { ProfileSourceArchiveSource::Embedded { archive } => archive .verify() .map_err(|err| format!("failed to verify embedded profile source archive: {err}")), ProfileSourceArchiveSource::Http { location } => { - self.fetch_profile_source_archive(location).await + self.fetch_profile_source_archive(location, request_audience) + .await } } } @@ -471,10 +489,18 @@ impl ProfileRuntimeWorkerFactory { async fn fetch_profile_source_archive( &self, location: &ProfileSourceArchiveHttpRef, + request_audience: Option<&str>, ) -> Result { if let Some(cached) = self.profile_archive_cache.get(&location.archive.digest) { - let response = - fetch_profile_source_archive_http(location, Some(&location.archive.digest)).await?; + let response = fetch_profile_source_archive_http( + location, + Some(&location.archive.digest), + self.worker_mutation_identity.as_ref(), + self.runtime_request_audience + .as_deref() + .or(request_audience), + ) + .await?; if let Some(fetched) = response { self.profile_archive_cache.insert(fetched.clone()); fetched.verify().map_err(|err| { @@ -486,12 +512,19 @@ impl ProfileRuntimeWorkerFactory { .map_err(|err| format!("failed to verify cached profile source archive: {err}")) } } else { - let archive = fetch_profile_source_archive_http(location, None) - .await? - .ok_or_else(|| { - "profile source archive HTTP revalidation returned 304 without a cached archive" - .to_string() - })?; + let archive = fetch_profile_source_archive_http( + location, + None, + self.worker_mutation_identity.as_ref(), + self.runtime_request_audience + .as_deref() + .or(request_audience), + ) + .await? + .ok_or_else(|| { + "profile source archive HTTP revalidation returned 304 without a cached archive" + .to_string() + })?; self.profile_archive_cache.insert(archive.clone()); archive .verify() @@ -527,6 +560,7 @@ impl RuntimeWorkspaceBackendRef { worker_ref: &WorkerRef, workspace_scope: Option<&crate::runtime::RuntimeWorkspaceScope>, mutation_identity: Option<&RuntimeIdentityMaterial>, + runtime_request_audience: Option<&str>, embedded_dispatcher: Option<&Arc>, prompt_projection_cache: Option>, ) -> WorkerWorkspaceContext { @@ -546,6 +580,13 @@ impl RuntimeWorkspaceBackendRef { if let Some(cache) = prompt_projection_cache { client = client.with_prompt_projection_cache(cache); } + if let Some(identity) = mutation_identity { + let audience = runtime_request_audience + .or_else(|| workspace_scope.map(|scope| scope.server_id.as_str())); + if let Some(audience) = audience { + client = client.with_runtime_request_source(identity, audience.to_owned()); + } + } if let (Some(scope), Some(identity)) = (workspace_scope, mutation_identity) { client = client.with_worker_remove(RuntimeWorkerMutationForwarder::remote( identity, @@ -576,9 +617,40 @@ impl RuntimeWorkspaceBackendRef { async fn fetch_profile_source_archive_http( location: &ProfileSourceArchiveHttpRef, cached_digest: Option<&str>, + identity: Option<&RuntimeIdentityMaterial>, + audience: Option<&str>, ) -> Result, String> { let client = reqwest::Client::new(); - let mut request = client.get(&location.url); + let url = reqwest::Url::parse(&location.url) + .map_err(|error| format!("profile source archive URL is invalid: {error}"))?; + let path = url.path().to_owned(); + let workspace_id = path + .split('/') + .collect::>() + .windows(2) + .find_map(|parts| (parts[0] == "w").then_some(parts[1])) + .filter(|value| !value.is_empty()) + .ok_or_else(|| "profile source archive URL is not workspace-scoped".to_owned())?; + let mut request = client.get(url); + if let Some(identity) = identity { + let audience = audience.ok_or_else(|| { + "profile source archive request proof audience is unavailable".to_owned() + })?; + let proof = RuntimeRequestSourceSigner::from_identity(identity) + .issue( + audience, + workspace_id, + None, + BACKEND_RESOURCE_FETCH_PERMISSION, + "GET", + &path, + b"", + i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX), + 30, + ) + .map_err(|error| error.to_string())?; + request = request.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof); + } if cached_digest == Some(location.archive.digest.as_str()) { if let Some(etag) = location.etag.as_deref() { request = request.header(reqwest::header::IF_NONE_MATCH, etag); @@ -620,6 +692,8 @@ async fn fetch_profile_source_archive_http( async fn fetch_profile_source_archive_http( _location: &ProfileSourceArchiveHttpRef, _cached_digest: Option<&str>, + _identity: Option<&RuntimeIdentityMaterial>, + _audience: Option<&str>, ) -> Result, String> { Err( "HTTP profile source archive fetch requires the worker-runtime http-server feature" @@ -743,12 +817,19 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { &request.worker_ref, request.workspace_scope.as_ref(), self.worker_mutation_identity.as_ref(), + self.runtime_request_audience.as_deref(), self.embedded_worker_mutation_dispatcher.as_ref(), Some(self.prompt_projection_cache.clone()), ); let selector = profile.as_ref(); let archive = self - .resolve_profile_source_archive(&request.request.profile_source) + .resolve_profile_source_archive( + &request.request.profile_source, + request + .workspace_scope + .as_ref() + .map(|scope| scope.server_id.as_str()), + ) .await?; let (mut manifest, mut loader) = { let manifest = archive @@ -909,6 +990,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { &request.worker_ref, request.workspace_scope.as_ref(), self.worker_mutation_identity.as_ref(), + self.runtime_request_audience.as_deref(), self.embedded_worker_mutation_dispatcher.as_ref(), Some(self.prompt_projection_cache.clone()), ); @@ -2187,12 +2269,18 @@ mod tests { let scope = crate::runtime::RuntimeWorkspaceScope::new("workspace-a", "server-main"); let before_restart = - backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None); + backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None, None); let adapter = WorkerRuntimeExecutionBackend::new(FailingFactory).unwrap(); let (after_restore_kind, after_restore_workspace_id) = adapter .run_on_adapter_runtime(async move { - let after_restore = - backend.worker_context(&worker_ref, Some(&scope), Some(&identity), None, None); + let after_restore = backend.worker_context( + &worker_ref, + Some(&scope), + Some(&identity), + None, + None, + None, + ); let client = after_restore.client_handle(); Ok(( client.kind().to_string(), @@ -2383,6 +2471,7 @@ mod tests { None, None, None, + None, ); let workspace_client = workspace_context.client_handle(); self.observed_workspace_clients.lock().unwrap().push(( @@ -2781,7 +2870,7 @@ mod tests { archive: bundle.profile_source_archive.clone().unwrap(), }; factory - .resolve_profile_source_archive(&source) + .resolve_profile_source_archive(&source, None) .await .expect("embedded archive should resolve without Backend resource client"); } diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index fbf6f328..68ea1845 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -7,8 +7,9 @@ use worker::{ }; use crate::auth::{ - RuntimeAuthError, RuntimeIdentityMaterial, RuntimeWorkerMutationSourceSigner, - WORKER_REMOVE_PERMISSION, WorkerMutationActorKind, WorkerMutationOperation, + RUNTIME_REQUEST_SOURCE_PROOF_HEADER, RuntimeAuthError, RuntimeIdentityMaterial, + RuntimeRequestSourceSigner, RuntimeWorkerMutationSourceSigner, WORKER_REMOVE_PERMISSION, + WORKSPACE_REQUEST_PERMISSION, WorkerMutationActorKind, WorkerMutationOperation, WorkerMutationSourceClaims, new_token_id, }; use crate::runtime::RuntimeWorkspaceScope; @@ -289,6 +290,8 @@ pub struct RuntimeOwnedWorkspaceClient { worker_id: String, request_timeout: Option, worker_remove: Option, + request_source_signer: Option, + request_source_audience: Option, prompt_projection_cache: Option>, } @@ -306,6 +309,8 @@ impl RuntimeOwnedWorkspaceClient { worker_id: worker_id.into(), request_timeout: None, worker_remove: None, + request_source_signer: None, + request_source_audience: None, prompt_projection_cache: None, } } @@ -315,6 +320,16 @@ impl RuntimeOwnedWorkspaceClient { self } + pub fn with_runtime_request_source( + mut self, + identity: &RuntimeIdentityMaterial, + audience: impl Into, + ) -> Self { + self.request_source_signer = Some(RuntimeRequestSourceSigner::from_identity(identity)); + self.request_source_audience = Some(audience.into()); + self + } + pub(crate) fn with_prompt_projection_cache( mut self, cache: Arc, @@ -363,15 +378,21 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { request: WorkspaceRequest, ) -> Result { let base_url = self.base_url.clone(); + let workspace_id = self.workspace_id.clone(); let runtime_id = self.runtime_id.clone(); let worker_id = self.worker_id.clone(); + let request_source_signer = self.request_source_signer.clone(); + let request_source_audience = self.request_source_audience.clone(); let request_timeout = self.request_timeout; if tokio::runtime::Handle::try_current().is_ok() { std::thread::spawn(move || { execute_runtime_owned_workspace_http( &base_url, + &workspace_id, &runtime_id, &worker_id, + request_source_signer.as_ref(), + request_source_audience.as_deref(), request_timeout, request, ) @@ -383,8 +404,11 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { } else { execute_runtime_owned_workspace_http( &self.base_url, + &self.workspace_id, &self.runtime_id, &self.worker_id, + self.request_source_signer.as_ref(), + self.request_source_audience.as_deref(), self.request_timeout, request, ) @@ -484,8 +508,11 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { fn execute_runtime_owned_workspace_http( base_url: &str, + workspace_id: &str, runtime_id: &str, worker_id: &str, + request_source_signer: Option<&RuntimeRequestSourceSigner>, + request_source_audience: Option<&str>, request_timeout: Option, request: WorkspaceRequest, ) -> Result { @@ -510,11 +537,33 @@ fn execute_runtime_owned_workspace_http( )) })?; let request_label = format!("{method} {}", request.path); + let body = request.body.unwrap_or_default(); let mut request_builder = client - .request(method, url) + .request(method.clone(), url) .header("x-yoi-runtime-id", runtime_id) .header("x-yoi-worker-id", worker_id); - if let Some(body) = request.body { + if let Some(signer) = request_source_signer { + let audience = request_source_audience.ok_or_else(|| { + WorkspaceClientError::Request( + "runtime request proof audience is unavailable".to_owned(), + ) + })?; + let proof = signer + .issue( + audience, + workspace_id, + Some(worker_id), + WORKSPACE_REQUEST_PERMISSION, + method.as_str(), + &request.path, + body.as_bytes(), + i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX), + 30, + ) + .map_err(|error| WorkspaceClientError::Request(error.to_string()))?; + request_builder = request_builder.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof); + } + if !body.is_empty() { request_builder = request_builder .header(reqwest::header::CONTENT_TYPE, "application/json") .body(body); @@ -797,7 +846,7 @@ mod tests { } #[test] - fn ordinary_workspace_forwarding_stamps_legacy_source_only_inside_runtime() { + fn ordinary_workspace_forwarding_stamps_runtime_identity_and_signed_source_proof() { use std::io::{Read, Write}; use std::net::TcpListener; use std::sync::Mutex; @@ -817,12 +866,14 @@ mod tests { .unwrap(); }); + let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap(); let client = RuntimeOwnedWorkspaceClient::new( "workspace-a", format!("http://{address}"), "runtime-a", "worker-a", - ); + ) + .with_runtime_request_source(&identity, "server-a"); let response = client .execute(WorkspaceRequest::get("/api/w/workspace-a/tickets/search")) .unwrap(); @@ -831,6 +882,7 @@ mod tests { let request = received.lock().unwrap().to_ascii_lowercase(); assert!(request.contains("x-yoi-runtime-id: runtime-a")); assert!(request.contains("x-yoi-worker-id: worker-a")); + assert!(request.contains("x-yoi-runtime-request-proof: yoi-runtime-request-v1.")); assert!(!request.contains("authorization:")); }