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

@@ -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

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"

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", &"<redacted>")
+            .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.

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<String> {
+    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<String> {
+    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::<Exchanged>(&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());
+    }
+}

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<WsStream> {
+    // 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<WsStream> {
     // 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.

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::<usize>().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");