From 71e7fd8b20d3242a0c02e74eebb6a68ca5acab1e Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 12 Oct 2026 00:36:48 +0200 Subject: [PATCH] swarmctl: serve login callbacks concurrently, retry a refused token once, test the refresh lock and the socket path The loopback listener served one connection at a time with no read timeout, so a connection that never sent a request (a browser preconnect, any local process) held up the real form_post until the 5-minute wait ran out. Each connection now runs on its own task and the first final answer wins. The callback body is capped at 16 KiB; over the cap it reads as empty and can't complete the login. A request the gateway answers 401 now triggers one refresh (under the cache lock, skipped if another process already replaced the refused token) and one retry; a second 401 says to run `swarmctl login`. The login cache directory is a `Swarm` field so tests can point it at a scratch directory. New tests: an idle connection does not block the real callback; an oversized body is refused without ending the wait; two concurrent refreshers refresh once and share the token; a refused token is refreshed once and retried; a second 401 says to log in without a further retry; with no `--swarm`, the verbs use the controller socket. --- swarmctl/README.md | 5 +- swarmctl/src/login.rs | 175 +++++++++++++++-------- swarmctl/src/main.rs | 72 +++++++++- swarmctl/src/remote.rs | 313 +++++++++++++++++++++++++++++++++++++---- 4 files changed, 477 insertions(+), 88 deletions(-) diff --git a/swarmctl/README.md b/swarmctl/README.md index 7b2fbfbb..88c2a547 100644 --- a/swarmctl/README.md +++ b/swarmctl/README.md @@ -188,8 +188,9 @@ $ swarmctl schedule list `login` opens the swarm's SSO login in a browser and receives the result on a loopback port, so run it on the machine whose browser you use. swarmctl keeps the login, mode 0600, under `$XDG_STATE_HOME/swarmctl/` (default -`~/.local/state`) and refreshes it as it's used. When it can't be refreshed, -any verb says to run `swarmctl login` again. To log out, delete the file. +`~/.local/state`) and refreshes it as it's used, and once more if the gateway +refuses it. When it can't be refreshed, any verb says to run `swarmctl login` +again. To log out, delete the file. A swarm whose gateway serves a certificate from the swarm's own CA needs that CA: `--ca-file` or `SWARMCTL_CA_FILE`. The `user` verbs edit the local diff --git a/swarmctl/src/login.rs b/swarmctl/src/login.rs index 8ad55cbc..ac05e031 100644 --- a/swarmctl/src/login.rs +++ b/swarmctl/src/login.rs @@ -7,12 +7,12 @@ //! listens on. The registered redirect URI carries no port; authelia matches a //! loopback redirect on any port (RFC 8252 §7.3). -use std::sync::{Arc, Mutex}; +use std::sync::Arc; use std::time::Duration; use anyhow::{Context as _, Result, bail}; use base64::Engine as _; -use http_body_util::{BodyExt as _, Full}; +use http_body_util::{BodyExt as _, Full, Limited}; use hyper_util::rt::TokioIo; use serde::Deserialize; use sha2::Digest as _; @@ -23,6 +23,8 @@ use crate::remote::{CLIENT_ID, Cached, Swarm, lock, oauth_json, token_request, w /// authelia's PAR `request_uri` lives this long, so waiting longer is futile. const WAIT: Duration = Duration::from_mins(5); const CALLBACK_PATH: &str = "/callback"; +/// A `form_post` answer is a few hundred bytes. +const MAX_CALLBACK_BODY: usize = 16 * 1024; /// Run `swarmctl login`. pub(crate) fn login(swarm: &Swarm) -> Result<()> { @@ -30,7 +32,7 @@ pub(crate) fn login(swarm: &Swarm) -> Result<()> { .enable_all() .build() .context("starting a tokio runtime for the login")?; - let path = swarm.cache_path()?; + let path = swarm.cache_path(); rt.block_on(async { let http = swarm.http()?; let auth = discover_auth(&http, swarm).await?; @@ -213,52 +215,74 @@ fn callback(method: &hyper::Method, path: &str, body: &[u8], state: &str, auth: } } -/// Serve the loopback listener until this login's answer arrives. +/// Serve the loopback listener until this login's answer arrives. Each +/// connection runs on its own task, so one that never sends a request can't +/// hold up the one carrying the answer; the first final answer wins. async fn receive_code(listener: &TcpListener, state: &str, auth: &str) -> Result { + let expected = Arc::new((state.to_owned(), auth.to_owned())); + let (answers, mut answer) = tokio::sync::mpsc::unbounded_channel(); loop { - let (stream, _) = listener.accept().await.context("accepting the callback")?; - let outcome: Arc>> = Arc::default(); - let service = hyper::service::service_fn(|req: hyper::Request| { - let outcome = Arc::clone(&outcome); - async move { - let (parts, body) = req.into_parts(); - let body = body.collect().await?.to_bytes(); - let judged = callback(&parts.method, parts.uri.path(), &body, state, auth); - let (status, text) = match &judged { - Callback::Ignored(status, text) => (*status, *text), - Callback::Code(_) => ( - hyper::StatusCode::OK, - "swarmctl is logged in. You can close this tab.", - ), - Callback::Refused(_) => ( - hyper::StatusCode::BAD_REQUEST, - "The login failed; swarmctl prints why.", - ), - }; - if !matches!(judged, Callback::Ignored(..)) { - *outcome.lock().expect("callback lock") = Some(judged); - } - Ok::<_, hyper::Error>( - hyper::Response::builder() - .status(status) - .header(hyper::header::CONTENT_TYPE, "text/plain; charset=utf-8") - .body(Full::new(bytes::Bytes::from_static(text.as_bytes()))) - .expect("static response"), - ) + tokio::select! { + accepted = listener.accept() => { + let (stream, _) = accepted.context("accepting the callback")?; + tokio::spawn(serve_callback(stream, Arc::clone(&expected), answers.clone())); } - }); - let _ = hyper::server::conn::http1::Builder::new() - .keep_alive(false) - .serve_connection(TokioIo::new(stream), service) - .await; - match outcome.lock().expect("callback lock").take() { - Some(Callback::Code(code)) => return Ok(code), - Some(Callback::Refused(why)) => bail!("{why}"), - Some(Callback::Ignored(..)) | None => {} + Some(outcome) = answer.recv() => return outcome.map_err(anyhow::Error::msg), } } } +/// Answer one loopback connection, sending a code or a refusal to `answers`. +async fn serve_callback( + stream: tokio::net::TcpStream, + expected: Arc<(String, String)>, + answers: tokio::sync::mpsc::UnboundedSender>, +) { + let service = hyper::service::service_fn(move |req: hyper::Request| { + let expected = Arc::clone(&expected); + let answers = answers.clone(); + async move { + let (parts, body) = req.into_parts(); + // Over the cap reads as an empty body, which carries no state. + let body = Limited::new(body, MAX_CALLBACK_BODY) + .collect() + .await + .map(http_body_util::Collected::to_bytes) + .unwrap_or_default(); + let (state, auth) = &*expected; + let (status, text) = match callback(&parts.method, parts.uri.path(), &body, state, auth) + { + Callback::Ignored(status, text) => (status, text), + Callback::Code(code) => { + let _ = answers.send(Ok(code)); + ( + hyper::StatusCode::OK, + "swarmctl is logged in. You can close this tab.", + ) + } + Callback::Refused(why) => { + let _ = answers.send(Err(why)); + ( + hyper::StatusCode::BAD_REQUEST, + "The login failed; swarmctl prints why.", + ) + } + }; + Ok::<_, std::convert::Infallible>( + hyper::Response::builder() + .status(status) + .header(hyper::header::CONTENT_TYPE, "text/plain; charset=utf-8") + .body(Full::new(bytes::Bytes::from_static(text.as_bytes()))) + .expect("static response"), + ) + } + }); + let _ = hyper::server::conn::http1::Builder::new() + .keep_alive(false) + .serve_connection(TokioIo::new(stream), service) + .await; +} + #[cfg(test)] mod tests { use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; @@ -360,28 +384,67 @@ mod tests { let addr = listener.local_addr().expect("addr"); let server = tokio::spawn(async move { receive_code(&listener, "s1", AUTH).await }); - let send = |body: &'static str| async move { - let mut s = tokio::net::TcpStream::connect(addr).await.expect("connect"); - let req = format!( - "POST /callback HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: \ - application/x-www-form-urlencoded\r\nContent-Length: {}\r\n\r\n{body}", - body.len() - ); - s.write_all(req.as_bytes()).await.expect("write"); - let mut answer = String::new(); - s.read_to_string(&mut answer).await.expect("read"); - answer - }; assert!( - send("code=stale&state=s0") + post_to(addr, "code=stale&state=s0") .await .starts_with("HTTP/1.1 400") ); assert!( - send("code=fresh&state=s1") + post_to(addr, "code=fresh&state=s1") .await .starts_with("HTTP/1.1 200") ); assert_eq!(server.await.expect("join").expect("code"), "fresh"); } + + /// One form POST to the loopback listener; the whole answer as text. + async fn post_to(addr: std::net::SocketAddr, body: &str) -> String { + let mut s = tokio::net::TcpStream::connect(addr).await.expect("connect"); + let req = format!( + "POST /callback HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: \ + application/x-www-form-urlencoded\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + ); + s.write_all(req.as_bytes()).await.expect("write"); + let mut answer = String::new(); + s.read_to_string(&mut answer).await.expect("read"); + answer + } + + /// A connection that never sends a request (a browser's preconnect, any + /// local process) must not hold up the one carrying the answer. + #[tokio::test] + async fn an_idle_connection_does_not_block_the_real_callback() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let server = tokio::spawn(async move { receive_code(&listener, "s1", AUTH).await }); + + let _idle = tokio::net::TcpStream::connect(addr).await.expect("connect"); + tokio::time::sleep(Duration::from_millis(100)).await; + let answer = + tokio::time::timeout(Duration::from_secs(5), post_to(addr, "code=c1&state=s1")) + .await + .expect("the callback is answered while another connection idles"); + assert!(answer.starts_with("HTTP/1.1 200"), "{answer}"); + assert_eq!(server.await.expect("join").expect("code"), "c1"); + } + + /// A body past the cap is not read, so it can't complete the login; the + /// wait goes on for a real answer. + #[tokio::test] + async fn an_oversized_callback_body_is_refused_without_ending_the_wait() { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("addr"); + let server = tokio::spawn(async move { receive_code(&listener, "s1", AUTH).await }); + + let pad = "a".repeat(MAX_CALLBACK_BODY); + let big = post_to(addr, &format!("code=big&state=s1&pad={pad}")).await; + assert!(big.starts_with("HTTP/1.1 400"), "{big}"); + assert!( + post_to(addr, "code=c1&state=s1") + .await + .starts_with("HTTP/1.1 200") + ); + assert_eq!(server.await.expect("join").expect("code"), "c1"); + } } diff --git a/swarmctl/src/main.rs b/swarmctl/src/main.rs index 717ce805..84edc540 100644 --- a/swarmctl/src/main.rs +++ b/swarmctl/src/main.rs @@ -83,7 +83,8 @@ impl RemoteArgs { .ca_file .clone() .or_else(|| std::env::var_os("SWARMCTL_CA_FILE").map(PathBuf::from)); - url.map(|u| remote::Swarm::parse(&u, ca_file)).transpose() + url.map(|u| remote::Swarm::parse(&u, ca_file, remote::state_dir()?)) + .transpose() } /// The swarm when one is named, the controller's socket otherwise. @@ -991,6 +992,75 @@ fn write_atomic(path: &Path, contents: &str) -> Result<()> { mod tests { use super::*; + /// Root on the controller host, with no `--swarm` and no login: the verbs + /// still go over the unix socket, unchanged by the remote path. + #[test] + fn with_no_swarm_named_the_verbs_use_the_controller_socket() { + assert!( + std::env::var_os("SWARMCTL_SWARM").is_none(), + "this test needs SWARMCTL_SWARM unset" + ); + let dir = std::env::temp_dir().join(format!("swarmctl-socket-{}", std::process::id())); + let _ = fs::remove_dir_all(&dir); + fs::create_dir_all(&dir).expect("dir"); + let socket = dir.join("controller.sock"); + let listener = std::os::unix::net::UnixListener::bind(&socket).expect("bind"); + listener.set_nonblocking(true).expect("nonblocking"); + let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let server_seen = std::sync::Arc::clone(&seen); + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + rt.block_on(async move { + let listener = tokio::net::UnixListener::from_std(listener).expect("listener"); + let (stream, _) = listener.accept().await.expect("accept"); + let service = hyper::service::service_fn( + move |req: hyper::Request| { + server_seen.lock().expect("seen").push(format!( + "{} {} auth={}", + req.method(), + req.uri(), + req.headers().contains_key(hyper::header::AUTHORIZATION) + )); + async { + Ok::<_, std::convert::Infallible>(hyper::Response::new( + http_body_util::Full::new(bytes::Bytes::from_static(b"[]")), + )) + } + }, + ); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(hyper_util::rt::TokioIo::new(stream), service) + .await; + }); + }); + + let cli = Cli::try_parse_from([ + "swarmctl", + "schedule", + "list", + "--controller-socket", + socket.to_str().expect("utf-8"), + ]) + .expect("parses"); + let Verb::Schedule { + command: ScheduleVerb::List(args), + } = cli.command + else { + panic!("parsed as another verb"); + }; + let target = args.socket.resolve(&cli.remote).expect("resolves"); + assert!(matches!(&target, Target::Socket(p) if *p == socket)); + schedule::list(&target, None).expect("lists over the socket"); + assert_eq!( + *seen.lock().expect("seen"), + ["GET /api/schedules auth=false"] + ); + fs::remove_dir_all(&dir).expect("cleanup"); + } + /// clap's own consistency checks over the whole tree — a duplicated /// long flag or a malformed `value_name` panics at parse time in /// production and is otherwise only found by running the binary. diff --git a/swarmctl/src/remote.rs b/swarmctl/src/remote.rs index d3c2f7a2..44a31614 100644 --- a/swarmctl/src/remote.rs +++ b/swarmctl/src/remote.rs @@ -39,10 +39,12 @@ pub(crate) struct Swarm { pub origin: url::Url, /// A PEM CA certificate to trust beside the system roots. pub ca_file: Option, + /// Where the login cache lives: `$XDG_STATE_HOME/swarmctl` by default. + pub state_dir: PathBuf, } impl Swarm { - pub(crate) fn parse(url: &str, ca_file: Option) -> Result { + pub(crate) fn parse(url: &str, ca_file: Option, state_dir: PathBuf) -> Result { let parsed = url::Url::parse(url).with_context(|| format!("--swarm {url:?} is not a URL"))?; if parsed.scheme() != "https" || parsed.host_str().is_none() { @@ -50,7 +52,11 @@ impl Swarm { } let origin = url::Url::parse(&parsed.origin().ascii_serialization()) .context("re-parsing the swarm's origin")?; - Ok(Self { origin, ca_file }) + Ok(Self { + origin, + ca_file, + state_dir, + }) } /// What the token must name for the gateway to admit it at `/api/`. @@ -73,19 +79,13 @@ impl Swarm { builder.build().context("building the HTTPS client") } - pub(crate) fn cache_path(&self) -> Result { - let state = std::env::var_os("XDG_STATE_HOME") - .map(PathBuf::from) - .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state"))) - .context( - "neither XDG_STATE_HOME nor HOME is set, so there is nowhere to keep the login", - )?; + pub(crate) fn cache_path(&self) -> PathBuf { let host = self.origin.host_str().unwrap_or_default(); let name = match self.origin.port() { Some(port) => format!("{host}_{port}.json"), None => format!("{host}.json"), }; - Ok(state.join("swarmctl").join(name)) + self.state_dir.join(name) } fn not_logged_in(&self) -> String { @@ -96,6 +96,8 @@ impl Swarm { } /// One request to `uri` (a path under the origin) with the cached login. + /// A 401 is answered by one refresh and one retry; a second 401 means the + /// login itself is gone. pub(crate) async fn request( &self, method: hyper::Method, @@ -103,19 +105,35 @@ impl Swarm { body: Option>, what: &str, ) -> Result<(hyper::StatusCode, bytes::Bytes)> { - let http = self.http()?; - let token = self.access_token(&http).await?; - let url = self.origin.join(uri).context("joining the request path")?; - let mut req = http.request(method, url).bearer_auth(token); - if let Some(body) = body { - req = req - .header(reqwest::header::CONTENT_TYPE, "application/json") - .body(body); - } - let resp = req - .send() + self.request_with(&self.http()?, method, uri, body, what) .await - .with_context(|| format!("sending the {what} request to {}", self.origin))?; + } + + async fn request_with( + &self, + http: &reqwest::Client, + method: hyper::Method, + uri: &str, + body: Option>, + what: &str, + ) -> Result<(hyper::StatusCode, bytes::Bytes)> { + let url = self.origin.join(uri).context("joining the request path")?; + let send = |token: String| { + let mut req = http.request(method.clone(), url.clone()).bearer_auth(token); + if let Some(body) = &body { + req = req + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(body.clone()); + } + req.send() + }; + let context = || format!("sending the {what} request to {}", self.origin); + let token = self.access_token(http, None).await?; + let mut resp = send(token.clone()).await.with_context(context)?; + if resp.status() == reqwest::StatusCode::UNAUTHORIZED { + let fresh = self.access_token(http, Some(&token)).await?; + resp = send(fresh).await.with_context(context)?; + } let status = resp.status(); if status == reqwest::StatusCode::UNAUTHORIZED { bail!( @@ -135,17 +153,20 @@ impl Swarm { Ok((status, body)) } - /// The cached access token, refreshed first if it is about to expire. - async fn access_token(&self, http: &reqwest::Client) -> Result { - let path = self.cache_path()?; + /// The cached access token, refreshed first if it is about to expire or is + /// `rejected`, the one the gateway just refused. + async fn access_token(&self, http: &reqwest::Client, rejected: Option<&str>) -> Result { + let path = self.cache_path(); let now = chrono::Utc::now().timestamp(); + let due = |c: &Cached| needs_refresh(c, now) || rejected == Some(c.access_token.as_str()); let cached = read_cache(&path)?.with_context(|| self.not_logged_in())?; - if !needs_refresh(&cached, now) { + if !due(&cached) { return Ok(cached.access_token); } let _lock = lock(&path)?; + // Another process may have refreshed while this one waited for the lock. let cached = read_cache(&path)?.with_context(|| self.not_logged_in())?; - if !needs_refresh(&cached, now) { + if !due(&cached) { return Ok(cached.access_token); } let token = token_request( @@ -165,6 +186,15 @@ impl Swarm { } } +/// `$XDG_STATE_HOME/swarmctl`, else `~/.local/state/swarmctl`. +pub(crate) fn state_dir() -> Result { + std::env::var_os("XDG_STATE_HOME") + .map(PathBuf::from) + .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".local/state"))) + .map(|state| state.join("swarmctl")) + .context("neither XDG_STATE_HOME nor HOME is set, so there is nowhere to keep the login") +} + /// The token cache: one file per swarm, mode 0600, in a 0700 directory. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct Cached { @@ -288,6 +318,7 @@ pub(crate) fn lock(path: &Path) -> Result { #[cfg(test)] mod tests { use std::os::unix::fs::PermissionsExt as _; + use std::sync::{Arc, Mutex}; use super::*; @@ -345,8 +376,232 @@ mod tests { #[test] fn the_audience_is_the_api_prefix_of_the_origin() { - let swarm = Swarm::parse("https://swarm.example.test/some/page", None).expect("parses"); + let swarm = Swarm::parse("https://swarm.example.test/some/page", None, "/s".into()) + .expect("parses"); assert_eq!(swarm.audience(), "https://swarm.example.test/api/"); - assert!(Swarm::parse("http://swarm.example.test", None).is_err()); + assert!(Swarm::parse("http://swarm.example.test", None, "/s".into()).is_err()); + } + + /// authelia's token endpoint and the gateway's `/api/`, as far as these + /// tests need them: a refresh rotates both tokens, a used refresh token + /// revokes the login (as authelia does), and `/api/` admits only the + /// current access token. + struct Mock { + refresh: Option, + access: Option, + refreshes: u32, + api_calls: u32, + api_refuses_all: bool, + } + + fn mock() -> Arc> { + Arc::new(Mutex::new(Mock { + refresh: Some("rt-0".into()), + access: Some("at-0".into()), + refreshes: 0, + api_calls: 0, + api_refuses_all: false, + })) + } + + /// Serve `mock` on a loopback port from its own thread, so callers on + /// other threads can block on the cache lock without stalling it. + fn serve(mock: &Arc>) -> url::Url { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind"); + let addr = listener.local_addr().expect("addr"); + listener.set_nonblocking(true).expect("nonblocking"); + let mock = Arc::clone(mock); + std::thread::spawn(move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + rt.block_on(async move { + let listener = tokio::net::TcpListener::from_std(listener).expect("listener"); + loop { + let (stream, _) = listener.accept().await.expect("accept"); + let mock = Arc::clone(&mock); + tokio::spawn(async move { + let service = hyper::service::service_fn(move |req| { + let mock = Arc::clone(&mock); + async move { Ok::<_, std::convert::Infallible>(answer(&mock, req).await) } + }); + let _ = hyper::server::conn::http1::Builder::new() + .serve_connection(hyper_util::rt::TokioIo::new(stream), service) + .await; + }); + } + }); + }); + url::Url::parse(&format!("http://{addr}/")).expect("url") + } + + async fn answer( + mock: &Mutex, + req: hyper::Request, + ) -> hyper::Response> { + use http_body_util::BodyExt as _; + let reply = |status: u16, body: serde_json::Value| { + hyper::Response::builder() + .status(status) + .body(http_body_util::Full::new(body.to_string().into())) + .expect("response") + }; + let path = req.uri().path().to_owned(); + let bearer = req + .headers() + .get(hyper::header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .map(str::to_owned); + let body = req.into_body().collect().await.expect("body").to_bytes(); + if path == "/api/oidc/token" { + // Wide enough that two unlocked refreshers are both in flight. + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let presented = url::form_urlencoded::parse(&body) + .find(|(k, _)| k == "refresh_token") + .map(|(_, v)| v.into_owned()); + let mut m = mock.lock().expect("mock"); + if presented.is_none() || presented != m.refresh { + m.refresh = None; + m.access = None; + return reply(400, serde_json::json!({ "error": "invalid_grant" })); + } + m.refreshes += 1; + m.access = Some(format!("at-{}", m.refreshes)); + m.refresh = Some(format!("rt-{}", m.refreshes)); + return reply( + 200, + serde_json::json!({ + "access_token": m.access, "refresh_token": m.refresh, "expires_in": 3600, + }), + ); + } + let mut m = mock.lock().expect("mock"); + m.api_calls += 1; + if !m.api_refuses_all && bearer.is_some() && bearer == m.access { + reply(200, serde_json::json!({ "ok": true })) + } else { + reply(401, serde_json::json!({})) + } + } + + fn scratch(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("swarmctl-{name}-{}", std::process::id())); + let _ = fs::remove_dir_all(&dir); + dir + } + + /// A swarm whose API and authelia are both `origin`, with a cached login. + fn logged_in(origin: &url::Url, dir: &Path, access_token: &str, expires_at: i64) -> Swarm { + let swarm = Swarm { + origin: origin.clone(), + ca_file: None, + state_dir: dir.to_owned(), + }; + let cache = Cached { + auth: origin.as_str().trim_end_matches('/').to_owned(), + access_token: access_token.into(), + refresh_token: "rt-0".into(), + expires_at, + }; + write_cache(&swarm.cache_path(), &cache).expect("cache"); + swarm + } + + /// Plain HTTP only: no system roots, which the build sandbox doesn't have. + fn client() -> reqwest::Client { + reqwest::Client::builder() + .tls_certs_only([]) + .build() + .expect("client") + } + + fn block_on(fut: impl std::future::Future) -> T { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime") + .block_on(fut) + } + + /// Two refreshers on one expired cache, as two `swarmctl` processes would + /// be: one refresh, and both get the token it issued. Without the lock + /// and the reread under it, the second presents the used refresh token + /// and authelia revokes the login. + #[test] + fn concurrent_refreshers_refresh_once_and_share_the_token() { + let mock = mock(); + let origin = serve(&mock); + let dir = scratch("concurrent"); + logged_in(&origin, &dir, "at-0", 0); + + let start = Arc::new(std::sync::Barrier::new(2)); + let refreshers: Vec<_> = (0..2) + .map(|_| { + let swarm = Swarm { + origin: origin.clone(), + ca_file: None, + state_dir: dir.clone(), + }; + let start = Arc::clone(&start); + std::thread::spawn(move || { + let http = client(); + start.wait(); + block_on(swarm.access_token(&http, None)) + }) + }) + .collect(); + let tokens: Vec = refreshers + .into_iter() + .map(|r| r.join().expect("thread").expect("refreshed")) + .collect(); + + let m = mock.lock().expect("mock"); + assert_eq!(m.refreshes, 1); + assert_eq!(tokens, ["at-1", "at-1"]); + assert_eq!(m.access.as_deref(), Some("at-1"), "the login is still live"); + fs::remove_dir_all(&dir).expect("cleanup"); + } + + /// The clock says the token is good and the gateway refuses it: refresh + /// once and retry. + #[test] + fn a_refused_token_is_refreshed_once_and_the_request_retried() { + let mock = mock(); + let origin = serve(&mock); + let dir = scratch("retry"); + let swarm = logged_in(&origin, &dir, "at-revoked", i64::MAX / 2); + + let (status, _) = + block_on(swarm.request_with(&client(), hyper::Method::GET, "/api/x", None, "x")) + .expect("the retry succeeds"); + assert_eq!(status, hyper::StatusCode::OK); + let m = mock.lock().expect("mock"); + assert_eq!((m.refreshes, m.api_calls), (1, 2)); + drop(m); + let cached = read_cache(&swarm.cache_path()) + .expect("read") + .expect("cache"); + assert_eq!(cached.access_token, "at-1"); + fs::remove_dir_all(&dir).expect("cleanup"); + } + + /// A 401 after the refresh means the login itself is gone: say so, once. + #[test] + fn a_second_401_says_to_log_in_and_does_not_retry_again() { + let mock = mock(); + mock.lock().expect("mock").api_refuses_all = true; + let origin = serve(&mock); + let dir = scratch("refused"); + let swarm = logged_in(&origin, &dir, "at-0", i64::MAX / 2); + + let err = block_on(swarm.request_with(&client(), hyper::Method::GET, "/api/x", None, "x")) + .expect_err("still refused") + .to_string(); + assert!(err.contains("(401)") && err.contains("login`"), "{err}"); + let m = mock.lock().expect("mock"); + assert_eq!((m.refreshes, m.api_calls), (1, 2)); + fs::remove_dir_all(&dir).expect("cleanup"); } }