test(auth): backfill worker jwt regression coverage

This commit is contained in:
Bryan Helmkamp 2026-04-23 11:35:08 -04:00
parent 519f46038d
commit 891b7f90ae
No known key found for this signature in database
5 changed files with 927 additions and 46 deletions

View file

@ -18,7 +18,7 @@ These values belong to the server runtime and are read via `state.server_secret(
| Secret | Used by |
|---|---|
| `SESSION_SECRET` | Cookie encryption and JWT signing derivation |
| `FABRO_DEV_TOKEN` | Dev-token auth for worker/server interactions |
| `FABRO_DEV_TOKEN` | Dev-token user auth when `server.auth.methods` includes `dev-token` |
| `GITHUB_APP_PRIVATE_KEY` | GitHub App credentials |
| `GITHUB_APP_WEBHOOK_SECRET` | GitHub webhook verification |
| `GITHUB_APP_CLIENT_SECRET` | GitHub OAuth login |
@ -46,7 +46,7 @@ There is no compatibility layer for removed secrets and no startup-time secret g
## Subprocess Boundaries
- Worker and render-graph subprocesses start from `env_clear()` and re-add only explicit allowlisted variables.
- Authority-bearing values are re-injected intentionally.
- Authority-bearing values are re-injected intentionally. For worker subprocesses this is `FABRO_WORKER_TOKEN`, not user auth state such as `FABRO_DEV_TOKEN` or `auth.json`.
- The daemon child inherits the parent env unchanged except for output-format hygiene (`FABRO_JSON` removal).
## Tests

View file

@ -388,6 +388,8 @@ async fn wait_for_server_ready(http_client: &fabro_http::HttpClient) -> Result<(
mod tests {
use chrono::{Duration as ChronoDuration, Utc};
use fabro_client::{AuthEntry, StoredSubject};
use httpmock::Method::{GET, POST};
use serde_json::json;
use super::*;
@ -553,6 +555,87 @@ mod tests {
assert!(local_dev_token_fallback(&target));
}
#[tokio::test]
async fn connect_server_target_with_bearer_sends_worker_bearer_token() {
let server = httpmock::MockServer::start();
let info_mock = server.mock(|when, then| {
when.method(GET)
.path("/api/v1/system/info")
.header("authorization", "Bearer worker-token");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"version": "1.2.3",
"git_sha": "abcdef0",
"build_date": "2026-04-20",
"profile": "release",
"os": "darwin",
"arch": "arm64",
"storage_dir": "/tmp/fabro-worker-auth",
"storage_engine": "slatedb",
"runs": { "total": 0, "active": 0 },
"uptime_secs": 42
}));
});
let target = ServerTarget::http_url(server.base_url()).unwrap();
let client = connect_server_target_with_bearer(&target, "worker-token")
.await
.unwrap();
let info = client.get_system_info().await.unwrap();
assert_eq!(info.version.as_deref(), Some("1.2.3"));
info_mock.assert();
}
#[tokio::test]
async fn connect_server_target_with_bearer_does_not_attempt_oauth_refresh() {
let server = httpmock::MockServer::start();
let info_mock = server.mock(|when, then| {
when.method(GET)
.path("/api/v1/system/info")
.header("authorization", "Bearer worker-token");
then.status(401)
.header("Content-Type", "application/json")
.json_body(json!({
"errors": [{
"status": "401",
"title": "Unauthorized",
"detail": "Access token expired.",
"code": "access_token_expired"
}]
}));
});
let refresh_mock = server.mock(|when, then| {
when.method(POST).path("/auth/cli/refresh");
then.status(200)
.header("Content-Type", "application/json")
.json_body(json!({
"access_token": "unused",
"access_token_expires_at": (Utc::now() + ChronoDuration::minutes(10)).to_rfc3339(),
"refresh_token": "unused",
"refresh_token_expires_at": (Utc::now() + ChronoDuration::days(30)).to_rfc3339(),
"subject": {
"idp_issuer": "https://github.com",
"idp_subject": "12345",
"login": "octocat",
"name": "Octo Cat",
"email": "octocat@example.com"
}
}));
});
let target = ServerTarget::http_url(server.base_url()).unwrap();
let client = connect_server_target_with_bearer(&target, "worker-token")
.await
.unwrap();
let err = client.get_system_info().await.unwrap_err();
assert!(err.to_string().contains("Access token expired"));
info_mock.assert();
assert_eq!(refresh_mock.calls(), 0);
}
fn oauth_entry(
access_token_expires_at: chrono::DateTime<chrono::Utc>,
refresh_token_expires_at: chrono::DateTime<chrono::Utc>,

View file

@ -69,6 +69,7 @@ mod upgrade;
mod validate;
mod version;
mod wait;
mod worker_auth;
mod workflow;
mod workflow_create;
mod workflow_list;

View file

@ -0,0 +1,527 @@
#![expect(
clippy::disallowed_methods,
reason = "These worker-auth regressions start a real server subprocess, write isolated auth fixtures, and spawn the compiled fabro binary."
)]
#![expect(
clippy::disallowed_types,
reason = "These regressions intentionally own Child processes to exercise the real server-dispatched worker path."
)]
use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};
use chrono::{Duration as ChronoDuration, Utc};
use fabro_client::{AuthEntry, AuthStore, ServerTarget, StoredSubject};
use fabro_config::{Storage, envfile};
use fabro_store::EventEnvelope;
use fabro_test::{apply_test_isolation, expect_reqwest_json, isolated_storage_dir, test_context};
use fabro_types::{IdpIdentity, RunAuthMethod};
use hkdf::Hkdf;
use jsonwebtoken::{Algorithm, EncodingKey, Header};
use sha2::Sha256;
use super::support::{find_run_dir, output_stderr, output_stdout};
use crate::support::{parse_event_envelopes, unique_run_id};
const COMMAND_TIMEOUT: Duration = Duration::from_secs(30);
const TEST_SESSION_SECRET: &str =
"0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
const TEST_GITHUB_CLIENT_SECRET: &str = "github-client-secret";
const WORKER_TOKEN_ISSUER: &str = "fabro-server-worker";
const WORKER_TOKEN_SCOPE: &str = "run:worker";
const WORKER_TOKEN_TTL_SECS: u64 = 72 * 60 * 60;
struct RunningGithubOnlyServer {
child: Option<Child>,
home_root: tempfile::TempDir,
worker_home: PathBuf,
_storage_root: tempfile::TempDir,
storage_dir: PathBuf,
api_base_url: String,
}
impl RunningGithubOnlyServer {
async fn start() -> Self {
let home_root = tempfile::tempdir_in("/tmp").unwrap();
let worker_home = home_root.path().join("worker-home");
std::fs::create_dir_all(&worker_home).unwrap();
let storage_root = isolated_storage_dir();
let storage_dir = storage_root.path().join("storage");
let port = reserve_port();
let api_base_url = format!("http://127.0.0.1:{port}");
let config_path = home_root.path().join("settings.toml");
std::fs::write(
&config_path,
format!(
r#"_version = 1
[server.web]
enabled = true
url = "{api_base_url}"
[server.auth]
methods = ["github"]
[server.auth.github]
allowed_usernames = ["octocat"]
[server.integrations.github]
client_id = "github-client-id"
"#
),
)
.unwrap();
envfile::merge_env_file(
&Storage::new(&storage_dir).runtime_directory().env_path(),
[
("SESSION_SECRET", TEST_SESSION_SECRET),
("GITHUB_APP_CLIENT_SECRET", TEST_GITHUB_CLIENT_SECRET),
],
)
.unwrap();
let mut cmd = Command::new(env!("CARGO_BIN_EXE_fabro"));
apply_test_isolation(&mut cmd, home_root.path());
cmd.env("FABRO_HOME", &worker_home);
cmd.args(["server", "start", "--foreground"])
.arg("--storage-dir")
.arg(&storage_dir)
.arg("--bind")
.arg(format!("127.0.0.1:{port}"))
.arg("--config")
.arg(&config_path)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::piped());
let mut child = cmd.spawn().expect("github-only server should spawn");
wait_for_http_ready(&api_base_url, &mut child).await;
Self {
child: Some(child),
home_root,
worker_home,
_storage_root: storage_root,
storage_dir,
api_base_url,
}
}
fn target(&self) -> String {
format!("{}/api/v1", self.api_base_url)
}
async fn shutdown(mut self) {
let mut stop = Command::new(env!("CARGO_BIN_EXE_fabro"));
apply_test_isolation(&mut stop, self.home_root.path());
stop.env("FABRO_HOME", &self.worker_home);
stop.args(["server", "stop"])
.arg("--storage-dir")
.arg(&self.storage_dir);
let output = stop.output().expect("server stop should run");
assert!(
output.status.success(),
"server stop failed\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
let output = self
.child
.take()
.expect("server child should still be present")
.wait_with_output()
.expect("server output should be readable");
assert!(
output.status.success(),
"github-only server exited unsuccessfully\nstderr:\n{}",
String::from_utf8_lossy(&output.stderr)
);
}
}
impl Drop for RunningGithubOnlyServer {
fn drop(&mut self) {
if let Some(child) = self.child.as_mut() {
if child.try_wait().ok().flatten().is_none() {
let _ = child.kill();
let _ = child.wait();
}
}
}
}
#[derive(serde::Serialize)]
struct WorkerTokenClaims {
iss: String,
iat: u64,
exp: u64,
run_id: String,
scope: String,
jti: String,
}
#[derive(serde::Serialize)]
struct TestJwtClaims {
iss: String,
aud: String,
sub: String,
exp: u64,
iat: u64,
jti: String,
idp_issuer: String,
idp_subject: String,
login: String,
name: String,
email: String,
avatar_url: String,
user_url: String,
auth_method: RunAuthMethod,
}
fn reserve_port() -> u16 {
std::net::TcpListener::bind("127.0.0.1:0")
.unwrap()
.local_addr()
.unwrap()
.port()
}
fn issue_test_github_jwt(issuer: &str) -> String {
let key = derived_jwt_key();
let identity = IdpIdentity::new("https://github.com", "12345").unwrap();
let now = Utc::now();
let claims = TestJwtClaims {
iss: issuer.to_string(),
aud: "fabro-cli".to_string(),
sub: identity.subject().to_string(),
exp: (now + ChronoDuration::minutes(10))
.timestamp()
.try_into()
.expect("expiration time should be positive"),
iat: now
.timestamp()
.try_into()
.expect("issued-at time should be positive"),
jti: format!("{:032x}", rand::random::<u128>()),
idp_issuer: identity.issuer().to_string(),
idp_subject: identity.subject().to_string(),
login: "octocat".to_string(),
name: "The Octocat".to_string(),
email: "octocat@example.com".to_string(),
avatar_url: "https://example.com/octocat.png".to_string(),
user_url: "https://github.com/octocat".to_string(),
auth_method: RunAuthMethod::Github,
};
jsonwebtoken::encode(
&Header::new(Algorithm::HS256),
&claims,
&EncodingKey::from_secret(&key),
)
.unwrap()
}
fn derived_jwt_key() -> [u8; 32] {
let hkdf = Hkdf::<Sha256>::new(None, TEST_SESSION_SECRET.as_bytes());
let mut key = [0_u8; 32];
hkdf.expand(b"fabro-jwt-hs256-v1", &mut key)
.expect("jwt hkdf output should fit");
key
}
fn write_submitter_auth(home_dir: &Path, target: &str, access_token: &str) {
let auth_store = AuthStore::new(home_dir.join(".fabro").join("auth.json"));
let target = ServerTarget::http_url(target).unwrap();
let now = Utc::now();
auth_store
.put(&target, AuthEntry {
access_token: access_token.to_string(),
access_token_expires_at: now + ChronoDuration::minutes(10),
refresh_token: "refresh-unused".to_string(),
refresh_token_expires_at: now + ChronoDuration::days(30),
subject: StoredSubject {
idp_issuer: "https://github.com".to_string(),
idp_subject: "12345".to_string(),
login: "octocat".to_string(),
name: "The Octocat".to_string(),
email: "octocat@example.com".to_string(),
},
logged_in_at: now,
})
.unwrap();
}
fn write_probe_workflow(path: &Path) {
std::fs::write(
path,
r#"digraph WorkerAuthProbe {
graph [goal="Verify github-only worker auth", default_max_retries=0]
start [shape=Mdiamond]
exit [shape=Msquare]
probe [shape=parallelogram, script="printf worker-auth-ok"]
start -> probe -> exit
}
"#,
)
.unwrap();
}
fn issue_worker_token_for_run(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())
.expect("server env should load")
.get("SESSION_SECRET")
.cloned()
.expect("server env should include SESSION_SECRET");
let hkdf = Hkdf::<Sha256>::new(None, session_secret.as_bytes());
let mut key = [0_u8; 32];
hkdf.expand(b"fabro-worker-jwt-v1", &mut key)
.expect("worker jwt hkdf output should fit");
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let claims = WorkerTokenClaims {
iss: WORKER_TOKEN_ISSUER.to_string(),
iat: now,
exp: now + WORKER_TOKEN_TTL_SECS,
run_id: run_id.to_string(),
scope: WORKER_TOKEN_SCOPE.to_string(),
jti: format!("{:032x}", rand::random::<u128>()),
};
jsonwebtoken::encode(
&Header::new(Algorithm::HS256),
&claims,
&EncodingKey::from_secret(&key),
)
.expect("worker token should encode")
}
fn wait_for_run_dir(storage_dir: &Path, run_id: &str) -> PathBuf {
let deadline = Instant::now() + COMMAND_TIMEOUT;
loop {
if let Some(run_dir) = find_run_dir(storage_dir, run_id) {
return run_dir;
}
assert!(
Instant::now() < deadline,
"timed out waiting for run dir for {run_id}"
);
std::thread::sleep(Duration::from_millis(50));
}
}
async fn wait_for_http_ready(base_url: &str, child: &mut Child) {
let client = fabro_test::test_http_client();
let deadline = Instant::now() + Duration::from_secs(5);
loop {
match client.get(format!("{base_url}/health")).send().await {
Ok(response) if response.status().is_success() => return,
Ok(_) | Err(_) if Instant::now() < deadline => {
if let Some(status) = child.try_wait().expect("server process should poll") {
let mut stderr = Vec::new();
if let Some(stderr_pipe) = child.stderr.as_mut() {
stderr_pipe
.read_to_end(&mut stderr)
.expect("server stderr should be readable");
}
panic!(
"github-only server exited before becoming ready with status {status}\nstderr:\n{}",
String::from_utf8_lossy(&stderr)
);
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
Ok(response) => panic!("server at {base_url} was not ready: {}", response.status()),
Err(err) => panic!("server at {base_url} was not ready: {err}"),
}
}
}
async fn run_events(api_base_url: &str, run_id: &str, access_token: &str) -> Vec<EventEnvelope> {
let response = fabro_test::test_http_client()
.get(format!("{api_base_url}/api/v1/runs/{run_id}/events"))
.bearer_auth(access_token)
.send()
.await
.expect("event request should succeed");
let body: serde_json::Value = expect_reqwest_json(
response,
fabro_http::StatusCode::OK,
format!("GET /api/v1/runs/{run_id}/events"),
)
.await;
parse_event_envelopes(&body)
}
async fn wait_for_completed_events(
api_base_url: &str,
run_id: &str,
access_token: &str,
) -> Vec<EventEnvelope> {
let deadline = Instant::now() + COMMAND_TIMEOUT;
loop {
let events = run_events(api_base_url, run_id, access_token).await;
if events
.iter()
.any(|event| event.event.event_name() == "run.completed")
{
return events;
}
assert!(
Instant::now() < deadline,
"timed out waiting for run.completed for {run_id}"
);
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
#[tokio::test(flavor = "multi_thread")]
async fn github_only_server_dispatched_worker_succeeds_without_worker_auth_store() {
let context = test_context!();
let server = RunningGithubOnlyServer::start().await;
let target = server.target();
let access_token = issue_test_github_jwt(&server.api_base_url);
write_submitter_auth(&context.home_dir, &target, &access_token);
assert!(!server.worker_home.join("auth.json").exists());
assert!(!server.worker_home.join("auth.lock").exists());
let workflow = context.temp_dir.join("worker-auth.fabro");
write_probe_workflow(&workflow);
let run_id = unique_run_id();
let output = context
.run_cmd()
.args([
"--server",
&target,
"--run-id",
&run_id,
"--detach",
"--dry-run",
"--auto-approve",
"--no-retro",
"--sandbox",
"local",
workflow.to_str().unwrap(),
])
.output()
.expect("detached run should execute");
assert!(
output.status.success(),
"github-only detached run failed\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
assert_eq!(output_stdout(&output).trim(), run_id);
let _run_dir = wait_for_run_dir(&server.storage_dir, &run_id);
let events = wait_for_completed_events(&server.api_base_url, &run_id, &access_token).await;
assert!(events.iter().any(|event| {
event
.event
.actor
.as_ref()
.and_then(|actor| actor.display.as_deref())
== Some("system:worker")
}));
assert!(!server.worker_home.join("auth.json").exists());
assert!(!server.worker_home.join("auth.lock").exists());
server.shutdown().await;
}
#[test]
fn runner_rejects_bogus_worker_token_against_github_only_server() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
runtime.block_on(async {
let context = test_context!();
let server = RunningGithubOnlyServer::start().await;
let target = server.target();
let access_token = issue_test_github_jwt(&server.api_base_url);
write_submitter_auth(&context.home_dir, &target, &access_token);
let workflow = context.temp_dir.join("worker-auth-negative.fabro");
write_probe_workflow(&workflow);
let run_id = unique_run_id();
let create_output = context
.create_cmd()
.args([
"--server",
&target,
"--run-id",
&run_id,
"--dry-run",
"--auto-approve",
"--no-retro",
"--sandbox",
"local",
workflow.to_str().unwrap(),
])
.output()
.expect("remote create should execute");
assert!(
create_output.status.success(),
"github-only create failed\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&create_output.stdout),
String::from_utf8_lossy(&create_output.stderr)
);
assert_eq!(output_stdout(&create_output).trim(), run_id);
let run_dir = wait_for_run_dir(&server.storage_dir, &run_id);
let worker_root = tempfile::tempdir_in("/tmp").unwrap();
let worker_home = worker_root.path().join("fabro-home");
std::fs::create_dir_all(&worker_home).unwrap();
let auth_file = worker_root.path().join("missing").join("auth.json");
let bogus_token = issue_worker_token_for_run(&server.storage_dir, &unique_run_id());
let mut cmd = Command::new(env!("CARGO_BIN_EXE_fabro"));
apply_test_isolation(&mut cmd, worker_root.path());
cmd.env("FABRO_HOME", &worker_home);
cmd.env("FABRO_AUTH_FILE", &auth_file);
cmd.env("FABRO_WORKER_TOKEN", bogus_token);
cmd.args([
"__run-worker",
"--server",
&target,
"--run-dir",
run_dir.to_str().unwrap(),
"--run-id",
&run_id,
"--mode",
"start",
]);
cmd.stdin(Stdio::null());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
let output = cmd.output().expect("worker should execute");
assert!(
!output.status.success(),
"worker should fail with a bogus token\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
let stderr = output_stderr(&output);
assert!(
stderr.contains("403")
|| stderr.contains("Forbidden")
|| stderr.contains("Authentication required")
|| stderr.contains("Access denied"),
"{stderr}"
);
assert!(!auth_file.exists());
assert!(!auth_file.with_extension("lock").exists());
server.shutdown().await;
});
}

View file

@ -208,7 +208,19 @@ impl FromRequestParts<Arc<AppState>> for AuthorizeStageArtifact {
#[cfg(test)]
mod tests {
use std::sync::{Arc, Mutex as StdMutex};
use axum::http::header;
use axum::http::request::Parts;
use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use chrono::Duration as ChronoDuration;
use jsonwebtoken::{Algorithm, Header, decode};
use serde_json::json;
use tracing::field::{Field, Visit};
use tracing::{Event, Subscriber, subscriber};
use tracing_subscriber::layer::{Context, SubscriberExt};
use tracing_subscriber::{Layer, Registry};
use uuid::Uuid;
use super::{
@ -217,14 +229,152 @@ mod tests {
};
use crate::auth;
const TEST_SECRET: &[u8] = b"0123456789abcdef0123456789abcdef";
const OTHER_SECRET: &[u8] = b"fedcba9876543210fedcba9876543210";
fn keys(secret: &[u8]) -> WorkerTokenKeys {
WorkerTokenKeys::from_master_secret(secret).expect("worker keys should derive")
}
fn run_id() -> fabro_types::RunId {
"01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()
}
fn other_run_id() -> fabro_types::RunId {
"01ARZ3NDEKTSV4RRFFQ69G5FAW".parse().unwrap()
}
fn request_parts(authorization: Option<&str>) -> Parts {
let mut builder = axum::http::Request::builder();
if let Some(authorization) = authorization {
builder = builder.header(header::AUTHORIZATION, authorization);
}
let (parts, _) = builder.body(()).unwrap().into_parts();
parts
}
fn bearer_parts(token: &str) -> Parts {
request_parts(Some(&format!("Bearer {token}")))
}
fn wrong_scope_token(keys: &WorkerTokenKeys, run_id: &fabro_types::RunId) -> String {
let claims = WorkerTokenClaims {
iss: WORKER_TOKEN_ISSUER.to_string(),
iat: 1,
exp: u64::MAX / 2,
run_id: run_id.to_string(),
scope: "wrong:scope".to_string(),
jti: Uuid::new_v4().simple().to_string(),
};
jsonwebtoken::encode(&Header::new(Algorithm::HS256), &claims, &keys.encoding)
.expect("test token should encode")
}
fn expired_worker_token(keys: &WorkerTokenKeys, run_id: &fabro_types::RunId) -> String {
let claims = WorkerTokenClaims {
iss: WORKER_TOKEN_ISSUER.to_string(),
iat: 1,
exp: 2,
run_id: run_id.to_string(),
scope: WORKER_TOKEN_SCOPE.to_string(),
jti: Uuid::new_v4().simple().to_string(),
};
jsonwebtoken::encode(&Header::new(Algorithm::HS256), &claims, &keys.encoding)
.expect("expired test token should encode")
}
fn alg_none_token(run_id: &fabro_types::RunId) -> String {
let header = URL_SAFE_NO_PAD.encode(
serde_json::to_vec(&json!({
"alg": "none",
"typ": "JWT",
}))
.expect("jwt header should serialize"),
);
let payload = URL_SAFE_NO_PAD.encode(
serde_json::to_vec(&json!({
"iss": WORKER_TOKEN_ISSUER,
"iat": 1_u64,
"exp": u64::MAX / 2,
"run_id": run_id.to_string(),
"scope": WORKER_TOKEN_SCOPE,
"jti": Uuid::new_v4().simple().to_string(),
}))
.expect("jwt payload should serialize"),
);
format!("{header}.{payload}.")
}
fn issue_user_jwt() -> String {
let subject = auth::JwtSubject {
identity: fabro_types::IdpIdentity::new("https://github.com", "12345").unwrap(),
login: "octocat".to_string(),
name: "The Octocat".to_string(),
email: "octocat@example.com".to_string(),
avatar_url: "https://example.com/octocat.png".to_string(),
user_url: "https://github.com/octocat".to_string(),
auth_method: fabro_types::RunAuthMethod::Github,
};
let key = auth::derive_jwt_key(TEST_SECRET).expect("user jwt key should derive");
auth::issue(
&key,
"https://fabro.example",
&subject,
ChronoDuration::minutes(10),
)
}
#[derive(Debug)]
struct LogCapture {
target: String,
fields: Vec<(String, String)>,
}
#[derive(Default)]
struct LogCaptureVisitor {
fields: Vec<(String, String)>,
}
impl Visit for LogCaptureVisitor {
fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
self.fields
.push((field.name().to_string(), format!("{value:?}")));
}
}
struct LogCaptureLayer {
events: Arc<StdMutex<Vec<LogCapture>>>,
}
impl<S: Subscriber> Layer<S> for LogCaptureLayer {
fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
if event.metadata().target() != "worker_auth" {
return;
}
let mut visitor = LogCaptureVisitor::default();
event.record(&mut visitor);
self.events.lock().unwrap().push(LogCapture {
target: event.metadata().target().to_string(),
fields: visitor.fields,
});
}
}
fn capture_logs<T>(f: impl FnOnce() -> T) -> (T, Arc<StdMutex<Vec<LogCapture>>>) {
let events = Arc::new(StdMutex::new(Vec::<LogCapture>::new()));
let layer = LogCaptureLayer {
events: Arc::clone(&events),
};
let subscriber = Registry::default().with(layer);
let result = subscriber::with_default(subscriber, f);
(result, events)
}
#[test]
fn issue_worker_token_round_trips_claims() {
let run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap();
let keys = keys(b"0123456789abcdef0123456789abcdef");
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = issue_worker_token(&keys, &run_id).expect("worker token should issue");
let decoded = decode::<WorkerTokenClaims>(&token, &keys.decoding, &keys.validation)
@ -244,10 +394,9 @@ mod tests {
#[test]
fn worker_token_survives_key_rederivation() {
let secret = b"0123456789abcdef0123456789abcdef";
let run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap();
let first = keys(secret);
let second = keys(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::<WorkerTokenClaims>(&token, &second.decoding, &second.validation)
@ -258,9 +407,9 @@ mod tests {
#[test]
fn worker_token_fails_under_rotated_secret() {
let run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap();
let first = keys(b"0123456789abcdef0123456789abcdef");
let second = keys(b"fedcba9876543210fedcba9876543210");
let run_id = run_id();
let first = keys(TEST_SECRET);
let second = keys(OTHER_SECRET);
let token = issue_worker_token(&first, &run_id).expect("worker token should issue");
let err = decode::<WorkerTokenClaims>(&token, &second.decoding, &second.validation)
@ -273,38 +422,30 @@ mod tests {
#[test]
fn worker_key_is_distinct_from_user_jwt_key() {
let secret = b"0123456789abcdef0123456789abcdef";
let user_key = auth::derive_jwt_key(secret).expect("user key should derive");
let worker_key = auth::derive_worker_jwt_key(secret).expect("worker key should derive");
let user_key = auth::derive_jwt_key(TEST_SECRET).expect("user key should derive");
let worker_key =
auth::derive_worker_jwt_key(TEST_SECRET).expect("worker key should derive");
assert_ne!(user_key.as_bytes(), worker_key);
}
#[test]
fn authorize_worker_token_accepts_matching_run_id() {
let run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap();
let keys = keys(b"0123456789abcdef0123456789abcdef");
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = issue_worker_token(&keys, &run_id).expect("worker token should issue");
let request = axum::http::Request::builder()
.header(axum::http::header::AUTHORIZATION, format!("Bearer {token}"))
.body(())
.unwrap();
let (parts, _) = request.into_parts();
let parts = bearer_parts(&token);
assert!(authorize_worker_token(&parts, &run_id, &keys).unwrap());
}
#[test]
fn authorize_worker_token_rejects_cross_run_reuse() {
let run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap();
let other_run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAW".parse().unwrap();
let keys = keys(b"0123456789abcdef0123456789abcdef");
let run_id = run_id();
let other_run_id = other_run_id();
let keys = keys(TEST_SECRET);
let token = issue_worker_token(&keys, &other_run_id).expect("worker token should issue");
let request = axum::http::Request::builder()
.header(axum::http::header::AUTHORIZATION, format!("Bearer {token}"))
.body(())
.unwrap();
let (parts, _) = request.into_parts();
let parts = bearer_parts(&token);
let err = authorize_worker_token(&parts, &run_id, &keys)
.expect_err("mismatched run should reject");
@ -313,26 +454,155 @@ mod tests {
#[test]
fn authorize_worker_token_rejects_wrong_scope() {
let run_id: fabro_types::RunId = "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap();
let keys = keys(b"0123456789abcdef0123456789abcdef");
let claims = WorkerTokenClaims {
iss: WORKER_TOKEN_ISSUER.to_string(),
iat: 1,
exp: u64::MAX / 2,
run_id: run_id.to_string(),
scope: "wrong:scope".to_string(),
jti: Uuid::new_v4().simple().to_string(),
};
let token = jsonwebtoken::encode(&Header::new(Algorithm::HS256), &claims, &keys.encoding)
.expect("test token should encode");
let request = axum::http::Request::builder()
.header(axum::http::header::AUTHORIZATION, format!("Bearer {token}"))
.body(())
.unwrap();
let (parts, _) = request.into_parts();
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = wrong_scope_token(&keys, &run_id);
let parts = bearer_parts(&token);
let err =
authorize_worker_token(&parts, &run_id, &keys).expect_err("wrong scope should reject");
assert_eq!(err.status(), axum::http::StatusCode::FORBIDDEN);
}
#[test]
fn authorize_worker_token_falls_through_without_header() {
let run_id = run_id();
let keys = keys(TEST_SECRET);
let parts = request_parts(None);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
assert!(!result.unwrap());
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn authorize_worker_token_falls_through_for_user_jwt_without_worker_logs() {
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = issue_user_jwt();
let parts = bearer_parts(&token);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
assert!(!result.unwrap());
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn authorize_worker_token_falls_through_for_expired_token_without_worker_logs() {
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = expired_worker_token(&keys, &run_id);
let parts = bearer_parts(&token);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
assert!(!result.unwrap());
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn authorize_worker_token_falls_through_for_bad_signature_without_worker_logs() {
let run_id = run_id();
let signer = keys(OTHER_SECRET);
let verifier = keys(TEST_SECRET);
let token = issue_worker_token(&signer, &run_id).expect("worker token should issue");
let parts = bearer_parts(&token);
let (result, captured) =
capture_logs(|| authorize_worker_token(&parts, &run_id, &verifier));
assert!(!result.unwrap());
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn authorize_worker_token_falls_through_for_alg_none_without_worker_logs() {
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = alg_none_token(&run_id);
let parts = bearer_parts(&token);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
assert!(!result.unwrap());
assert!(captured.lock().unwrap().is_empty());
}
#[test]
fn authorize_worker_token_logs_acceptance() {
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = issue_worker_token(&keys, &run_id).expect("worker token should issue");
let parts = bearer_parts(&token);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
assert!(result.unwrap());
let events = captured.lock().unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].target, "worker_auth");
assert!(events[0]
.fields
.iter()
.any(|(field, value)| field == "message" && value.contains("worker token accepted")));
assert!(
events[0]
.fields
.iter()
.any(|(field, value)| field == "run_id" && value.contains(&run_id.to_string()))
);
assert!(
events[0]
.fields
.iter()
.any(|(field, value)| field == "jti" && !value.is_empty())
);
}
#[test]
fn authorize_worker_token_logs_run_id_mismatch() {
let run_id = run_id();
let other_run_id = other_run_id();
let keys = keys(TEST_SECRET);
let token = issue_worker_token(&keys, &other_run_id).expect("worker token should issue");
let parts = bearer_parts(&token);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
let err = result.expect_err("mismatched run should reject");
assert_eq!(err.status(), axum::http::StatusCode::FORBIDDEN);
let events = captured.lock().unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].target, "worker_auth");
assert!(
events[0]
.fields
.iter()
.any(|(field, value)| field == "reason" && value.contains("run_id_mismatch"))
);
}
#[test]
fn authorize_worker_token_logs_wrong_scope() {
let run_id = run_id();
let keys = keys(TEST_SECRET);
let token = wrong_scope_token(&keys, &run_id);
let parts = bearer_parts(&token);
let (result, captured) = capture_logs(|| authorize_worker_token(&parts, &run_id, &keys));
let err = result.expect_err("wrong scope should reject");
assert_eq!(err.status(), axum::http::StatusCode::FORBIDDEN);
let events = captured.lock().unwrap();
assert_eq!(events.len(), 1);
assert_eq!(events[0].target, "worker_auth");
assert!(
events[0]
.fields
.iter()
.any(|(field, value)| field == "reason" && value.contains("wrong_scope"))
);
}
}