agentmux_srv/muxbus/
discovery.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! AgentMux Cloud settings discovery
5//! (docs/specs/SPEC_CLOUD_SETTINGS_DISCOVERY_2026_09_27.md).
6//!
7//! The cloud publishes where its sign-in and relay live at
8//! `{relay REST base}/.well-known/agentmux-cloud.json`. srv reads it so the
9//! desktop follows the cloud instead of compiled-in values: a new Cognito
10//! user pool means a new client id, and installed builds would otherwise keep
11//! the old one. The relay REST base (`relay::rest_base_url`) stays the root of
12//! trust; everything else comes from the document it serves, with today's
13//! compiled values as the fallback.
14
15use std::sync::Mutex;
16use std::time::{Duration, Instant};
17
18use serde::Deserialize;
19
20/// Path of the discovery document under the relay REST base.
21pub const DISCOVERY_PATH: &str = "/.well-known/agentmux-cloud.json";
22
23/// How long a fetched document (or a failed fetch) is reused before asking again.
24const CACHE_TTL: Duration = Duration::from_secs(300);
25const FETCH_TIMEOUT: Duration = Duration::from_secs(5);
26
27/// What the cloud publishes. Only the fields srv uses; others are ignored.
28#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
29pub struct CloudSettings {
30    /// Relay REST base URL.
31    pub api: String,
32    /// Relay WebSocket URL.
33    pub ws: String,
34    /// Cloud Console URL.
35    #[serde(default)]
36    pub console: String,
37    pub cognito: CognitoSettings,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
41#[serde(rename_all = "camelCase")]
42pub struct CognitoSettings {
43    /// Hosted-UI base URL.
44    pub domain: String,
45    /// The desktop's public PKCE app client.
46    pub client_id: String,
47    #[serde(default)]
48    pub region: String,
49    #[serde(default)]
50    pub user_pool_id: String,
51}
52
53/// Parse and check a discovery document. `None` for anything srv shouldn't
54/// act on: another format version, a missing field, or a URL that isn't
55/// https/wss (plain http/ws only when the relay itself is plain http, i.e. a
56/// local dev relay).
57pub fn parse(body: &str, rest_base: &str) -> Option<CloudSettings> {
58    let value: serde_json::Value = serde_json::from_str(body).ok()?;
59    if value.get("version").and_then(|v| v.as_u64()) != Some(1) {
60        return None;
61    }
62    let settings: CloudSettings = serde_json::from_value(value).ok()?;
63    let local = rest_base.starts_with("http://");
64    let secure = |url: &str, scheme: &str, dev: &str| url.starts_with(scheme) || (local && url.starts_with(dev));
65    let ok = secure(&settings.api, "https://", "http://")
66        && secure(&settings.ws, "wss://", "ws://")
67        && secure(&settings.cognito.domain, "https://", "http://")
68        && !settings.cognito.client_id.is_empty();
69    ok.then_some(settings)
70}
71
72struct Cache {
73    fetched_at: Instant,
74    /// The document from the last successful fetch, kept across failed ones.
75    last_good: Option<CloudSettings>,
76}
77
78static CACHE: Mutex<Option<Cache>> = Mutex::new(None);
79
80/// The cloud's published settings, or `None` when it publishes none (an older
81/// relay, a network failure with no earlier copy). Cached for five minutes; a
82/// failed fetch keeps the last good copy and also waits five minutes before
83/// trying again, so an unreachable relay doesn't turn every caller into a
84/// request.
85pub async fn cloud_settings(http: &reqwest::Client) -> Option<CloudSettings> {
86    if let Some(cache) = CACHE.lock().unwrap_or_else(|e| e.into_inner()).as_ref() {
87        if cache.fetched_at.elapsed() < CACHE_TTL {
88            return cache.last_good.clone();
89        }
90    }
91    let base = super::relay::rest_base_url();
92    let fetched = fetch(http, &base).await;
93    let mut guard = CACHE.lock().unwrap_or_else(|e| e.into_inner());
94    let last_good = keep_last_good(guard.as_ref().and_then(|c| c.last_good.as_ref()), fetched);
95    *guard = Some(Cache { fetched_at: Instant::now(), last_good: last_good.clone() });
96    last_good
97}
98
99/// A fresh document replaces the cached one; a failed fetch keeps it.
100fn keep_last_good(previous: Option<&CloudSettings>, fetched: Option<CloudSettings>) -> Option<CloudSettings> {
101    fetched.or_else(|| previous.cloned())
102}
103
104async fn fetch(http: &reqwest::Client, base: &str) -> Option<CloudSettings> {
105    let url = format!("{}{}", base.trim_end_matches('/'), DISCOVERY_PATH);
106    let resp = match http.get(&url).timeout(FETCH_TIMEOUT).send().await {
107        Ok(r) => r,
108        Err(e) => {
109            tracing::debug!(url = %url, error = %e, "cloud discovery: fetch failed; using defaults");
110            return None;
111        }
112    };
113    if !resp.status().is_success() {
114        tracing::debug!(url = %url, status = %resp.status(), "cloud discovery: no document; using defaults");
115        return None;
116    }
117    let body = resp.text().await.ok()?;
118    let parsed = parse(&body, base);
119    if parsed.is_none() {
120        tracing::warn!(url = %url, "cloud discovery: document rejected (version, fields or URL schemes); using defaults");
121    }
122    parsed
123}
124
125/// Where the cloud subscriber connects: the published WebSocket URL, else the
126/// compiled one.
127pub async fn ws_url(http: &reqwest::Client) -> String {
128    // The same kind of override as `relay::rest_base_url`'s
129    // `AGENTMUX_MUXBUS_REST_URL`: for tests and a local muxbus server.
130    if let Some(url) = std::env::var("AGENTMUX_MUXBUS_WS_URL")
131        .ok()
132        .filter(|v| !v.is_empty())
133    {
134        return url;
135    }
136    cloud_settings(http)
137        .await
138        .map(|s| s.ws)
139        .unwrap_or_else(|| super::cloud_subscriber::MUXBUS_WS_URL.to_string())
140}
141
142/// The Cognito domain and client id to sign in with: the published ones when
143/// the cloud has a document, else what the caller supplied (the build's
144/// compiled values). Both of a pair come from the same source, never mixed.
145pub fn resolve_login(
146    discovered: Option<&CloudSettings>,
147    supplied_domain: &str,
148    supplied_client_id: &str,
149) -> Result<(String, String), String> {
150    if let Some(s) = discovered {
151        return Ok((s.cognito.domain.clone(), s.cognito.client_id.clone()));
152    }
153    if !supplied_domain.is_empty() && !supplied_client_id.is_empty() {
154        return Ok((supplied_domain.to_string(), supplied_client_id.to_string()));
155    }
156    Err("AgentMux Cloud sign-in isn't configured: the cloud publishes no settings and this build has none".to_string())
157}
158
159/// Was a stored sign-in issued to a Cognito client the cloud no longer
160/// publishes? Then it belongs to a user pool the cloud has left, and neither
161/// its tokens nor a refresh will work. `false` when that can't be told: no
162/// document, or a credential saved without its client id.
163pub fn client_id_superseded(discovered: Option<&CloudSettings>, stored_client_id: &str) -> bool {
164    match discovered {
165        Some(s) => !stored_client_id.is_empty() && s.cognito.client_id != stored_client_id,
166        None => false,
167    }
168}
169
170#[cfg(test)]
171mod tests {
172    use super::*;
173
174    const DOC: &str = r#"{
175        "version": 1,
176        "api": "https://muxbus.example.test",
177        "ws": "wss://muxbus-ws.example.test",
178        "console": "https://cloud.example.test",
179        "cognito": { "domain": "https://auth.example.test", "clientId": "abc123", "region": "us-east-1", "userPoolId": "us-east-1_X" },
180        "someFutureField": true
181    }"#;
182
183    #[test]
184    fn a_version_1_document_parses_and_ignores_unknown_fields() {
185        let s = parse(DOC, "https://muxbus.example.test").unwrap();
186        assert_eq!(s.ws, "wss://muxbus-ws.example.test");
187        assert_eq!(s.cognito.client_id, "abc123");
188        assert_eq!(s.cognito.user_pool_id, "us-east-1_X");
189    }
190
191    #[test]
192    fn other_versions_and_incomplete_documents_are_ignored() {
193        let base = "https://muxbus.example.test";
194        assert!(parse(&DOC.replace("\"version\": 1", "\"version\": 2"), base).is_none());
195        assert!(parse(&DOC.replace("\"clientId\": \"abc123\"", "\"clientId\": \"\""), base).is_none());
196        assert!(parse(r#"{"version":1,"api":"https://a","ws":"wss://b"}"#, base).is_none());
197        assert!(parse("not json", base).is_none());
198    }
199
200    #[test]
201    fn insecure_urls_are_refused_except_from_a_local_dev_relay() {
202        let plain = DOC.replace("wss://muxbus-ws", "ws://muxbus-ws");
203        assert!(parse(&plain, "https://muxbus.example.test").is_none());
204        assert!(parse(&plain, "http://localhost:3100").is_some());
205        let plain_auth = DOC.replace("https://auth", "http://auth");
206        assert!(parse(&plain_auth, "https://muxbus.example.test").is_none());
207    }
208
209    #[test]
210    fn login_uses_the_published_pair_else_the_builds_else_errors() {
211        let s = parse(DOC, "https://muxbus.example.test").unwrap();
212        assert_eq!(
213            resolve_login(Some(&s), "https://compiled", "compiled-id").unwrap(),
214            ("https://auth.example.test".to_string(), "abc123".to_string())
215        );
216        assert_eq!(
217            resolve_login(None, "https://compiled", "compiled-id").unwrap(),
218            ("https://compiled".to_string(), "compiled-id".to_string())
219        );
220        assert!(resolve_login(None, "https://compiled", "").is_err());
221        assert!(resolve_login(None, "", "").is_err());
222    }
223
224    #[test]
225    fn a_sign_in_from_another_client_is_superseded_only_when_the_cloud_says_so() {
226        let s = parse(DOC, "https://muxbus.example.test").unwrap();
227        assert!(client_id_superseded(Some(&s), "old-client"));
228        assert!(!client_id_superseded(Some(&s), "abc123"));
229        assert!(!client_id_superseded(None, "old-client"), "no document: can't tell");
230        assert!(!client_id_superseded(Some(&s), ""), "no recorded client id: can't tell");
231    }
232
233    /// A relay that answers every request with `status` and `body`.
234    async fn fake_relay(status: &'static str, body: String) -> String {
235        use tokio::io::{AsyncReadExt, AsyncWriteExt};
236        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
237        let addr = listener.local_addr().unwrap();
238        tokio::spawn(async move {
239            while let Ok((mut stream, _)) = listener.accept().await {
240                let body = body.clone();
241                tokio::spawn(async move {
242                    let mut buf = [0u8; 2048];
243                    let _ = stream.read(&mut buf).await;
244                    let resp = format!(
245                        "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
246                        body.len()
247                    );
248                    let _ = stream.write_all(resp.as_bytes()).await;
249                });
250            }
251        });
252        format!("http://{addr}")
253    }
254
255    #[tokio::test]
256    async fn a_relay_serving_the_document_is_discovered() {
257        let base = fake_relay("200 OK", DOC.to_string()).await;
258        let s = fetch(&reqwest::Client::new(), &base).await.expect("document served");
259        assert_eq!(s.cognito.client_id, "abc123");
260    }
261
262    #[tokio::test]
263    async fn an_unreachable_relay_or_one_without_the_document_is_not_discovered() {
264        let http = reqwest::Client::new();
265        // Nothing listens on port 1.
266        assert!(fetch(&http, "http://127.0.0.1:1").await.is_none());
267        let base = fake_relay("404 Not Found", "{}".to_string()).await;
268        assert!(fetch(&http, &base).await.is_none());
269        let base = fake_relay("200 OK", DOC.replace("\"version\": 1", "\"version\": 2")).await;
270        assert!(fetch(&http, &base).await.is_none());
271    }
272
273    #[test]
274    fn a_failed_fetch_keeps_the_last_good_document() {
275        let old = parse(DOC, "https://muxbus.example.test").unwrap();
276        let new = parse(&DOC.replace("abc123", "def456"), "https://muxbus.example.test").unwrap();
277        assert_eq!(keep_last_good(Some(&old), None), Some(old.clone()));
278        assert_eq!(keep_last_good(Some(&old), Some(new.clone())), Some(new));
279        assert_eq!(keep_last_good(None, None), None);
280    }
281}
agentmux_srv/muxbus/discovery.rs

agentmux_srv/muxbus/
discovery.rs
