diff --git a/swarm-authelia-bridge/src/introspect.rs b/swarm-authelia-bridge/src/introspect.rs index 70bb1e917..4616d9fc0 100644 --- a/swarm-authelia-bridge/src/introspect.rs +++ b/swarm-authelia-bridge/src/introspect.rs @@ -27,11 +27,15 @@ use std::time::Duration; use serde::Deserialize; -/// Generous relative to `swarm-nats-auth`'s 1.5s (that one is bounded by -/// the NATS server's own 2s `auth_callout` timeout; an HTTP request here -/// has no such external deadline to stay under), but still bounded — a -/// hung introspection call must not wedge a `CreateIdentity` job forever. -pub const INTROSPECTION_TIMEOUT: Duration = Duration::from_secs(5); +/// One attempt, no retry. authelia's introspection endpoint takes up to +/// ~19s when its host is overloaded, and this covers that. It must stay +/// below the controller's 30s bound on a whole bridge call +/// (`swarm-controller`'s `auth::HTTP_TIMEOUT`), or the controller gives up +/// on the bridge before the bridge has answered. A timeout is still +/// [`Verdict::Unavailable`], never an admission. +pub const INTROSPECTION_TIMEOUT: Duration = Duration::from_secs(20); + +const _: () = assert!(INTROSPECTION_TIMEOUT.as_secs() < 30); /// `active` decides whether to admit; `client_id` says as whom, which is /// what lets one verb be scoped to a single client. Same two fields, for the @@ -91,12 +95,33 @@ pub async fn introspect( client_id: &str, client_secret: &str, token: &str, +) -> Verdict { + introspect_within( + http, + url, + client_id, + client_secret, + token, + INTROSPECTION_TIMEOUT, + ) + .await +} + +/// [`introspect`] with the deadline as a parameter, so a test can exercise +/// the timeout without waiting out [`INTROSPECTION_TIMEOUT`]. +async fn introspect_within( + http: &reqwest::Client, + url: &str, + client_id: &str, + client_secret: &str, + token: &str, + timeout: Duration, ) -> Verdict { let resp = match http .post(url) .basic_auth(client_id, Some(client_secret)) .form(&[("token", token)]) - .timeout(INTROSPECTION_TIMEOUT) + .timeout(timeout) .send() .await { @@ -129,6 +154,94 @@ pub async fn introspect( mod tests { use super::*; + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + /// An HTTP server on loopback. Each connection is answered with + /// `status` and `body` after `delay`; a `None` delay holds the + /// connection open and never answers. + async fn serve(delay: Option, status: u16, body: &'static str) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}/introspect", listener.local_addr().unwrap()); + tokio::spawn(async move { + while let Ok((mut conn, _)) = listener.accept().await { + tokio::spawn(async move { + let mut buf = vec![0u8; 8192]; + let _ = conn.read(&mut buf).await; + let Some(delay) = delay else { + std::future::pending::<()>().await; + return; + }; + tokio::time::sleep(delay).await; + let resp = format!( + "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\n\ + content-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ); + let _ = conn.write_all(resp.as_bytes()).await; + }); + } + }); + url + } + + const ACTIVE: &str = r#"{"active":true,"client_id":"c"}"#; + + fn active() -> Verdict { + Verdict::Active { + client_id: Some("c".to_owned()), + } + } + + async fn ask(url: &str, timeout: Duration) -> Verdict { + introspect_within( + &reqwest::Client::new(), + url, + "bridge", + "secret", + "tok", + timeout, + ) + .await + } + + /// Presence control for the timeout tests: the same server shape admits + /// when it answers in time. + #[tokio::test] + async fn an_active_answer_within_the_deadline_admits() { + let url = serve(Some(Duration::ZERO), 200, ACTIVE).await; + assert_eq!(ask(&url, Duration::from_secs(5)).await, active()); + } + + /// A slow authelia is waited for: the answer's latency does not matter + /// as long as it beats the deadline. + #[tokio::test] + async fn a_slow_answer_inside_the_deadline_still_admits() { + let url = serve(Some(Duration::from_millis(600)), 200, ACTIVE).await; + assert_eq!(ask(&url, Duration::from_secs(5)).await, active()); + } + + /// Fail closed: no answer by the deadline is `Unavailable`, not an + /// admission. + #[tokio::test] + async fn no_answer_by_the_deadline_is_unavailable() { + let url = serve(None, 200, "").await; + assert_eq!( + ask(&url, Duration::from_millis(300)).await, + Verdict::Unavailable + ); + } + + /// An answer that arrives after the deadline is not used, even though + /// it would have admitted. + #[tokio::test] + async fn an_active_answer_after_the_deadline_is_unavailable() { + let url = serve(Some(Duration::from_millis(900)), 200, ACTIVE).await; + assert_eq!( + ask(&url, Duration::from_millis(300)).await, + Verdict::Unavailable + ); + } + #[test] fn only_active_true_deserializes_to_a_grant() { let yes: IntrospectionResponse = serde_json::from_str(r#"{"active":true}"#).unwrap(); diff --git a/swarm-authelia-bridge/src/main.rs b/swarm-authelia-bridge/src/main.rs index a68e77422..bfed6c38e 100644 --- a/swarm-authelia-bridge/src/main.rs +++ b/swarm-authelia-bridge/src/main.rs @@ -284,8 +284,10 @@ impl Refusal { /// 503 rather than 500 because the overwhelmingly common cause is /// transient by construction: this bridge's own secret is minted on /// authelia's first boot and delivered by a separate unit, so the - /// window where it is absent closes on its own. A client retrying is - /// doing the right thing. + /// window where it is absent closes on its own; an authelia that + /// answers slower than [`introspect::INTROSPECTION_TIMEOUT`] is + /// transient too. A client retrying is doing the right thing, and + /// `swarm-controller` does. fn status(self) -> StatusCode { match self { Refusal::Caller => StatusCode::UNAUTHORIZED, diff --git a/swarm-controller/src/auth.rs b/swarm-controller/src/auth.rs index ab5563953..0a4e2240a 100644 --- a/swarm-controller/src/auth.rs +++ b/swarm-controller/src/auth.rs @@ -127,14 +127,56 @@ impl AuthBridge { .await .context("minting a bearer token for swarm-authelia-bridge")?; - let response = self - .http - .post(format!("{}/requests", self.base_url)) + post_with_retry(&self.http, &self.base_url, &token, req, RETRY_BACKOFF).await + } +} + +/// Waits between attempts at one bridge call: three tries in all. +const RETRY_BACKOFF: &[std::time::Duration] = &[ + std::time::Duration::from_secs(1), + std::time::Duration::from_secs(4), +]; + +/// POST `req` to the bridge, retrying a `503` or a failure to connect after +/// each wait in `backoff`. +/// +/// `503` is the bridge saying it could not get an answer out of authelia, so +/// the token was never evaluated; the same call usually succeeds once +/// authelia recovers. `401` and `403` are verdicts on the token and are +/// final, as is any other status. A timeout after the connection was made +/// is not retried: the request may have been acted on, and a whole attempt +/// already took [`HTTP_TIMEOUT`]. Every bridge request is safe to repeat +/// regardless (idempotent ensure, or a read). +async fn post_with_retry( + http: &reqwest::Client, + base_url: &str, + token: &str, + req: &BridgeRequest, + backoff: &[std::time::Duration], +) -> Result { + let mut waits = backoff.iter(); + loop { + let sent = http + .post(format!("{base_url}/requests")) .bearer_auth(token) .json(req) .send() - .await - .map_err(|e| { + .await; + let (failure, retryable) = match sent { + Ok(response) => { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + if status.is_success() { + return serde_json::from_str(&body) + .context("parsing swarm-authelia-bridge's response"); + } + ( + anyhow::anyhow!("swarm-authelia-bridge refused the request ({status}): {body}"), + status == reqwest::StatusCode::SERVICE_UNAVAILABLE, + ) + } + Err(e) => { + let retryable = e.is_connect(); let what = if e.is_connect() && e.is_timeout() { format!( "calling swarm-authelia-bridge: connect timed out after {HTTP_CONNECT_TIMEOUT:?}" @@ -144,16 +186,16 @@ impl AuthBridge { } else { "calling swarm-authelia-bridge".to_owned() }; - anyhow::Error::new(e).context(what) - })?; - - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - if !status.is_success() { - anyhow::bail!("swarm-authelia-bridge refused the request ({status}): {body}"); + (anyhow::Error::new(e).context(what), retryable) + } + }; + match waits.next() { + Some(wait) if retryable => { + tracing::warn!(error = %failure, ?wait, "swarm-authelia-bridge call failed; retrying"); + tokio::time::sleep(*wait).await; + } + _ => return Err(failure), } - - serde_json::from_str(&body).context("parsing swarm-authelia-bridge's response") } } @@ -161,6 +203,106 @@ impl AuthBridge { mod tests { use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::Duration; + + const NO_WAIT: &[Duration] = &[Duration::ZERO, Duration::ZERO]; + + /// A bridge on loopback answering each request with the next entry of + /// `statuses` (the last one repeats), and counting the requests. + async fn bridge(statuses: &'static [u16]) -> (String, Arc) { + use axum::http::StatusCode; + let hits = Arc::new(AtomicUsize::new(0)); + let counter = hits.clone(); + let app = axum::Router::new().route( + "/requests", + axum::routing::post(move || { + let n = counter.fetch_add(1, Ordering::SeqCst); + let code = statuses[n.min(statuses.len() - 1)]; + async move { + let body = if code == 200 { + r#"{"status":"operators","operators":["mara"]}"# + } else { + r#"{"status":"error","message":"refused"}"# + }; + (StatusCode::from_u16(code).unwrap(), body) + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (url, hits) + } + + async fn call(url: &str, backoff: &[Duration]) -> Result { + post_with_retry( + &reqwest::Client::new(), + url, + "tok", + &BridgeRequest::ListOperators, + backoff, + ) + .await + } + + #[tokio::test] + async fn a_503_is_retried_until_the_bridge_answers() { + let (url, hits) = bridge(&[503, 503, 200]).await; + let got = call(&url, NO_WAIT).await.expect("third try succeeds"); + assert!(matches!(got, BridgeResponse::Operators { .. }), "{got:?}"); + assert_eq!(hits.load(Ordering::SeqCst), 3); + } + + #[tokio::test] + async fn a_persistent_503_fails_after_three_tries() { + let (url, hits) = bridge(&[503]).await; + let err = call(&url, NO_WAIT).await.expect_err("never recovers"); + assert!(format!("{err:#}").contains("503"), "{err:#}"); + assert_eq!(hits.load(Ordering::SeqCst), 3); + } + + /// 401 and 403 are verdicts on the token: one try each. + #[tokio::test] + async fn a_401_or_403_is_final() { + for code in [401, 403] { + let (url, hits) = bridge(match code { + 401 => &[401], + _ => &[403], + }) + .await; + let err = call(&url, NO_WAIT).await.expect_err("refused"); + assert!(format!("{err:#}").contains(&code.to_string()), "{err:#}"); + assert_eq!(hits.load(Ordering::SeqCst), 1, "status {code}"); + } + } + + #[tokio::test] + async fn a_connection_refusal_is_retried_then_fails() { + // Bind then drop, so the port refuses connections. + let port = { + let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + l.local_addr().unwrap().port() + }; + let started = tokio::time::Instant::now(); + let err = call( + &format!("http://127.0.0.1:{port}"), + &[Duration::from_millis(150); 2], + ) + .await + .expect_err("nothing listens"); + assert!( + format!("{err:#}").contains("calling swarm-authelia-bridge"), + "{err:#}" + ); + assert!( + started.elapsed() >= Duration::from_millis(300), + "both waits were taken: {:?}", + started.elapsed() + ); + } + /// The common case: no bridge wired up for this deployment. #[test] fn absent_url_is_not_an_error() {