diff --git a/docs/internal/server-secrets-strategy.md b/docs/internal/server-secrets-strategy.md index 3f492ea46..28c74add4 100644 --- a/docs/internal/server-secrets-strategy.md +++ b/docs/internal/server-secrets-strategy.md @@ -126,6 +126,7 @@ There is no startup-time secret generation or import of optional integration sec - Worker and render-graph subprocesses start from `env_clear()` and re-add only explicit allowlisted variables. - Authority-bearing values are re-injected intentionally. For worker subprocesses this is `FABRO_WORKER_TOKEN`, plus any explicitly required internal value such as a vault-derived `GITHUB_APP_PRIVATE_KEY`; it is not user auth state such as `FABRO_DEV_TOKEN` or `auth.json`. - The worker reads `FABRO_WORKER_TOKEN` from its env at startup (in `main()` before Tokio initializes) and immediately calls `std::env::remove_var` to scrub it. The token then flows through function arguments to `runner::execute`. Every descendant process (hooks, sandbox commands, MCP stdio, etc.) therefore inherits a worker env that no longer contains the bearer, so an unscrubbed spawn site cannot leak it. +- Worker tokens name the issuing server instance in their JWT audience. A new server instance rejects tokens from the previous instance before handling any request, even when `SESSION_SECRET` is unchanged. Startup recovery replaces in-flight workers; an old worker that survives the restart must not publish platform lifecycle records or reacquire a released Petri lease. This audience is a random, non-secret instance identifier, not a new bootstrap secret. User credentials keep their existing lifetime. - The daemon child inherits the parent env unchanged except for output-format hygiene (`FABRO_JSON` removal). ## Tests diff --git a/lib/apps/fabro-cli/tests/it/support/auth_tokens.rs b/lib/apps/fabro-cli/tests/it/support/auth_tokens.rs index d9b4fcb6b..d031f50df 100644 --- a/lib/apps/fabro-cli/tests/it/support/auth_tokens.rs +++ b/lib/apps/fabro-cli/tests/it/support/auth_tokens.rs @@ -63,6 +63,7 @@ struct TestJwtClaims { #[derive(serde::Serialize)] struct WorkerTokenClaims { iss: String, + aud: String, iat: u64, exp: u64, run_id: String, @@ -138,6 +139,10 @@ fn derived_jwt_key() -> [u8; 32] { key } +#[expect( + clippy::disallowed_methods, + reason = "Synchronous CLI test setup reads the isolated server audience before spawning a worker." +)] pub(crate) fn issue_test_worker_jwt(storage_dir: &Path, run_id: &str) -> String { let runtime_directory = Storage::new(storage_dir).runtime_directory(); let session_secret = envfile::read_env_file(&runtime_directory.env_path()) @@ -155,6 +160,12 @@ pub(crate) fn issue_test_worker_jwt(storage_dir: &Path, run_id: &str) -> String .expect("current timestamp should be positive"); let claims = WorkerTokenClaims { iss: WORKER_TOKEN_ISSUER.to_string(), + aud: std::fs::read_to_string( + runtime_directory + .record_path() + .with_file_name("test-worker-audience"), + ) + .expect("test server should publish its worker token audience"), iat: now, exp: now + WORKER_TOKEN_TTL_SECS, run_id: run_id.to_string(), diff --git a/lib/apps/fabro-server/src/petri_runs.rs b/lib/apps/fabro-server/src/petri_runs.rs index 1690fbe0e..0da398de1 100644 --- a/lib/apps/fabro-server/src/petri_runs.rs +++ b/lib/apps/fabro-server/src/petri_runs.rs @@ -74,7 +74,7 @@ impl PetriRuns { /// The writer handle `owner` holds on the run: the one kept from its /// open, or a reopen when the store's lease row still names the owner - /// (the server restarted, or the open's reply was lost). An owner the + /// (the open's reply was lost). An owner the /// lease no longer names gets `StaleOwner`. pub(crate) async fn writer( &self, @@ -515,6 +515,201 @@ mod tests { drop(before); } + /// A crash leaves the old worker alive, but only the replacement may + /// publish to the restarted server. Exercise the HTTP boundary before + /// the replacement starts, while it runs, and after it has settled. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn restarted_server_fences_stale_completion_and_keeps_resumed_result() { + for terminal in [ + RunStatus::Succeeded { + reason: SuccessReason::Completed, + }, + RunStatus::Failed { + reason: FailureReason::WorkflowError, + }, + ] { + let (store, artifacts) = test_store_bundle(); + let vault_path = test_secret_store_path(); + let before = TestAppStateBuilder::new() + .store_bundle(Arc::clone(&store), artifacts.clone()) + .vault_path(vault_path.clone()) + .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")]) + .build(); + let app = build_test_router(Arc::clone(&before)); + let run_id = create_and_start_petri_run(&app).await; + let old_token = before.test_issue_worker_token(&run_id); + run_to_running_as_worker(&app, run_id, &old_token).await; + let old_handle = before + .petri_runs + .open(run_id, Access::Create { + owner: OwnerId::new("old-worker"), + }) + .await + .expect("the old worker holds the lease"); + + let runtime = Arc::new(HeldWorkerRuntime::default()); + let after = TestAppStateBuilder::new() + .store_bundle(store, artifacts) + .vault_path(vault_path) + .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")]) + .worker_runtime(Arc::clone(&runtime) as Arc) + .build(); + let app = build_test_router(Arc::clone(&after)); + assert_eq!( + reconcile_incomplete_runs_on_startup(&after).await.unwrap(), + 1 + ); + assert_stale_completion_rejected(&after, &app, run_id, &old_token).await; + + write_test_server_record(&after); + spawn_scheduler(Arc::clone(&after)); + runtime.wait_for_start().await; + assert_eq!(runtime.launched_mode(), Some("resume")); + let token = after.test_issue_worker_token(&run_id); + let resumed = after + .petri_runs + .open(run_id, Access::Write { + owner: OwnerId::new("worker-1"), + }) + .await + .expect("the replacement takes the lease"); + run_to_running_as_worker(&app, run_id, &token).await; + assert_stale_completion_rejected(&after, &app, run_id, &old_token).await; + + let transition = if matches!(terminal, RunStatus::Succeeded { .. }) { + record_finish_as_worker(&app, run_id, &token).await; + RunLifecycleKind::Succeeded + } else { + // A current worker's failure before an engine finish must + // still be able to end the run. + RunLifecycleKind::Failed + }; + record_lifecycle_as_worker(&app, run_id, &token, transition, terminal).await; + assert_stale_completion_rejected(&after, &app, run_id, &old_token).await; + + let response = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/api/v1/runs/{run_id}/state")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let view: serde_json::Value = fabro_test::expect_axum_json( + response, + StatusCode::OK, + "GET /api/v1/runs/{id}/state after resumed completion", + ) + .await; + assert_eq!(view["status"], serde_json::to_value(terminal).unwrap()); + let response = app + .clone() + .oneshot( + Request::builder() + .uri(format!("/api/v1/runs/{run_id}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let summary = fabro_test::expect_axum_json( + response, + StatusCode::OK, + "GET /api/v1/runs/{id} after resumed completion", + ) + .await; + assert_eq!(summary["lifecycle"]["status"], view["status"]); + assert_eq!(after.test_managed_run_status(&run_id), Some(terminal)); + + runtime.end_worker(); + drop(resumed); + drop(old_handle); + drop(before); + } + } + + async fn assert_stale_completion_rejected( + state: &AppState, + app: &axum::Router, + run_id: RunId, + token: &str, + ) { + let view = run_records::projection(state, run_id) + .await + .unwrap() + .unwrap(); + let managed = state.test_managed_run_status(&run_id); + let records = state + .stores + .run_summaries + .platform_records() + .read(&run_id) + .await + .unwrap(); + let response = app + .clone() + .oneshot( + Request::builder() + .method("POST") + .uri(format!("/api/v1/runs/{run_id}/petri/open")) + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .header(header::CONTENT_TYPE, "application/json") + .body(Body::from( + json!({ "access": "write", "owner": "old-worker" }).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + fabro_test::assert_axum_status( + response, + StatusCode::UNAUTHORIZED, + "superseded worker cannot reacquire the lease", + ) + .await; + for (transition, status) in [ + (RunLifecycleKind::Failed, RunStatus::Failed { + reason: FailureReason::WorkflowError, + }), + (RunLifecycleKind::Succeeded, RunStatus::Succeeded { + reason: SuccessReason::Completed, + }), + ] { + let response = append_lifecycle_as_worker(app, run_id, token, transition, status).await; + fabro_test::assert_axum_status( + response, + StatusCode::UNAUTHORIZED, + "superseded worker terminal write", + ) + .await; + } + let after = run_records::projection(state, run_id) + .await + .unwrap() + .unwrap(); + assert_eq!( + after.status, view.status, + "stale completions leave the public result unchanged" + ); + assert_eq!(state.test_managed_run_status(&run_id), managed); + assert_eq!( + serde_json::to_value( + state + .stores + .run_summaries + .platform_records() + .read(&run_id) + .await + .unwrap() + ) + .unwrap(), + serde_json::to_value(records).unwrap(), + "rejected records never enter the durable log" + ); + } + /// A server whose one worker is held open by the test, with a run the /// worker has taken over the API: the run's id and the worker's token. async fn held_worker_run( @@ -558,13 +753,23 @@ mod tests { transition: RunLifecycleKind, status: RunStatus, ) { + let response = append_lifecycle_as_worker(app, run_id, token, transition, status).await; + fabro_test::assert_axum_status(response, StatusCode::OK, "worker lifecycle append").await; + } + + async fn append_lifecycle_as_worker( + app: &axum::Router, + run_id: RunId, + token: &str, + transition: RunLifecycleKind, + status: RunStatus, + ) -> axum::response::Response { let record = PlatformRecord::RunLifecycle(RunLifecycleRecord::new(transition).with_status(status)); let body = json!({ "record": serde_json::to_value(&record).expect("the record encodes"), }); - let response = app - .clone() + app.clone() .oneshot( Request::builder() .method("POST") @@ -575,8 +780,7 @@ mod tests { .expect("the append request builds"), ) .await - .expect("the append request completes"); - assert_eq!(response.status(), StatusCode::OK); + .expect("the append request completes") } /// Petri's own finish of the run, stored the way its worker stores it: diff --git a/lib/apps/fabro-server/src/principal_middleware.rs b/lib/apps/fabro-server/src/principal_middleware.rs index 190583e22..1e5c2c796 100644 --- a/lib/apps/fabro-server/src/principal_middleware.rs +++ b/lib/apps/fabro-server/src/principal_middleware.rs @@ -568,13 +568,12 @@ mod tests { use fabro_types::settings::ServerAuthMethod; use fabro_types::{AuthMethod, IdpIdentity, RunId}; use jsonwebtoken::EncodingKey; - use uuid::Uuid; use super::*; use crate::auth::{self, AuthErrorCode}; use crate::worker_token::{ - WORKER_RUN_TOOLS_SCOPE, WORKER_TOKEN_ISSUER, WORKER_TOKEN_SCOPE, WorkerScopeSet, - WorkerTokenClaims, issue_worker_token, worker_token_header, + WORKER_RUN_TOOLS_SCOPE, WORKER_TOKEN_SCOPE, WorkerScopeSet, WorkerTokenClaims, + issue_worker_token, worker_token_header, }; const TEST_JWT_ISSUER: &str = "https://fabro.example"; @@ -635,24 +634,29 @@ mod tests { let secret = state .server_secret(EnvVars::SESSION_SECRET) .expect("test state should have session secret"); - issue_worker_claims_with_secret(secret.as_bytes(), run_id, exp, scope) + issue_worker_claims_with_secret(state, secret.as_bytes(), run_id, exp, scope) } fn issue_worker_claims_with_secret( + state: &AppState, secret: &[u8], run_id: RunId, exp: u64, scope: &str, ) -> String { let worker_key = auth::derive_worker_jwt_key(secret).unwrap(); - let claims = WorkerTokenClaims { - iss: WORKER_TOKEN_ISSUER.to_string(), - iat: 1, - exp, - run_id: run_id.to_string(), - scope: scope.to_string(), - jti: Uuid::new_v4().simple().to_string(), - }; + let keys = state.worker_token_keys(); + let token = issue_worker_token(keys, &run_id).unwrap(); + let mut claims = jsonwebtoken::decode::( + &token, + keys.decoding_key(), + keys.validation(), + ) + .unwrap() + .claims; + claims.iat = 1; + claims.exp = exp; + claims.scope = scope.to_string(); jsonwebtoken::encode( &worker_token_header(), &claims, @@ -758,6 +762,7 @@ mod tests { fn classifies_invalid_worker_jwt_signature_as_invalid() { let state = crate::test_support::test_app_state(); let token = issue_worker_claims_with_secret( + state.as_ref(), b"other-principal-middleware-secret-0001", RunId::new(), u64::MAX / 2, diff --git a/lib/apps/fabro-server/src/serve.rs b/lib/apps/fabro-server/src/serve.rs index deedd584f..4e090304f 100644 --- a/lib/apps/fabro-server/src/serve.rs +++ b/lib/apps/fabro-server/src/serve.rs @@ -41,6 +41,8 @@ use crate::server::{ use crate::server_secrets::{ServerSecrets, process_env_snapshot}; use crate::startup::{resolve_startup, validate_startup_configuration}; use crate::static_files; +#[cfg(any(test, feature = "test-support"))] +use crate::test_support; pub const DEFAULT_TCP_PORT: u16 = 32276; type EnvLookup = Arc Option + Send + Sync>; @@ -749,6 +751,8 @@ where #[cfg(any(test, feature = "test-support"))] automation_materializer_override: None, })?; + #[cfg(any(test, feature = "test-support"))] + test_support::write_test_worker_audience(&state).await?; let reconciled = reconcile_incomplete_runs_on_startup(&state).await?; if reconciled > 0 { info!( diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index baaa5351e..1f2fcbe1a 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -523,10 +523,10 @@ async fn http_log_records_user_principal_fields() { #[tokio::test(flavor = "current_thread")] async fn http_log_records_worker_principal_fields() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_bearer = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_bearer).await; - let worker_bearer = issue_test_worker_token(&run_id); + let worker_bearer = issue_test_worker_token(&state, &run_id); let (_guard, events) = capture_server_logs(); let response = app @@ -663,17 +663,14 @@ fn issue_test_user_jwt() -> String { ) } -fn issue_test_worker_token(run_id: &RunId) -> String { - let keys = WorkerTokenKeys::from_master_secret(TEST_SESSION_SECRET.as_bytes()) - .expect("worker keys should derive"); - crate::worker_token::issue_worker_token(&keys, run_id).expect("worker token should issue") +fn issue_test_worker_token(state: &AppState, run_id: &RunId) -> String { + crate::worker_token::issue_worker_token(state.worker_token_keys(), run_id) + .expect("worker token should issue") } -fn issue_test_run_tools_worker_token(run_id: &RunId) -> String { - let keys = WorkerTokenKeys::from_master_secret(TEST_SESSION_SECRET.as_bytes()) - .expect("worker keys should derive"); +fn issue_test_run_tools_worker_token(state: &AppState, run_id: &RunId) -> String { crate::worker_token::issue_worker_token_with_scopes( - &keys, + state.worker_token_keys(), run_id, crate::worker_token::WorkerScopeSet::run_worker_with_agent_run_tools(), ) @@ -846,12 +843,12 @@ async fn next_worker_control_frame( #[tokio::test(flavor = "current_thread")] async fn worker_control_stream_rejects_missing_user_and_cross_run_auth() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_bearer = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_bearer).await; - let worker_bearer = issue_test_worker_token(&run_id); + let worker_bearer = issue_test_worker_token(&state, &run_id); let other_run_id = create_run_with_bearer(&app, &user_bearer).await; - let other_worker_bearer = issue_test_worker_token(&other_run_id); + let other_worker_bearer = issue_test_worker_token(&state, &other_run_id); let server = WorkerControlWsTestServer::spawn(app).await; assert_worker_control_ws_rejected(&server, run_id, None, None, StatusCode::UNAUTHORIZED).await; @@ -883,7 +880,7 @@ async fn worker_control_stream_start_subscription_delivers_frames() { let (state, app) = jwt_auth_app(); let user_bearer = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_bearer).await; - let worker_bearer = issue_test_worker_token(&run_id); + let worker_bearer = issue_test_worker_token(&state, &run_id); let server = WorkerControlWsTestServer::spawn(app).await; let mut socket = connect_worker_control_ws(&server, run_id, &worker_bearer, None).await; @@ -904,7 +901,7 @@ async fn worker_control_stream_after_subscription_delivers_only_later_frames() { let (state, app) = jwt_auth_app(); let user_bearer = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_bearer).await; - let worker_bearer = issue_test_worker_token(&run_id); + let worker_bearer = issue_test_worker_token(&state, &run_id); let first = state .worker_control_bus .publish(run_id, WorkerControlEnvelope::cancel_run()) @@ -934,7 +931,7 @@ async fn worker_control_stream_acknowledgements_settle_the_waiting_caller() { let (state, app) = jwt_auth_app(); let user_bearer = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_bearer).await; - let worker_bearer = issue_test_worker_token(&run_id); + let worker_bearer = issue_test_worker_token(&state, &run_id); let server = WorkerControlWsTestServer::spawn(app).await; let mut socket = connect_worker_control_ws(&server, run_id, &worker_bearer, None).await; @@ -973,10 +970,10 @@ async fn worker_control_stream_acknowledgements_settle_the_waiting_caller() { #[tokio::test(flavor = "current_thread")] async fn worker_control_stream_invalid_cursor_is_http_gone_before_upgrade() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_bearer = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_bearer).await; - let worker_bearer = issue_test_worker_token(&run_id); + let worker_bearer = issue_test_worker_token(&state, &run_id); let server = WorkerControlWsTestServer::spawn(app).await; assert_worker_control_ws_rejected( @@ -4108,7 +4105,7 @@ async fn run_tools_worker_cannot_select_server_folder_from_clone_based_parent() let (state, app) = jwt_auth_app(); let user_token = issue_test_user_jwt(); let parent_run_id = create_run_with_bearer(&app, &user_token).await; - let worker_token = issue_test_run_tools_worker_token(&parent_run_id); + let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id); let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await; let mut intent = folder_intent(workflow_version_id, missing_target.to_string_lossy()); intent["environment_id"] = json!("local"); @@ -4147,7 +4144,7 @@ async fn run_tools_worker_cannot_select_server_folder_from_clone_based_parent() async fn run_tools_worker_folder_target_from_missing_parent_run_is_not_found() { let dir = tempfile::tempdir().unwrap(); let (state, app) = jwt_auth_app(); - let worker_token = issue_test_run_tools_worker_token(&RunId::new()); + let worker_token = issue_test_run_tools_worker_token(&state, &RunId::new()); let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await; let mut intent = folder_intent(workflow_version_id, dir.path().to_string_lossy()); intent["environment_id"] = json!("local"); @@ -4196,7 +4193,7 @@ async fn run_tools_worker_can_select_server_folder_from_local_parent() { .unwrap(); let parent = response_json!(response, StatusCode::CREATED).await; let parent_run_id = parent["id"].as_str().unwrap().parse::().unwrap(); - let worker_token = issue_test_run_tools_worker_token(&parent_run_id); + let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id); let mut child_intent = folder_intent(workflow_version_id, dir.path().to_string_lossy()); child_intent["environment_id"] = json!("local"); child_intent["parent_id"] = json!(parent_run_id); @@ -7316,9 +7313,9 @@ async fn worker_token_accepts_run_scoped_routes_and_falls_back_to_user_jwt() { let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_worker_token(&run_id); + let worker_token = issue_test_worker_token(&state, &run_id); let other_run_id = create_run_with_bearer(&app, &user_jwt).await; - let other_worker_token = issue_test_worker_token(&other_run_id); + let other_worker_token = issue_test_worker_token(&state, &other_run_id); let blob_hash = state .store_ref() .blobs() @@ -7422,7 +7419,7 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() { let user_jwt = issue_test_user_jwt(); let parent_run_id = create_run_with_bearer(&app, &user_jwt).await; let target_run_id = create_run_with_bearer(&app, &user_jwt).await; - let run_tool_worker_token = issue_test_run_tools_worker_token(&parent_run_id); + let run_tool_worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id); let response = app .clone() @@ -7575,11 +7572,11 @@ async fn run_tool_worker_token_can_use_client_backend_routes_across_runs() { #[tokio::test] async fn cross_run_base_worker_remains_forbidden_from_pair_routes() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let origin_run_id = create_run_with_bearer(&app, &user_jwt).await; let target_run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_worker_token(&origin_run_id); + let worker_token = issue_test_worker_token(&state, &origin_run_id); let response = app .clone() @@ -7596,11 +7593,11 @@ async fn cross_run_base_worker_remains_forbidden_from_pair_routes() { #[tokio::test] async fn run_tools_worker_cannot_call_user_only_non_mcp_routes() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let origin_run_id = create_run_with_bearer(&app, &user_jwt).await; let target_run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_run_tools_worker_token(&origin_run_id); + let worker_token = issue_test_run_tools_worker_token(&state, &origin_run_id); for (method, path) in [ (Method::POST, format!("/runs/{target_run_id}/approve")), @@ -7629,10 +7626,10 @@ async fn run_tools_worker_cannot_call_user_only_non_mcp_routes() { #[tokio::test] async fn base_worker_token_is_rejected_by_run_tool_only_routes() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_worker_token(&run_id); + let worker_token = issue_test_worker_token(&state, &run_id); for (method, path) in [ (Method::GET, "/runs".to_string()), @@ -7662,10 +7659,10 @@ async fn base_worker_token_is_rejected_by_run_tool_only_routes() { #[tokio::test] async fn worker_token_is_rejected_on_user_only_routes() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_worker_token(&run_id); + let worker_token = issue_test_worker_token(&state, &run_id); let blob_hash = BlobHash::new(b"blob"); let user_only_routes = vec![ (Method::GET, "/runs".to_string()), @@ -7832,7 +7829,7 @@ async fn worker_started_child_run_requires_approval_before_becoming_runnable() { let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let parent_run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_run_tools_worker_token(&parent_run_id); + let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id); let mut child_intent = test_intent_with_bearer(&app, "workflow.fabro", MINIMAL_DOT, None, Some(&user_jwt)).await; child_intent["parent_id"] = json!(parent_run_id.to_string()); @@ -7931,10 +7928,10 @@ async fn worker_started_child_run_requires_approval_before_becoming_runnable() { #[tokio::test] async fn denying_pending_child_run_fails_with_approval_denied() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let parent_run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_run_tools_worker_token(&parent_run_id); + let worker_token = issue_test_run_tools_worker_token(&state, &parent_run_id); let mut child_intent = test_intent_with_bearer(&app, "workflow.fabro", MINIMAL_DOT, None, Some(&user_jwt)).await; child_intent["parent_id"] = json!(parent_run_id.to_string()); @@ -8947,10 +8944,10 @@ fn batch_delete_body(run_ids: &[RunId], force: bool) -> serde_json::Value { #[tokio::test] async fn batch_lifecycle_requires_user_authentication() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_worker_token(&run_id); + let worker_token = issue_test_worker_token(&state, &run_id); let body = batch_lifecycle_body(&[run_id]); let unauthenticated = app @@ -8984,10 +8981,10 @@ async fn batch_lifecycle_requires_user_authentication() { #[tokio::test] async fn batch_delete_requires_user_authentication() { - let (_state, app) = jwt_auth_app(); + let (state, app) = jwt_auth_app(); let user_jwt = issue_test_user_jwt(); let run_id = create_run_with_bearer(&app, &user_jwt).await; - let worker_token = issue_test_worker_token(&run_id); + let worker_token = issue_test_worker_token(&state, &run_id); let body = batch_delete_body(&[run_id], false); let unauthenticated = app @@ -10637,10 +10634,13 @@ async fn workflow_version_registration_requires_user_or_run_tools_capability() { for (token, expected) in [ (issue_test_user_jwt(), StatusCode::CREATED), ( - issue_test_run_tools_worker_token(&run_id), + issue_test_run_tools_worker_token(&state, &run_id), StatusCode::CREATED, ), - (issue_test_worker_token(&run_id), StatusCode::FORBIDDEN), + ( + issue_test_worker_token(&state, &run_id), + StatusCode::FORBIDDEN, + ), ] { let response = app .clone() @@ -10805,7 +10805,7 @@ fn slack_service_respects_disabled_server_config_even_with_vault_tokens() { async fn run_tools_worker_registers_contents_then_creates_by_version_id() { let (state, app) = jwt_auth_app(); let parent_id = create_run_with_bearer(&app, &issue_test_user_jwt()).await; - let token = issue_test_run_tools_worker_token(&parent_id); + let token = issue_test_run_tools_worker_token(&state, &parent_id); let response = app .clone() .oneshot(json_bearer_request( diff --git a/lib/apps/fabro-server/src/server/tests/artifact_storage.rs b/lib/apps/fabro-server/src/server/tests/artifact_storage.rs index 84ad08499..35bd9eb20 100644 --- a/lib/apps/fabro-server/src/server/tests/artifact_storage.rs +++ b/lib/apps/fabro-server/src/server/tests/artifact_storage.rs @@ -25,7 +25,7 @@ async fn artifact_upload_accepts_large_and_exact_limit_content_without_sqlite_or let (state, app) = jwt_auth_app(); let user = issue_test_user_jwt(); let run = create_run_with_bearer(&app, &user).await; - let worker = issue_test_worker_token(&run); + let worker = issue_test_worker_token(&state, &run); for size in [3 * 1024 * 1024, ARTIFACT_MAX_FILE_BYTES] { let bytes = vec![0xa3; size]; let hash = BlobHash::new(&bytes); @@ -87,8 +87,8 @@ async fn artifact_upload_rejects_unauthorized_invalid_and_oversized_bodies_witho let (state, app) = jwt_auth_app(); let user = issue_test_user_jwt(); let run = create_run_with_bearer(&app, &user).await; - let worker = issue_test_worker_token(&run); - let other_worker = issue_test_worker_token(&RunId::new()); + let worker = issue_test_worker_token(&state, &run); + let other_worker = issue_test_worker_token(&state, &RunId::new()); let hash = BlobHash::new(b"valid content"); state .artifact_store @@ -152,7 +152,7 @@ async fn artifact_upload_rejects_unauthorized_invalid_and_oversized_bodies_witho .oneshot(upload( missing_run, &digest, - &issue_test_worker_token(&missing_run), + &issue_test_worker_token(&state, &missing_run), Body::from("valid content"), )) .await @@ -246,7 +246,7 @@ async fn artifact_worker_client_uses_the_configured_s3_backend_and_prefix() { }) .await; let server = WorkerControlWsTestServer::spawn(app).await; - let worker_token = issue_test_worker_token(&run); + let worker_token = issue_test_worker_token(&state, &run); let mut headers = fabro_http::header::HeaderMap::new(); headers.insert( fabro_http::header::AUTHORIZATION, diff --git a/lib/apps/fabro-server/src/test_support.rs b/lib/apps/fabro-server/src/test_support.rs index dc3350fc3..9e35c70a5 100644 --- a/lib/apps/fabro-server/src/test_support.rs +++ b/lib/apps/fabro-server/src/test_support.rs @@ -28,6 +28,7 @@ use fabro_types::{ use fabro_vault::{SecretType, Vault}; use lithos_llm::catalog::ProviderId; use object_store::memory::InMemory as MemoryObjectStore; +use tokio::fs as tokio_fs; use tokio::runtime::Builder as TokioRuntimeBuilder; use tokio_util::sync::CancellationToken; use ulid::Ulid; @@ -745,6 +746,20 @@ pub fn build_test_router(state: Arc) -> Router { with_test_user(server::build_router(state, test_auth_mode())) } +/// The CLI's direct-worker tests sign their own tokens with the isolated +/// server's test secret. Publish only the non-secret audience so those +/// tokens address the actual server instance, without overriding the +/// production fence or persisting any worker credential. +pub(crate) async fn write_test_worker_audience(state: &AppState) -> anyhow::Result<()> { + let path = Storage::new(state.server_storage_dir()) + .runtime_directory() + .record_path() + .with_file_name("test-worker-audience"); + tokio_fs::create_dir_all(path.parent().expect("the runtime path has a parent")).await?; + tokio_fs::write(path, state.worker_token_keys().test_audience()).await?; + Ok(()) +} + pub fn build_test_router_with_options(state: Arc, options: RouterOptions) -> Router { with_test_user(server::build_router_with_options( state, diff --git a/lib/apps/fabro-server/src/worker_token.rs b/lib/apps/fabro-server/src/worker_token.rs index b10568c91..276597802 100644 --- a/lib/apps/fabro-server/src/worker_token.rs +++ b/lib/apps/fabro-server/src/worker_token.rs @@ -22,20 +22,27 @@ pub(crate) struct WorkerTokenKeys { encoding: Arc, decoding: Arc, validation: Arc, + /// Startup recovery replaces workers. Their credentials must therefore + /// be valid only for the server instance that launched them, including + /// platform writes made before acquiring or after releasing a lease. + audience: String, } impl WorkerTokenKeys { pub(crate) fn from_master_secret(secret: &[u8]) -> Result { let key = auth::derive_worker_jwt_key(secret)?; + let audience = Uuid::new_v4().to_string(); let mut validation = Validation::new(Algorithm::HS256); validation.validate_nbf = false; - validation.set_required_spec_claims(&["iss", "iat", "exp"]); + validation.set_required_spec_claims(&["iss", "iat", "exp", "aud"]); validation.set_issuer(&[WORKER_TOKEN_ISSUER]); + validation.set_audience(&[&audience]); Ok(Self { - encoding: Arc::new(EncodingKey::from_secret(&key)), - decoding: Arc::new(DecodingKey::from_secret(&key)), + encoding: Arc::new(EncodingKey::from_secret(&key)), + decoding: Arc::new(DecodingKey::from_secret(&key)), validation: Arc::new(validation), + audience, }) } @@ -48,11 +55,17 @@ impl WorkerTokenKeys { pub(crate) fn validation(&self) -> &Validation { &self.validation } + + #[cfg(any(test, feature = "test-support"))] + pub(crate) fn test_audience(&self) -> &str { + &self.audience + } } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] pub(crate) struct WorkerTokenClaims { pub(crate) iss: String, + pub(crate) aud: String, pub(crate) iat: u64, pub(crate) exp: u64, pub(crate) run_id: String, @@ -117,6 +130,7 @@ pub(crate) fn issue_worker_token_with_scopes( .map_or(0, |duration| duration.as_secs()); let claims = WorkerTokenClaims { iss: WORKER_TOKEN_ISSUER.to_string(), + aud: keys.audience.clone(), iat: now, exp: now + WORKER_TOKEN_TTL_SECS, run_id: run_id.to_string(), @@ -220,6 +234,7 @@ mod tests { ) -> String { let claims = WorkerTokenClaims { iss: WORKER_TOKEN_ISSUER.to_string(), + aud: keys.audience.clone(), iat: 1, exp: u64::MAX / 2, run_id: run_id.to_string(), @@ -233,6 +248,7 @@ mod tests { fn expired_worker_token(keys: &WorkerTokenKeys, run_id: &fabro_types::RunId) -> String { let claims = WorkerTokenClaims { iss: WORKER_TOKEN_ISSUER.to_string(), + aud: keys.audience.clone(), iat: 1, exp: 2, run_id: run_id.to_string(), @@ -276,6 +292,7 @@ mod tests { assert_eq!(decoded.claims, WorkerTokenClaims { iss: WORKER_TOKEN_ISSUER.to_string(), + aud: keys.audience.clone(), iat: decoded.claims.iat, exp: decoded.claims.exp, run_id: run_id.to_string(), @@ -288,16 +305,49 @@ mod tests { } #[test] - fn worker_token_survives_key_rederivation() { + fn worker_token_is_rejected_by_a_restarted_server_with_the_same_secret() { let run_id = run_id(); let first = keys(TEST_SECRET); let second = keys(TEST_SECRET); let token = issue_worker_token(&first, &run_id).expect("worker token should issue"); - let decoded = decode::(&token, &second.decoding, &second.validation) - .expect("worker token should decode after re-derivation"); + assert_eq!( + decode_worker_token(&token, &second), + Err(JwtError::AccessTokenInvalid) + ); + let replacement = issue_worker_token(&second, &run_id).unwrap(); + assert_eq!( + decode_worker_token(&replacement, &second).unwrap().run_id, + run_id + ); + } - assert_eq!(decoded.claims.run_id, run_id.to_string()); + #[test] + fn worker_token_is_valid_for_clones_of_the_issuing_server_keys() { + let keys = keys(TEST_SECRET); + let token = issue_worker_token(&keys, &run_id()).unwrap(); + assert_eq!( + decode_worker_token(&token, &keys.clone()).unwrap().run_id, + run_id() + ); + } + + #[test] + fn worker_token_without_a_server_audience_is_rejected() { + let keys = keys(TEST_SECRET); + let token = issue_worker_token(&keys, &run_id()).unwrap(); + let mut claims = serde_json::to_value( + decode::(&token, &keys.decoding, &keys.validation) + .unwrap() + .claims, + ) + .unwrap(); + claims.as_object_mut().unwrap().remove("aud"); + let legacy = jsonwebtoken::encode(&worker_token_header(), &claims, &keys.encoding).unwrap(); + assert_eq!( + decode_worker_token(&legacy, &keys), + Err(JwtError::AccessTokenInvalid) + ); } #[test]