use std::net::IpAddr; use std::path::Path; use std::time::Duration; use anyhow::{Context as _, Result, anyhow, bail}; use fabro_client::{ AuthEntry, AuthStore, Credential, OAuthSession, ServerTarget, TransportConnector, apply_bearer_token_auth, }; pub(crate) use fabro_client::{Client, RunEventStream}; use fabro_config::Storage; use fabro_config::bind::Bind; pub(crate) use fabro_types::RunProjection; use fabro_types::UserSettings; use fabro_util::dev_token; use tokio::time::{self, sleep}; use crate::args::ServerTargetArgs; use crate::commands::server::start; use crate::user_config::{self, cli_http_client_builder}; const SERVER_HEALTH_PROBE_TIMEOUT: Duration = Duration::from_millis(250); const CLI_CONTROL_PLANE_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); fn refreshable_oauth( target: &ServerTarget, credential: Option<&Credential>, ) -> Option { if matches!(credential, Some(Credential::OAuth(_))) { return Some(OAuthSession::new(target.clone(), AuthStore::default())); } None } pub(crate) async fn connect_server(storage_dir: &Path) -> Result { connect_local_api_client_bundle(storage_dir, &user_config::active_settings_path(None)).await } pub(crate) async fn connect_server_with_dev_token( storage_dir: &Path, dev_token: &str, ) -> Result { connect_local_api_client_bundle_with_dev_token( storage_dir, &user_config::active_settings_path(None), Some(dev_token), ) .await } pub(crate) async fn connect_server_target(target: &ServerTarget) -> Result { connect_target_api_client_bundle(target).await } pub(crate) async fn connect_server_target_with_bearer( target: &ServerTarget, bearer: &str, ) -> Result { build_client( target.clone(), Some(Credential::Worker(bearer.to_owned())), None, None, ) .await } pub(crate) async fn connect_server_with_settings( args: &ServerTargetArgs, settings: &UserSettings, storage_dir: &Path, base_config_path: &Path, ) -> Result { if let Some(target) = user_config::resolve_nondefault_server_target(args, settings)? { if let Some(path) = target.as_unix_socket_path() { return connect_managed_unix_socket_api_client_bundle( path, storage_dir, base_config_path, ) .await; } return connect_target_api_client_bundle(&target).await; } connect_local_api_client_bundle(storage_dir, base_config_path).await } async fn connect_managed_unix_socket_api_client_bundle( path: &Path, storage_dir: &Path, active_config_path: &Path, ) -> Result { let target = ServerTarget::unix_socket_path(path)?; let runtime_token_path = Storage::new(storage_dir) .runtime_directory() .dev_token_path(); let pre_spawn_credential = resolve_target_credential(&target)? .or_else(|| dev_token::read_dev_token_file(&runtime_token_path).map(Credential::DevToken)); let pre_spawn_bearer = pre_spawn_credential.as_ref().map(Credential::bearer_token); let (http_client, credential) = if let Ok(http_client) = try_connect_unix_socket_http_client(path, pre_spawn_bearer).await { (http_client, pre_spawn_credential) } else { start::ensure_server_running_on_socket(path, active_config_path, storage_dir) .await .with_context(|| format!("Failed to start fabro server for {}", path.display()))?; let post_spawn_credential = match resolve_target_credential(&target)? { Some(credential) => Some(credential), None => Some(Credential::DevToken( wait_for_runtime_dev_token(&runtime_token_path).await?, )), }; let post_spawn_bearer = post_spawn_credential.as_ref().map(Credential::bearer_token); let http_client = connect_unix_socket_http_client(path, post_spawn_bearer) .await .with_context(|| format!("Failed to connect to fabro server at {}", path.display()))?; (http_client, post_spawn_credential) }; let oauth_session = refreshable_oauth(&target, credential.as_ref()); build_client( target, credential, oauth_session, Some(("http://fabro".to_string(), http_client)), ) .await } async fn connect_local_api_client_bundle( storage_dir: &Path, active_config_path: &Path, ) -> Result { connect_local_api_client_bundle_with_dev_token(storage_dir, active_config_path, None).await } async fn connect_local_api_client_bundle_with_dev_token( storage_dir: &Path, active_config_path: &Path, bootstrap_dev_token: Option<&str>, ) -> Result { let bind = start::ensure_server_running_for_storage(storage_dir, active_config_path) .await .with_context(|| format!("Failed to start fabro server for {}", storage_dir.display()))?; match bind { Bind::Unix(path) => { let runtime_token_path = Storage::new(storage_dir) .runtime_directory() .dev_token_path(); let token = match bootstrap_dev_token { Some(token) => token.to_string(), None => wait_for_runtime_dev_token(&runtime_token_path).await?, }; let http_client = connect_unix_socket_http_client(&path, Some(&token)).await?; Ok(Client::builder() .transport("http://fabro", http_client) .request_timeout(CLI_CONTROL_PLANE_REQUEST_TIMEOUT) .connect() .await?) } Bind::Tcp(addr) => { let target = ServerTarget::http_url(format!("http://{addr}"))?; let credential = match bootstrap_dev_token { Some(token) => Some(Credential::DevToken(token.to_string())), None => resolve_target_credential(&target)?, }; let oauth_session = refreshable_oauth(&target, credential.as_ref()); build_client(target, credential, oauth_session, None).await } } } async fn connect_target_api_client_bundle(target: &ServerTarget) -> Result { let credential = resolve_target_credential(target)?; let oauth_session = refreshable_oauth(target, credential.as_ref()); build_client(target.clone(), credential, oauth_session, None).await } async fn build_client( target: ServerTarget, credential: Option, oauth_session: Option, transport: Option<(String, fabro_http::HttpClient)>, ) -> Result { let mut builder = Client::builder() .target(target.clone()) .transport_connector(build_cli_transport_connector(target)) .request_timeout(CLI_CONTROL_PLANE_REQUEST_TIMEOUT); if let Some((base_url, http_client)) = transport { builder = builder.transport(base_url, http_client); } if let Some(credential) = credential { builder = builder.credential(credential); } if let Some(oauth_session) = oauth_session { builder = builder.oauth_session(oauth_session); } builder.connect().await } fn build_cli_transport_connector(target: ServerTarget) -> TransportConnector { TransportConnector::new(move |bearer_token| { let target = target.clone(); async move { connect_cli_target_transport(&target, bearer_token.as_deref()) } }) } fn connect_cli_target_transport( target: &ServerTarget, bearer_token: Option<&str>, ) -> Result<(fabro_http::HttpClient, String)> { if let Some(api_url) = target.as_http_url() { let mut builder = cli_http_client_builder(); if should_bypass_proxy_for_http_target(api_url) { builder = builder.no_proxy(); } builder = match bearer_token { Some(token) => apply_bearer_token_auth(builder, token)?, None => builder, }; let http_client = builder.build()?; return Ok((http_client, api_url.to_string())); } let Some(path) = target.as_unix_socket_path() else { bail!("server target must be an http(s) URL or absolute Unix socket path"); }; let mut builder = cli_http_client_builder().unix_socket(path).no_proxy(); builder = match bearer_token { Some(token) => apply_bearer_token_auth(builder, token)?, None => builder, }; let http_client = builder .build() .context("Failed to build Unix-socket HTTP client for fabro server")?; Ok((http_client, "http://fabro".to_string())) } async fn wait_for_runtime_dev_token(path: &Path) -> Result { let deadline = std::time::Instant::now() + Duration::from_secs(5); while std::time::Instant::now() < deadline { if let Some(token) = dev_token::read_dev_token_file(path) { return Ok(token); } sleep(Duration::from_millis(50)).await; } bail!( "runtime dev token did not become available at {}", path.display() ); } fn build_authed_unix_socket_http_client( path: &Path, bearer_token: Option<&str>, ) -> Result { let builder = cli_http_client_builder().unix_socket(path).no_proxy(); let builder = if let Some(token) = bearer_token { apply_bearer_token_auth(builder, token)? } else { builder }; builder .build() .context("Failed to build Unix-socket HTTP client for fabro server") } fn build_unix_socket_probe_client(path: &Path) -> Result { cli_http_client_builder() .unix_socket(path) .no_proxy() .build() .context("Failed to build Unix-socket HTTP client for fabro server") } async fn try_connect_unix_socket_http_client( path: &Path, bearer_token: Option<&str>, ) -> Result { check_server_ready(&build_unix_socket_probe_client(path)?).await?; build_authed_unix_socket_http_client(path, bearer_token) } async fn connect_unix_socket_http_client( path: &Path, bearer_token: Option<&str>, ) -> Result { wait_for_server_ready(&build_unix_socket_probe_client(path)?).await?; build_authed_unix_socket_http_client(path, bearer_token) } fn resolve_target_credential_with_store( target: &ServerTarget, store: &AuthStore, now: chrono::DateTime, ) -> Result> { let Some(entry) = store.get(target)? else { return Ok(None); }; match entry { AuthEntry::DevToken(entry) => Ok(Some(Credential::DevToken(entry.token))), AuthEntry::OAuth(entry) if entry.access_token_expires_at > now || entry.refresh_token_expires_at > now => { Ok(Some(Credential::OAuth(entry))) } AuthEntry::OAuth(_) => Ok(None), } } fn resolve_target_credential(target: &ServerTarget) -> Result> { let store = AuthStore::default(); resolve_target_credential_with_store(target, &store, chrono::Utc::now()) } #[expect( clippy::disallowed_types, reason = "Proxy bypass classification parses a configured raw API target and does not log credential-bearing URLs." )] fn should_bypass_proxy_for_http_target(api_url: &str) -> bool { let Ok(url) = fabro_http::Url::parse(api_url) else { return false; }; let Some(host) = url.host_str() else { return false; }; if host.eq_ignore_ascii_case("localhost") { return true; } host.trim_matches(['[', ']']) .parse::() .is_ok_and(|ip| ip.is_loopback()) } async fn check_server_ready(http_client: &fabro_http::HttpClient) -> Result<()> { let response = match time::timeout( SERVER_HEALTH_PROBE_TIMEOUT, http_client.get("http://fabro/health").send(), ) .await { Ok(Ok(response)) => response, Ok(Err(err)) => return Err(anyhow!(err)), Err(_) => bail!("server health check timed out"), }; match response { response if response.status().is_success() => Ok(()), response => bail!("server health check returned status {}", response.status()), } } async fn wait_for_server_ready(http_client: &fabro_http::HttpClient) -> Result<()> { let deadline = std::time::Instant::now() + Duration::from_secs(5); let mut last_error = None; while std::time::Instant::now() < deadline { match check_server_ready(http_client).await { Ok(()) => return Ok(()), Err(err) => last_error = Some(err), } sleep(Duration::from_millis(50)).await; } Err(last_error.unwrap_or_else(|| anyhow!("server did not become ready in time"))) } #[cfg(test)] mod tests { use chrono::{Duration as ChronoDuration, Utc}; use fabro_client::{AuthEntry, DevTokenEntry, OAuthEntry, StoredSubject}; use httpmock::Method::{GET, POST}; use serde_json::json; use tokio::net::TcpListener; #[cfg(unix)] use tokio::net::UnixListener; use super::*; #[test] fn resolve_target_credential_uses_persisted_dev_token_entry() { let dir = tempfile::tempdir().unwrap(); let target = ServerTarget::http_url("http://127.0.0.1:32276").unwrap(); let store = AuthStore::new(dir.path().join("auth.json")); let token = "fabro_dev_abababababababababababababababababababababababababababababababab"; store .put( &target, AuthEntry::DevToken(DevTokenEntry { token: token.to_string(), logged_in_at: Utc::now(), }), ) .unwrap(); let credential = resolve_target_credential_with_store(&target, &store, Utc::now()).unwrap(); assert!(matches!(credential, Some(Credential::DevToken(found)) if found == token)); } #[cfg(unix)] #[tokio::test] async fn unix_socket_probe_times_out_when_peer_accepts_without_http_response() { let dir = tempfile::tempdir().unwrap(); let socket_path = dir.path().join("hung.sock"); let listener = UnixListener::bind(&socket_path).unwrap(); let server = tokio::spawn(async move { if let Ok((stream, _addr)) = listener.accept().await { let _stream = stream; sleep(Duration::from_secs(10)).await; } }); let result = time::timeout( Duration::from_millis(500), try_connect_unix_socket_http_client(&socket_path, None), ) .await; server.abort(); assert!( result.is_ok(), "Unix socket health probe should return its own timeout error instead of hanging" ); assert!(result.unwrap().is_err()); } #[tokio::test] async fn http_target_transport_times_out_when_peer_accepts_without_http_response() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); let server = tokio::spawn(async move { if let Ok((_stream, _addr)) = listener.accept().await { sleep(Duration::from_secs(10)).await; } }); let target = ServerTarget::http_url(format!("http://{addr}")).unwrap(); let client = connect_target_api_client_bundle(&target).await.unwrap(); let result = time::timeout(Duration::from_secs(5), client.get_health()).await; server.abort(); assert!( result.is_ok(), "HTTP target health check should return its own timeout error instead of hanging" ); assert!(result.unwrap().is_err()); } #[test] fn resolve_local_tcp_credential_uses_live_oauth_entry() { let dir = tempfile::tempdir().unwrap(); let target = ServerTarget::http_url("http://127.0.0.1:32276").unwrap(); let store = AuthStore::new(dir.path().join("auth.json")); let now = Utc::now(); store .put( &target, oauth_entry( now + ChronoDuration::minutes(5), now - ChronoDuration::minutes(1), ), ) .unwrap(); let credential = resolve_target_credential_with_store(&target, &store, now).unwrap(); assert!(matches!(credential, Some(Credential::OAuth(_)))); } #[test] fn resolve_local_tcp_credential_uses_refreshable_oauth_entry() { let dir = tempfile::tempdir().unwrap(); let target = ServerTarget::http_url("http://127.0.0.1:32276").unwrap(); let store = AuthStore::new(dir.path().join("auth.json")); let now = Utc::now(); store .put( &target, oauth_entry( now - ChronoDuration::minutes(1), now + ChronoDuration::minutes(5), ), ) .unwrap(); let credential = resolve_target_credential_with_store(&target, &store, now).unwrap(); assert!(matches!(credential, Some(Credential::OAuth(_)))); } #[test] fn resolve_local_tcp_credential_ignores_expired_oauth_entry() { let dir = tempfile::tempdir().unwrap(); let target = ServerTarget::http_url("http://127.0.0.1:32276").unwrap(); let store = AuthStore::new(dir.path().join("auth.json")); let now = Utc::now(); store .put( &target, oauth_entry( now - ChronoDuration::minutes(5), now - ChronoDuration::minutes(1), ), ) .unwrap(); let credential = resolve_target_credential_with_store(&target, &store, now).unwrap(); assert!(credential.is_none()); } #[test] fn bypasses_proxy_for_loopback_http_targets() { assert!(should_bypass_proxy_for_http_target( "http://127.0.0.1:32276" )); assert!(should_bypass_proxy_for_http_target("http://[::1]:32276")); assert!(should_bypass_proxy_for_http_target( "http://localhost:32276" )); assert!(!should_bypass_proxy_for_http_target( "https://fabro.example.com" )); assert!(!should_bypass_proxy_for_http_target( "http://fabro.example.com" )); } #[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, refresh_token_expires_at: chrono::DateTime, ) -> AuthEntry { AuthEntry::OAuth(OAuthEntry { access_token: "access-token".to_string(), access_token_expires_at, refresh_token: "refresh-token".to_string(), refresh_token_expires_at, subject: StoredSubject { idp_issuer: "https://github.com/login/oauth".to_string(), idp_subject: "subject-123".to_string(), login: "octocat".to_string(), name: "Octo Cat".to_string(), email: "octocat@example.com".to_string(), }, logged_in_at: Utc::now(), }) } }