248df20164
Swarm::http() built a reqwest client with no timeout. A token refresh runs under the exclusive flock on the login cache (kept: authelia rotates the refresh token, and a second refresher presenting the used one revokes the login), so an authelia that accepted the refresh and never answered held that lock forever and every other `swarmctl --swarm` invocation blocked behind it in flock. The client now has a 5 s connect timeout and a 30 s total timeout, the pair swarm-controller's own clients of swarm services use. No caller streams a response (every gateway call and every login POST reads one whole body), so a total bound fits all of them. The lock is unchanged; it is now held at most for one bounded request. Closes #5308
688 lines
26 KiB
Rust
688 lines
26 KiB
Rust
//! Reaching swarm-controller through the swarm domain's gateway instead of its
|
|
//! unix socket, as the operator's own SSO account.
|
|
//!
|
|
//! The gateway's `/api/` location runs authelia's `auth_request` and accepts an
|
|
//! OAuth 2.0 bearer token in place of a session cookie. `swarmctl login`
|
|
//! ([`crate::login`]) obtains one for the public `swarmctl` client and this
|
|
//! module caches it, refreshes it and presents it.
|
|
//!
|
|
//! ⚠️ authelia **rotates** the refresh token on every use, and presenting an
|
|
//! already-used one revokes the whole login. Every refresh therefore runs under
|
|
//! an exclusive lock on the cache, and rereads the cache once it holds it.
|
|
|
|
use std::fs::{self, DirBuilder, File, OpenOptions};
|
|
use std::io::Write as _;
|
|
use std::os::unix::fs::{DirBuilderExt as _, OpenOptionsExt as _};
|
|
use std::path::{Path, PathBuf};
|
|
use std::time::Duration;
|
|
|
|
use anyhow::{Context as _, Result, anyhow, bail};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
/// Registered under the same id in `swarm-authelia.nix`.
|
|
pub(crate) const CLIENT_ID: &str = "swarmctl";
|
|
|
|
/// Refresh this long before the access token's stated expiry, so a token
|
|
/// doesn't lapse between the check and the gateway's look at it.
|
|
const EXPIRY_MARGIN_SECS: i64 = 60;
|
|
|
|
/// Bound on reaching the gateway or authelia, as swarm-controller's clients of swarm services use.
|
|
const HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
|
|
/// Bound on one whole request, body included; a refresh holds the cache lock for that long at most.
|
|
const HTTP_TIMEOUT: Duration = Duration::from_secs(30);
|
|
|
|
/// Where a controller request goes.
|
|
pub(crate) enum Target {
|
|
/// swarm-controller's unix socket on this host.
|
|
Socket(PathBuf),
|
|
/// The swarm domain's gateway, with the cached login.
|
|
Swarm(Swarm),
|
|
}
|
|
|
|
/// A swarm reached by its domain.
|
|
pub(crate) struct Swarm {
|
|
/// `https://<swarm domain>`, with no path.
|
|
pub origin: url::Url,
|
|
/// A PEM CA certificate to trust beside the system roots.
|
|
pub ca_file: Option<PathBuf>,
|
|
/// Where the login cache lives: `$XDG_STATE_HOME/swarmctl` by default.
|
|
pub state_dir: PathBuf,
|
|
}
|
|
|
|
impl Swarm {
|
|
/// `url` with no scheme is taken as a bare domain, under `https://`.
|
|
pub(crate) fn parse(url: &str, ca_file: Option<PathBuf>, state_dir: PathBuf) -> Result<Self> {
|
|
let full = if url.contains("://") {
|
|
url.to_owned()
|
|
} else {
|
|
format!("https://{url}")
|
|
};
|
|
let parsed =
|
|
url::Url::parse(&full).with_context(|| format!("--swarm {url:?} is not a URL"))?;
|
|
if parsed.scheme() != "https" || parsed.host_str().is_none() {
|
|
bail!("--swarm {url:?} must be an https:// URL naming the swarm's domain");
|
|
}
|
|
let origin = url::Url::parse(&parsed.origin().ascii_serialization())
|
|
.context("re-parsing the swarm's origin")?;
|
|
Ok(Self {
|
|
origin,
|
|
ca_file,
|
|
state_dir,
|
|
})
|
|
}
|
|
|
|
/// What the token must name for the gateway to admit it at `/api/`.
|
|
/// authelia matches it as a URL prefix.
|
|
pub(crate) fn audience(&self) -> String {
|
|
format!("{}api/", self.origin)
|
|
}
|
|
|
|
/// An HTTP client that never follows a redirect: from the gateway one
|
|
/// means "log in", and following it would land on the login page.
|
|
pub(crate) fn http(&self) -> Result<reqwest::Client> {
|
|
let mut builder = bounded(
|
|
reqwest::Client::builder(),
|
|
HTTP_CONNECT_TIMEOUT,
|
|
HTTP_TIMEOUT,
|
|
);
|
|
if let Some(path) = &self.ca_file {
|
|
let pem = fs::read(path)
|
|
.with_context(|| format!("reading the CA certificate {}", path.display()))?;
|
|
let cert = reqwest::Certificate::from_pem(&pem)
|
|
.with_context(|| format!("parsing {} as a PEM certificate", path.display()))?;
|
|
builder = builder.add_root_certificate(cert);
|
|
}
|
|
builder.build().context("building the HTTPS client")
|
|
}
|
|
|
|
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"),
|
|
};
|
|
self.state_dir.join(name)
|
|
}
|
|
|
|
fn not_logged_in(&self) -> String {
|
|
format!(
|
|
"not logged in to {} — run `swarmctl --swarm {} login`",
|
|
self.origin, self.origin
|
|
)
|
|
}
|
|
|
|
/// 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,
|
|
uri: &str,
|
|
body: Option<Vec<u8>>,
|
|
what: &str,
|
|
) -> Result<(hyper::StatusCode, bytes::Bytes)> {
|
|
self.request_with(&self.http()?, method, uri, body, what)
|
|
.await
|
|
}
|
|
|
|
async fn request_with(
|
|
&self,
|
|
http: &reqwest::Client,
|
|
method: hyper::Method,
|
|
uri: &str,
|
|
body: Option<Vec<u8>>,
|
|
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!(
|
|
"{} refused the cached login (401). {}",
|
|
self.origin,
|
|
self.not_logged_in()
|
|
);
|
|
}
|
|
if status.is_redirection() {
|
|
bail!(
|
|
"{} answered {status} instead of the controller's response; its `/api/` \
|
|
did not accept a bearer token",
|
|
self.origin
|
|
);
|
|
}
|
|
let body = resp.bytes().await.context("reading the response")?;
|
|
Ok((status, body))
|
|
}
|
|
|
|
/// 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<String> {
|
|
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 !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 !due(&cached) {
|
|
return Ok(cached.access_token);
|
|
}
|
|
let token = token_request(
|
|
http,
|
|
&cached.auth,
|
|
&[
|
|
("grant_type", "refresh_token"),
|
|
("client_id", CLIENT_ID),
|
|
("refresh_token", &cached.refresh_token),
|
|
],
|
|
)
|
|
.await
|
|
.with_context(|| format!("refreshing the login. {}", self.not_logged_in()))?;
|
|
let fresh = refreshed(cached, token, now);
|
|
write_cache(&path, &fresh)?;
|
|
Ok(fresh.access_token)
|
|
}
|
|
}
|
|
|
|
/// `builder` following no redirect, with every request bounded by `connect`
|
|
/// and `total`. No caller streams a response, so a total bound fits them all.
|
|
fn bounded(
|
|
builder: reqwest::ClientBuilder,
|
|
connect: Duration,
|
|
total: Duration,
|
|
) -> reqwest::ClientBuilder {
|
|
builder
|
|
.redirect(reqwest::redirect::Policy::none())
|
|
.connect_timeout(connect)
|
|
.timeout(total)
|
|
}
|
|
|
|
/// `$XDG_STATE_HOME/swarmctl`, else `~/.local/state/swarmctl`.
|
|
pub(crate) fn state_dir() -> Result<PathBuf> {
|
|
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 {
|
|
/// authelia's origin, where the refresh goes.
|
|
pub auth: String,
|
|
pub access_token: String,
|
|
pub refresh_token: String,
|
|
/// Unix seconds.
|
|
pub expires_at: i64,
|
|
}
|
|
|
|
/// authelia's token endpoint answer.
|
|
#[derive(Deserialize)]
|
|
pub(crate) struct TokenResponse {
|
|
pub access_token: String,
|
|
pub expires_in: i64,
|
|
pub refresh_token: Option<String>,
|
|
}
|
|
|
|
pub(crate) fn needs_refresh(cached: &Cached, now: i64) -> bool {
|
|
cached.expires_at - EXPIRY_MARGIN_SECS <= now
|
|
}
|
|
|
|
/// The cache after a refresh. The new refresh token replaces the old one,
|
|
/// which is dead from the moment it was used.
|
|
pub(crate) fn refreshed(cached: Cached, token: TokenResponse, now: i64) -> Cached {
|
|
Cached {
|
|
refresh_token: token.refresh_token.unwrap_or(cached.refresh_token),
|
|
access_token: token.access_token,
|
|
expires_at: now + token.expires_in,
|
|
auth: cached.auth,
|
|
}
|
|
}
|
|
|
|
/// POST a form to authelia's token endpoint.
|
|
pub(crate) async fn token_request(
|
|
http: &reqwest::Client,
|
|
auth: &str,
|
|
form: &[(&str, &str)],
|
|
) -> Result<TokenResponse> {
|
|
let resp = http
|
|
.post(format!("{auth}/api/oidc/token"))
|
|
.form(form)
|
|
.send()
|
|
.await
|
|
.with_context(|| format!("calling {auth}/api/oidc/token"))?;
|
|
oauth_json(resp).await
|
|
}
|
|
|
|
/// Decode an OAuth 2.0 JSON answer, or turn its `error` into the message.
|
|
pub(crate) async fn oauth_json<T: serde::de::DeserializeOwned>(
|
|
resp: reqwest::Response,
|
|
) -> Result<T> {
|
|
let status = resp.status();
|
|
let body = resp.bytes().await.context("reading authelia's response")?;
|
|
if !status.is_success() {
|
|
let v: serde_json::Value = serde_json::from_slice(&body).unwrap_or_default();
|
|
let field = |k| v.get(k).and_then(serde_json::Value::as_str).unwrap_or("");
|
|
return Err(anyhow!(
|
|
"authelia answered {status}: {} {}",
|
|
field("error"),
|
|
field("error_description")
|
|
));
|
|
}
|
|
serde_json::from_slice(&body).context("decoding authelia's response")
|
|
}
|
|
|
|
pub(crate) fn read_cache(path: &Path) -> Result<Option<Cached>> {
|
|
match fs::read(path) {
|
|
Ok(bytes) => serde_json::from_slice(&bytes)
|
|
.map(Some)
|
|
.with_context(|| format!("decoding the login cache {}", path.display())),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
|
|
Err(e) => Err(e).with_context(|| format!("reading the login cache {}", path.display())),
|
|
}
|
|
}
|
|
|
|
/// Write through a 0600 temp file and rename, so the tokens are never
|
|
/// readable by anyone else, not even for the moment between create and chmod.
|
|
pub(crate) fn write_cache(path: &Path, cached: &Cached) -> Result<()> {
|
|
let dir = path
|
|
.parent()
|
|
.context("the login cache path has no directory")?;
|
|
DirBuilder::new()
|
|
.recursive(true)
|
|
.mode(0o700)
|
|
.create(dir)
|
|
.with_context(|| format!("creating {}", dir.display()))?;
|
|
let tmp = path.with_extension("json.tmp");
|
|
let _ = fs::remove_file(&tmp);
|
|
let mut file = OpenOptions::new()
|
|
.write(true)
|
|
.create_new(true)
|
|
.mode(0o600)
|
|
.open(&tmp)
|
|
.with_context(|| format!("creating {}", tmp.display()))?;
|
|
file.write_all(&serde_json::to_vec(cached)?)
|
|
.and_then(|()| file.sync_all())
|
|
.with_context(|| format!("writing {}", tmp.display()))?;
|
|
fs::rename(&tmp, path).with_context(|| format!("replacing {}", path.display()))
|
|
}
|
|
|
|
/// Hold an exclusive lock beside the cache until the returned file drops.
|
|
pub(crate) fn lock(path: &Path) -> Result<File> {
|
|
if let Some(dir) = path.parent() {
|
|
DirBuilder::new().recursive(true).mode(0o700).create(dir)?;
|
|
}
|
|
let lock_path = path.with_extension("lock");
|
|
let file = OpenOptions::new()
|
|
.create(true)
|
|
.truncate(false)
|
|
.write(true)
|
|
.mode(0o600)
|
|
.open(&lock_path)
|
|
.with_context(|| format!("opening {}", lock_path.display()))?;
|
|
rustix::fs::flock(&file, rustix::fs::FlockOperation::LockExclusive)
|
|
.with_context(|| format!("locking {}", lock_path.display()))?;
|
|
Ok(file)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use super::*;
|
|
|
|
fn cached(expires_at: i64) -> Cached {
|
|
Cached {
|
|
auth: "https://auth.example.test".into(),
|
|
access_token: "at-old".into(),
|
|
refresh_token: "rt-old".into(),
|
|
expires_at,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_token_is_refreshed_only_inside_the_margin() {
|
|
assert!(!needs_refresh(
|
|
&cached(1_000),
|
|
1_000 - EXPIRY_MARGIN_SECS - 1
|
|
));
|
|
assert!(needs_refresh(&cached(1_000), 1_000 - EXPIRY_MARGIN_SECS));
|
|
assert!(needs_refresh(&cached(1_000), 2_000));
|
|
}
|
|
|
|
/// authelia rotates the refresh token; keeping the used one would make
|
|
/// the next refresh revoke the whole login.
|
|
#[test]
|
|
fn a_refresh_replaces_both_tokens_and_the_expiry() {
|
|
let token = TokenResponse {
|
|
access_token: "at-new".into(),
|
|
expires_in: 3599,
|
|
refresh_token: Some("rt-new".into()),
|
|
};
|
|
let fresh = refreshed(cached(0), token, 5_000);
|
|
assert_eq!(fresh.access_token, "at-new");
|
|
assert_eq!(fresh.refresh_token, "rt-new");
|
|
assert_eq!(fresh.expires_at, 8_599);
|
|
assert_eq!(fresh.auth, "https://auth.example.test");
|
|
}
|
|
|
|
#[test]
|
|
fn the_cache_is_written_owner_only_and_reads_back() {
|
|
let dir = std::env::temp_dir().join(format!("swarmctl-cache-{}", std::process::id()));
|
|
let path = dir.join("swarmctl").join("swarm.example.test.json");
|
|
let _ = fs::remove_dir_all(&dir);
|
|
|
|
assert_eq!(read_cache(&path).expect("missing is not an error"), None);
|
|
write_cache(&path, &cached(42)).expect("writes");
|
|
write_cache(&path, &cached(43)).expect("overwrites");
|
|
|
|
let mode = |p: &Path| fs::metadata(p).expect("stat").permissions().mode() & 0o777;
|
|
assert_eq!(mode(&path), 0o600);
|
|
assert_eq!(mode(path.parent().expect("dir")), 0o700);
|
|
assert_eq!(read_cache(&path).expect("reads"), Some(cached(43)));
|
|
fs::remove_dir_all(&dir).expect("cleanup");
|
|
}
|
|
|
|
#[test]
|
|
fn the_audience_is_the_api_prefix_of_the_origin() {
|
|
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, "/s".into()).is_err());
|
|
}
|
|
|
|
/// A port is no scheme: `host:8443` would otherwise parse as a URL whose
|
|
/// scheme is `host`.
|
|
#[test]
|
|
fn a_bare_domain_is_taken_as_https() {
|
|
let origin = |url| Swarm::parse(url, None, "/s".into()).expect("parses").origin;
|
|
assert_eq!(
|
|
origin("swarm.example.test").as_str(),
|
|
"https://swarm.example.test/"
|
|
);
|
|
assert_eq!(
|
|
origin("swarm.example.test:8443").as_str(),
|
|
"https://swarm.example.test:8443/"
|
|
);
|
|
}
|
|
|
|
/// 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<String>,
|
|
access: Option<String>,
|
|
refreshes: u32,
|
|
api_calls: u32,
|
|
api_refuses_all: bool,
|
|
}
|
|
|
|
fn mock() -> Arc<Mutex<Mock>> {
|
|
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<Mutex<Mock>>) -> 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<Mock>,
|
|
req: hyper::Request<hyper::body::Incoming>,
|
|
) -> hyper::Response<http_body_util::Full<bytes::Bytes>> {
|
|
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<T>(fut: impl std::future::Future<Output = T>) -> 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<String> = 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");
|
|
}
|
|
|
|
/// authelia takes the refresh and never answers: the refresh fails at the
|
|
/// client's bound instead of holding the cache lock forever.
|
|
#[test]
|
|
fn a_stalled_refresh_fails_instead_of_hanging() {
|
|
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
|
|
let origin = url::Url::parse(&format!("http://{}/", listener.local_addr().expect("addr")))
|
|
.expect("url");
|
|
// Accepted and kept open, so the client sees neither an answer nor a close.
|
|
std::thread::spawn(move || {
|
|
let mut held = Vec::new();
|
|
while let Ok((stream, _)) = listener.accept() {
|
|
held.push(stream);
|
|
}
|
|
});
|
|
let dir = scratch("stalled");
|
|
let swarm = logged_in(&origin, &dir, "at-0", 0);
|
|
let http = bounded(
|
|
reqwest::Client::builder().tls_certs_only([]),
|
|
Duration::from_secs(1),
|
|
Duration::from_millis(200),
|
|
)
|
|
.build()
|
|
.expect("client");
|
|
|
|
let refresh = block_on(async {
|
|
tokio::time::timeout(Duration::from_secs(10), swarm.access_token(&http, None)).await
|
|
});
|
|
let err = refresh
|
|
.expect("the client's bound fires, not this one")
|
|
.expect_err("a silent authelia issues no token");
|
|
assert!(
|
|
format!("{err:#}").contains("refreshing the login"),
|
|
"{err:#}"
|
|
);
|
|
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");
|
|
}
|
|
}
|