From b4948076a0fa1621bf2707dfcfbc2abfdc5b313b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Thu, 29 Oct 2026 15:10:27 +0000 Subject: [PATCH] Trade the gateway token for an edge JWT on every connection The proxy put its configured gateway_token straight into the upgrade's Authorization header, but the backend only accepts a one-hour edge JWT (scope "edge") there, minted by POST /auth/edge-token from the farm's edge proxy token. Nothing in the proxy ever made that call, so a gateway configured with the token the console hands out could not connect at all, and one given a hand-minted JWT (which is what every test did) would have been locked out an hour later on its first reconnect. Each connection attempt now starts by trading gateway_token for a fresh JWT, at an address derived from the socket URL so there is one to configure. A 401 is reported as what it means (token rotated or wrong), since retrying will not fix it. Config's Debug no longer prints the token. tests/uplink_late_result.rs now plays the cloud's exchange endpoint and asserts the socket opens with the minted JWT, not the configured token. .env.example and continuum.example.toml say what the token is. --- .env.example | 10 ++-- continuum.example.toml | 2 + src/config.rs | 18 +++++- src/uplink/auth.rs | 107 ++++++++++++++++++++++++++++++++++++ src/uplink/mod.rs | 7 ++- tests/uplink_late_result.rs | 47 +++++++++++++++- 6 files changed, 184 insertions(+), 7 deletions(-) create mode 100644 src/uplink/auth.rs diff --git a/.env.example b/.env.example index 11a55f4..fbc8550 100644 --- a/.env.example +++ b/.env.example @@ -1,10 +1,12 @@ # Identity CONTINUUM_GATEWAY_ID=gw-farm01-rpi01 CONTINUUM_FARM_ID=replace-with-farm-uuid -# The edge JWT from POST /auth/edge-token. Sent as an Authorization: Bearer -# header on the uplink's WebSocket upgrade - not in a message and not in the -# URL - so the backend verifies it before any socket exists. -CONTINUUM_GATEWAY_TOKEN=replace-with-provisioning-token +# The farm's edge proxy token, shown once by the console's Setup page (or +# POST /farms/:id/edge-proxy-token). The gateway trades it for a one-hour edge +# JWT (POST /auth/edge-token) on every connection, and the JWT - sent as an +# Authorization: Bearer header on the WebSocket upgrade, never in a message +# or the URL - is what the backend verifies before any socket exists. +CONTINUUM_GATEWAY_TOKEN=replace-with-the-farms-edge-proxy-token # Cloud uplink CONTINUUM_UPLINK_URL=wss://api.continuum.local/ws/edge/v1 diff --git a/continuum.example.toml b/continuum.example.toml index 4a02873..da9d44d 100644 --- a/continuum.example.toml +++ b/continuum.example.toml @@ -19,6 +19,8 @@ gateway_id = "REPLACE_ME" farm_id = "REPLACE_ME" +# The farm's edge proxy token from the console (Setup page). Traded for a +# short-lived edge JWT on every connection - see src/uplink/auth.rs. gateway_token = "REPLACE_ME" # uplink_url = "wss://api.continuum.local/ws/edge/v1" diff --git a/src/config.rs b/src/config.rs index ba51aee..1153793 100644 --- a/src/config.rs +++ b/src/config.rs @@ -45,7 +45,7 @@ use std::time::Duration; use serde::Deserialize; -#[derive(Debug, Clone)] +#[derive(Clone)] pub struct Config { pub gateway_id: String, pub farm_id: String, @@ -287,6 +287,22 @@ impl fmt::Debug for WifiSettings { } } +/// By hand: the derived `Debug` would print `gateway_token`, the credential +/// that opens this farm's gateway socket, into any log line that formats the +/// config. +impl fmt::Debug for Config { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("Config") + .field("gateway_id", &self.gateway_id) + .field("farm_id", &self.farm_id) + .field("gateway_token", &"") + .field("uplink_url", &self.uplink_url) + .field("printers_path", &self.printers_path) + .field("store_path", &self.store_path) + .finish_non_exhaustive() + } +} + impl Config { /// Environment only - the behaviour from before there was a file, kept /// so existing callers and tests do not change. diff --git a/src/uplink/auth.rs b/src/uplink/auth.rs new file mode 100644 index 0000000..9e32a21 --- /dev/null +++ b/src/uplink/auth.rs @@ -0,0 +1,107 @@ +//! Getting from the gateway's long-lived credential to the short-lived one +//! the socket wants. +//! +//! The cloud hands a farm's owner an *edge proxy token*, which is what goes in +//! this gateway's settings as `gateway_token`. It does not open the socket +//! directly: the upgrade wants an *edge JWT*, which the cloud signs for an +//! hour in exchange for the proxy token (`POST /auth/edge-token`). So every +//! connection attempt starts by trading, which is also why a gateway that +//! has been up for days is not locked out when the first JWT expires - each +//! reconnect gets a fresh one. + +use std::time::Duration; + +use anyhow::Context; +use serde::Deserialize; + +use crate::config::Config; + +const EXCHANGE_TIMEOUT: Duration = Duration::from_secs(15); +const SOCKET_PATH: &str = "/ws/edge/v1"; + +/// Where to trade the token, derived from the socket URL so there is one +/// address to configure, not two that can disagree. +pub fn exchange_url(uplink_url: &str) -> anyhow::Result { + let (scheme, rest) = if let Some(rest) = uplink_url.strip_prefix("wss://") { + ("https://", rest) + } else if let Some(rest) = uplink_url.strip_prefix("ws://") { + ("http://", rest) + } else { + anyhow::bail!("the uplink URL must start with ws:// or wss://: {uplink_url}"); + }; + let base = rest + .strip_suffix(SOCKET_PATH) + .with_context(|| format!("the uplink URL must end with {SOCKET_PATH}: {uplink_url}"))?; + Ok(format!("{scheme}{base}/auth/edge-token")) +} + +#[derive(Deserialize)] +struct Exchanged { + token: String, +} + +/// Trades `gateway_token` for an edge JWT. +/// +/// A 401 is reported as what it means - the cloud does not accept this +/// farm/token pair - because retrying will not fix it, and the usual cause is +/// a token that was rotated in the console while this gateway kept the old +/// one. +pub async fn fetch_edge_token(config: &Config) -> anyhow::Result { + let url = exchange_url(&config.uplink_url)?; + let response = reqwest::Client::builder() + .timeout(EXCHANGE_TIMEOUT) + .build()? + .post(&url) + // Built by hand: the `json` feature of reqwest is off on purpose (see + // Cargo.toml), and this is one small body. + .header("Content-Type", "application/json") + .body( + serde_json::json!({ + "farmId": config.farm_id, + "edgeProxyToken": config.gateway_token, + }) + .to_string(), + ) + .send() + .await + .with_context(|| format!("could not reach {url} to get an edge token"))?; + + match response.status() { + status if status.is_success() => { + let body = response.bytes().await?; + Ok(serde_json::from_slice::(&body).context("the cloud's edge token reply was not understood")?.token) + } + reqwest::StatusCode::UNAUTHORIZED => { + anyhow::bail!("the cloud does not accept this farm_id / gateway_token - was the token rotated in the console?") + } + status => anyhow::bail!("the cloud answered {status} when asked for an edge token"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_exchange_address_follows_the_socket_address() { + assert_eq!( + exchange_url("wss://api.continuum.local/ws/edge/v1").unwrap(), + "https://api.continuum.local/auth/edge-token" + ); + assert_eq!( + exchange_url("ws://localhost:3040/ws/edge/v1").unwrap(), + "http://localhost:3040/auth/edge-token" + ); + // A prefix in front of the API survives. + assert_eq!( + exchange_url("wss://example.com/api/ws/edge/v1").unwrap(), + "https://example.com/api/auth/edge-token" + ); + } + + #[test] + fn an_address_it_cannot_follow_is_an_error_not_a_guess() { + assert!(exchange_url("https://api/ws/edge/v1").is_err()); + assert!(exchange_url("wss://api/somewhere/else").is_err()); + } +} diff --git a/src/uplink/mod.rs b/src/uplink/mod.rs index 30165f8..37ed3f0 100644 --- a/src/uplink/mod.rs +++ b/src/uplink/mod.rs @@ -26,6 +26,7 @@ //! readings for no consumer. See PLAN.md for what would actually be worth //! buffering. +mod auth; mod download; mod webrtc; pub mod protocol; @@ -176,6 +177,10 @@ pub async fn run( /// handshake from a connection that opened and later dropped; they want /// different retry behaviour. async fn connect(config: &Config) -> anyhow::Result { + // A fresh edge JWT per attempt: it lives an hour, so one fetched at + // startup would lock a long-running gateway out on its first reconnect. + let edge_token = auth::fetch_edge_token(config).await?; + // The edge token goes on the upgrade request, not in the first message. // The backend verifies it before the socket exists, so a bad token is an // HTTP 401 here rather than a connection that opens and is then closed - @@ -184,7 +189,7 @@ async fn connect(config: &Config) -> anyhow::Result { // those tools already know to redact. let request = Request::builder() .uri(&config.uplink_url) - .header("Authorization", format!("Bearer {}", config.gateway_token)) + .header("Authorization", format!("Bearer {edge_token}")) // tokio-tungstenite fills these in for a plain URL, but not once we // hand it a request we built ourselves - and a handshake missing any // of them is rejected by the server rather than ignored. diff --git a/tests/uplink_late_result.rs b/tests/uplink_late_result.rs index c2d5206..c250891 100644 --- a/tests/uplink_late_result.rs +++ b/tests/uplink_late_result.rs @@ -18,10 +18,12 @@ use std::time::Duration; use futures_util::{SinkExt, StreamExt}; use serde_json::{json, Value}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::sync::Mutex; use tokio_tungstenite::WebSocketStream; use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::handshake::server::{Request, Response}; use continuum_proxy::config::Config; use continuum_proxy::uplink; @@ -70,6 +72,36 @@ where } } +/// Plays the cloud's `/auth/edge-token`: reads one HTTP request, checks that +/// the gateway sent its farm and long-lived token, and answers with a JWT. +/// Returns the request text so the caller can assert on it. +async fn answer_token_exchange(listener: &TcpListener) -> String { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + let mut buf = [0u8; 4096]; + loop { + let n = tokio::time::timeout(WAIT, stream.read(&mut buf)).await.unwrap().unwrap(); + assert!(n > 0, "the gateway hung up before finishing its token request"); + request.extend_from_slice(&buf[..n]); + let text = String::from_utf8_lossy(&request).into_owned(); + if let Some((head, body)) = text.split_once("\r\n\r\n") { + let length = head + .lines() + .find_map(|l| l.to_ascii_lowercase().strip_prefix("content-length:").map(|v| v.trim().parse::().unwrap())) + .unwrap_or(0); + if body.len() >= length { + break; + } + } + } + let body = r#"{"token":"minted-edge-jwt","expiresInSeconds":3600}"#; + stream + .write_all(format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).as_bytes()) + .await + .unwrap(); + String::from_utf8_lossy(&request).into_owned() +} + #[tokio::test] async fn a_long_command_that_fails_reports_its_reason_after_the_fact() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -79,8 +111,21 @@ async fn a_long_command_that_fails_reports_its_reason_after_the_fact() { let (outbound, inbox) = uplink::channel(); tokio::spawn(uplink::run(config(port), fleet, None, outbound, inbox)); + // The gateway trades its long-lived token for an edge JWT first, and the + // JWT - not the token it was configured with - is what opens the socket. + let exchange = answer_token_exchange(&listener).await; + assert!(exchange.starts_with("POST /auth/edge-token "), "{exchange}"); + assert!(exchange.contains(r#""farmId":"test-farm""#) && exchange.contains(r#""edgeProxyToken":"test-token""#), "{exchange}"); + let (stream, _) = listener.accept().await.unwrap(); - let mut ws = tokio_tungstenite::accept_async(stream).await.unwrap(); + let mut upgrade_auth = String::new(); + let mut ws = tokio_tungstenite::accept_hdr_async(stream, |request: &Request, response: Response| { + upgrade_auth = request.headers().get("authorization").unwrap().to_str().unwrap().to_string(); + Ok(response) + }) + .await + .unwrap(); + assert_eq!(upgrade_auth, "Bearer minted-edge-jwt", "the socket must be opened with the minted JWT"); let hello = next_json(&mut ws).await; assert_eq!(hello["type"], "hello", "a gateway introduces itself first");