diff --git a/Cargo.lock b/Cargo.lock index 8d6d526..d8f6bce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -162,6 +162,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "chrono" version = "0.4.44" @@ -174,6 +180,35 @@ dependencies = [ "windows-link", ] +[[package]] +name = "cookie" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ddef33a339a91ea89fb53151bd0a4689cfce27055c291dfa69945475d22c747" +dependencies = [ + "percent-encoding", + "time", + "version_check", +] + +[[package]] +name = "cookie_store" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15b2c103cf610ec6cae3da84a766285b42fd16aad564758459e6ecf128c75206" +dependencies = [ + "cookie", + "document-features", + "idna", + "log", + "publicsuffix", + "serde", + "serde_derive", + "serde_json", + "time", + "url", +] + [[package]] name = "core-foundation-sys" version = "0.8.7" @@ -205,6 +240,15 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" +[[package]] +name = "deranged" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" +dependencies = [ + "powerfmt", +] + [[package]] name = "digest" version = "0.10.7" @@ -247,6 +291,15 @@ dependencies = [ "syn", ] +[[package]] +name = "document-features" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4b8a88685455ed29a21542a33abd9cb6510b6b129abadabdcef0f4c55bc8f61" +dependencies = [ + "litrs", +] + [[package]] name = "equivalent" version = "1.0.2" @@ -353,6 +406,20 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "r-efi 5.3.0", + "wasip2", + "wasm-bindgen", +] + [[package]] name = "getrandom" version = "0.4.2" @@ -361,7 +428,7 @@ checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 6.0.0", "wasip2", "wasip3", ] @@ -450,6 +517,23 @@ dependencies = [ "pin-project-lite", "smallvec", "tokio", + "want", +] + +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots", ] [[package]] @@ -458,13 +542,21 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64", "bytes", + "futures-channel", + "futures-util", "http", "http-body", "hyper", + "ipnet", + "libc", + "percent-encoding", "pin-project-lite", + "socket2", "tokio", "tower-service", + "tracing", ] [[package]] @@ -612,6 +704,12 @@ dependencies = [ "serde_core", ] +[[package]] +name = "ipnet" +version = "2.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" + [[package]] name = "itoa" version = "1.0.18" @@ -682,12 +780,24 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" +[[package]] +name = "litrs" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11d3d7f243d5c5a8b9bb5d6dd2b1602c0cb0b9db1621bafc7ed66e35ff9fe092" + [[package]] name = "log" version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "maestro-daemon" version = "0.1.0" @@ -703,7 +813,7 @@ dependencies = [ "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 1.0.69", "tokio", "tokio-tungstenite 0.21.0", "toml", @@ -730,12 +840,14 @@ dependencies = [ "chrono", "dirs", "futures-util", + "jsonwebtoken", "maestro-protocol", - "rand", + "rand 0.8.6", + "reqwest", "serde", "serde_json", "tempfile", - "thiserror", + "thiserror 1.0.69", "tokio", "tokio-tungstenite 0.21.0", "toml", @@ -790,6 +902,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "num-conv" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521739c6d2bac4aa25192232afe6841231376b2b26d4d9fae5ecf8ca5772e441" + [[package]] name = "num-traits" version = "0.2.19" @@ -832,6 +950,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "powerfmt" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -860,6 +984,77 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "psl-types" +version = "2.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33cb294fe86a74cbcf50d4445b37da762029549ebeea341421c7c70370f86cac" + +[[package]] +name = "publicsuffix" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f42ea446cab60335f76979ec15e12619a2165b5ae2c12166bef27d283a9fadf" +dependencies = [ + "idna", + "psl-types", +] + +[[package]] +name = "quinn" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20" +dependencies = [ + "bytes", + "cfg_aliases", + "pin-project-lite", + "quinn-proto", + "quinn-udp", + "rustc-hash", + "rustls", + "socket2", + "thiserror 2.0.18", + "tokio", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-proto" +version = "0.11.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" +dependencies = [ + "bytes", + "getrandom 0.3.4", + "lru-slab", + "rand 0.9.4", + "ring", + "rustc-hash", + "rustls", + "rustls-pki-types", + "slab", + "thiserror 2.0.18", + "tinyvec", + "tracing", + "web-time", +] + +[[package]] +name = "quinn-udp" +version = "0.5.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +dependencies = [ + "cfg_aliases", + "libc", + "once_cell", + "socket2", + "tracing", + "windows-sys 0.52.0", +] + [[package]] name = "quote" version = "1.0.45" @@ -869,6 +1064,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" @@ -882,8 +1083,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" dependencies = [ "libc", - "rand_chacha", - "rand_core", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", ] [[package]] @@ -893,7 +1104,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", ] [[package]] @@ -905,6 +1126,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "redox_users" version = "0.4.6" @@ -913,7 +1143,7 @@ checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43" dependencies = [ "getrandom 0.2.17", "libredox", - "thiserror", + "thiserror 1.0.69", ] [[package]] @@ -933,6 +1163,46 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "reqwest" +version = "0.12.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" +dependencies = [ + "base64", + "bytes", + "cookie", + "cookie_store", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", +] + [[package]] name = "ring" version = "0.17.14" @@ -947,6 +1217,12 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "rustc-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" + [[package]] name = "rustix" version = "1.1.4" @@ -960,6 +1236,41 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.23.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -1117,6 +1428,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.117" @@ -1133,6 +1450,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -1164,7 +1484,16 @@ version = "1.0.69" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" dependencies = [ - "thiserror-impl", + "thiserror-impl 1.0.69", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl 2.0.18", ] [[package]] @@ -1178,6 +1507,17 @@ dependencies = [ "syn", ] +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "thread_local" version = "1.1.9" @@ -1187,6 +1527,37 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "time" +version = "0.3.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" +dependencies = [ + "deranged", + "itoa", + "num-conv", + "powerfmt", + "serde_core", + "time-core", + "time-macros", +] + +[[package]] +name = "time-core" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" + +[[package]] +name = "time-macros" +version = "0.2.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tinystr" version = "0.8.3" @@ -1197,6 +1568,21 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinyvec" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e61e67053d25a4e82c844e8424039d9745781b3fc4f32b8d55ed50f5f667ef3" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.52.3" @@ -1224,6 +1610,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.21.0" @@ -1305,6 +1701,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -1379,6 +1793,12 @@ dependencies = [ "tracing-log", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "tungstenite" version = "0.21.0" @@ -1391,9 +1811,9 @@ dependencies = [ "http", "httparse", "log", - "rand", + "rand 0.8.6", "sha1", - "thiserror", + "thiserror 1.0.69", "url", "utf-8", ] @@ -1410,9 +1830,9 @@ dependencies = [ "http", "httparse", "log", - "rand", + "rand 0.8.6", "sha1", - "thiserror", + "thiserror 1.0.69", "utf-8", ] @@ -1488,6 +1908,15 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -1525,6 +1954,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.72" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9473dbd2991ae90b6291c3c32c30c6187ac49aa32f9905d1cce280ec1e110b0f" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.122" @@ -1591,6 +2030,35 @@ dependencies = [ "semver", ] +[[package]] +name = "web-sys" +version = "0.3.99" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d621441cfc37b84979402712047321980c178f299193a3589d05b99e8763436" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "webpki-roots" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52f5ee44c96cf55f1b349600768e3ece3a8f26010c05265ab73f945bb1a2eb9d" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -1971,6 +2439,12 @@ dependencies = [ "synstructure", ] +[[package]] +name = "zeroize" +version = "1.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" + [[package]] name = "zerotrie" version = "0.2.4" diff --git a/crates/maestro-daemon/src/config.rs b/crates/maestro-daemon/src/config.rs index dd93cf4..a330aec 100644 --- a/crates/maestro-daemon/src/config.rs +++ b/crates/maestro-daemon/src/config.rs @@ -23,6 +23,11 @@ pub struct DaemonConfig { pub hostname: String, pub registry_url: String, pub bind_addr: SocketAddr, + /// Reachable URL for this daemon's per-session WS endpoint, advertised in + /// `Hello` so the registry can hand it to clients via `connect-info`. + /// Distinct from `bind_addr` because the bind may be `0.0.0.0` while the + /// publicly reachable address is a Wireguard / VPN interface. + pub daemon_url: String, pub token: String, /// Root for everything daemon-local: sidecars, FIFOs, etc. /// Default: `~/.local/state/maestrod`. diff --git a/crates/maestro-daemon/src/main.rs b/crates/maestro-daemon/src/main.rs index 84179ff..2752549 100644 --- a/crates/maestro-daemon/src/main.rs +++ b/crates/maestro-daemon/src/main.rs @@ -71,6 +71,7 @@ async fn run() -> Result<(), Box> { config.registry_url.clone(), config.daemon_id.clone(), config.hostname.clone(), + config.daemon_url.clone(), config.token.clone(), manager.clone(), ); diff --git a/crates/maestro-daemon/src/registry_client.rs b/crates/maestro-daemon/src/registry_client.rs index 496d5d3..e8d73f5 100644 --- a/crates/maestro-daemon/src/registry_client.rs +++ b/crates/maestro-daemon/src/registry_client.rs @@ -41,6 +41,7 @@ pub struct RegistryClient { registry_url: String, daemon_id: String, hostname: String, + daemon_url: String, token: String, manager: SessionManager, } @@ -50,6 +51,7 @@ impl RegistryClient { registry_url: impl Into, daemon_id: impl Into, hostname: impl Into, + daemon_url: impl Into, token: impl Into, manager: SessionManager, ) -> Self { @@ -57,6 +59,7 @@ impl RegistryClient { registry_url: registry_url.into(), daemon_id: daemon_id.into(), hostname: hostname.into(), + daemon_url: daemon_url.into(), token: token.into(), manager, } @@ -91,6 +94,7 @@ impl RegistryClient { hostname: self.hostname.clone(), version: PROTOCOL_VERSION.into(), token: self.token.clone(), + daemon_url: self.daemon_url.clone(), }; sink.send(Message::Text(serde_json::to_string(&hello)?)) .await?; diff --git a/crates/maestro-daemon/tests/registry_client.rs b/crates/maestro-daemon/tests/registry_client.rs index d330cc2..1b5f569 100644 --- a/crates/maestro-daemon/tests/registry_client.rs +++ b/crates/maestro-daemon/tests/registry_client.rs @@ -57,6 +57,7 @@ async fn daemon_registers_and_streams_snapshot() { reg.url.clone(), "laptop", "naps-laptop", + "ws://127.0.0.1:8081", TOKEN, fx.manager.clone(), ); @@ -113,6 +114,7 @@ async fn daemon_reconnects_after_registry_drop() { url.clone(), "laptop", "naps-laptop", + "ws://127.0.0.1:8081", TOKEN, fx.manager.clone(), ); @@ -153,6 +155,7 @@ async fn registry_create_session_round_trips() { reg.url.clone(), "laptop", "naps-laptop", + "ws://127.0.0.1:8081", TOKEN, fx.manager.clone(), ); diff --git a/crates/maestro-protocol/src/lib.rs b/crates/maestro-protocol/src/lib.rs index 2c08f68..d5339db 100644 --- a/crates/maestro-protocol/src/lib.rs +++ b/crates/maestro-protocol/src/lib.rs @@ -95,6 +95,11 @@ pub enum DaemonOutbound { hostname: String, version: String, token: String, + /// Reachable URL of this daemon's per-session WS endpoint, as the + /// operator wants web clients to dial it. The registry hands this back + /// in `GET /api/sessions/:id/connect-info` so the client can connect + /// directly. Example: `ws://10.0.0.1:8081`. + daemon_url: String, }, SessionsSnapshot { sessions: Vec, @@ -135,6 +140,62 @@ pub enum DaemonInbound { }, } +// --------------------------------------------------------------------------- +// Client ↔ Registry API types +// --------------------------------------------------------------------------- + +/// Public, read-only view of a connected daemon. Returned by `GET /api/daemons` +/// and included in `RegistryEvent::Snapshot` / `RegistryEvent::DaemonAdded`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct DaemonInfo { + pub daemon_id: String, + pub hostname: String, + pub version: String, + pub daemon_url: String, +} + +/// Response body for `GET /api/sessions/:id/connect-info`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct ConnectInfo { + pub daemon_url: String, + pub jwt: String, +} + +/// Messages broadcast by the registry to web clients over `WS /api/events`. +/// +/// A subscriber receives a single `Snapshot` immediately on connect, then a +/// stream of deltas as daemons connect/disconnect and their session sets +/// change. Skipped events (broadcast channel lag) surface as `Lagged`. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type")] +pub enum RegistryEvent { + Snapshot { + daemons: Vec, + sessions: Vec, + }, + DaemonAdded { + daemon: DaemonInfo, + }, + DaemonRemoved { + daemon_id: String, + }, + SessionAdded { + session: Session, + }, + SessionRemoved { + daemon_id: String, + id: Uuid, + }, + SessionUpdated { + session: Session, + }, + /// Server-issued notification that the broadcast queue dropped frames. + /// Clients should re-fetch a full snapshot. + Lagged { + skipped: u64, + }, +} + // --------------------------------------------------------------------------- // Client ↔ Daemon control messages (text frames on the attach WS) // --------------------------------------------------------------------------- @@ -268,6 +329,7 @@ mod tests { hostname: "naps-laptop".into(), version: PROTOCOL_VERSION.into(), token: "shhh".into(), + daemon_url: "ws://10.0.0.1:8081".into(), }; let v = serde_json::to_value(&hello).unwrap(); assert_eq!(v["type"], "Hello"); @@ -372,6 +434,67 @@ mod tests { assert!(err.to_string().contains("Bogus"), "{}", err); } + #[test] + fn daemon_info_round_trips() { + let d = DaemonInfo { + daemon_id: "laptop".into(), + hostname: "naps-laptop".into(), + version: "0.1.0".into(), + daemon_url: "ws://10.0.0.1:8081".into(), + }; + assert_eq!(d, round_trip(&d)); + } + + #[test] + fn connect_info_round_trips() { + let c = ConnectInfo { + daemon_url: "ws://10.0.0.1:8081".into(), + jwt: "eyJ...".into(), + }; + assert_eq!(c, round_trip(&c)); + } + + #[test] + fn registry_event_variants_round_trip_and_tag() { + let daemon = DaemonInfo { + daemon_id: "laptop".into(), + hostname: "naps-laptop".into(), + version: "0.1.0".into(), + daemon_url: "ws://10.0.0.1:8081".into(), + }; + let session = sample_session(); + let variants = vec![ + RegistryEvent::Snapshot { + daemons: vec![daemon.clone()], + sessions: vec![session.clone()], + }, + RegistryEvent::DaemonAdded { + daemon: daemon.clone(), + }, + RegistryEvent::DaemonRemoved { + daemon_id: "laptop".into(), + }, + RegistryEvent::SessionAdded { + session: session.clone(), + }, + RegistryEvent::SessionRemoved { + daemon_id: "laptop".into(), + id: session.id, + }, + RegistryEvent::SessionUpdated { session }, + RegistryEvent::Lagged { skipped: 12 }, + ]; + for v in variants { + assert_eq!(v, round_trip(&v)); + } + + let json = serde_json::to_value(RegistryEvent::DaemonRemoved { + daemon_id: "x".into(), + }) + .unwrap(); + assert_eq!(json["type"], "DaemonRemoved"); + } + #[test] fn jwt_claims_round_trip_and_iat_optional() { let c = JwtClaims { diff --git a/crates/maestro-registry/Cargo.toml b/crates/maestro-registry/Cargo.toml index 0451e8f..532cdba 100644 --- a/crates/maestro-registry/Cargo.toml +++ b/crates/maestro-registry/Cargo.toml @@ -12,6 +12,7 @@ axum.workspace = true chrono.workspace = true dirs.workspace = true futures-util.workspace = true +jsonwebtoken.workspace = true maestro-protocol.workspace = true rand.workspace = true serde.workspace = true @@ -23,6 +24,7 @@ tracing.workspace = true uuid.workspace = true [dev-dependencies] +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "cookies"] } tempfile.workspace = true tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time", "net", "io-util", "test-util"] } tokio-tungstenite.workspace = true diff --git a/crates/maestro-registry/src/auth.rs b/crates/maestro-registry/src/auth.rs new file mode 100644 index 0000000..2525459 --- /dev/null +++ b/crates/maestro-registry/src/auth.rs @@ -0,0 +1,163 @@ +//! Cookie-based auth for the client-facing HTTP/WS surface. +//! +//! On the first request the registry sets an httpOnly cookie whose value is a +//! JWT signed with the shared `MAESTRO_TOKEN` (HS256). Subsequent requests +//! must present a cookie whose signature verifies; missing or tampered +//! cookies get a fresh cookie issued (effectively re-bootstrapping a +//! first-time client) and are still allowed through. The cookie is the second +//! factor against anyone routing to the bind address; per spec §Auth the +//! perimeter is the operator's network. +//! +//! Implementation notes: +//! +//! - We reuse `jsonwebtoken` so we don't pull in a cookie/HMAC crate just for +//! this one signature. +//! - The middleware passes the inbound `Cookie` header through to the inner +//! handler unchanged, and adds a `Set-Cookie` response header if the inbound +//! cookie was missing or invalid. + +use std::sync::Arc; + +use axum::body::Body; +use axum::extract::State; +use axum::http::{header, HeaderValue, Request, StatusCode}; +use axum::middleware::Next; +use axum::response::{IntoResponse, Response}; +use chrono::{Duration, Utc}; +use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation}; +use serde::{Deserialize, Serialize}; + +/// Name of the auth cookie surfaced to web clients. +pub const COOKIE_NAME: &str = "maestro_session"; + +/// 30 days. Cookie is the second factor — long enough not to log users out; +/// short enough to bound the blast radius if the secret rotates. +const COOKIE_TTL_DAYS: i64 = 30; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct CookieClaims { + /// Subject — currently always `"client"` because this is single-user. + sub: String, + exp: i64, + iat: i64, +} + +/// Issue a fresh cookie value (a JWT) signed with `secret`. +pub fn issue_cookie(secret: &str) -> String { + let now = Utc::now(); + let claims = CookieClaims { + sub: "client".into(), + iat: now.timestamp(), + exp: (now + Duration::days(COOKIE_TTL_DAYS)).timestamp(), + }; + let header = Header::new(Algorithm::HS256); + let key = EncodingKey::from_secret(secret.as_bytes()); + encode(&header, &claims, &key).expect("encode auth cookie") +} + +/// Returns true if `value` is a JWT validly signed by `secret`. +pub fn verify_cookie(secret: &str, value: &str) -> bool { + let key = DecodingKey::from_secret(secret.as_bytes()); + let validation = Validation::new(Algorithm::HS256); + decode::(value, &key, &validation).is_ok() +} + +/// Format the full `Set-Cookie` header value. +fn set_cookie_header(value: &str) -> String { + format!( + "{name}={value}; HttpOnly; SameSite=Strict; Path=/; Max-Age={age}", + name = COOKIE_NAME, + age = COOKIE_TTL_DAYS * 24 * 60 * 60, + ) +} + +/// Find our cookie inside an inbound `Cookie:` header. Header may contain +/// many `name=value` pairs separated by `;`. +pub fn extract_cookie<'a>(header_value: &'a str, name: &str) -> Option<&'a str> { + for part in header_value.split(';') { + let part = part.trim(); + if let Some(eq) = part.find('=') { + let (k, v) = part.split_at(eq); + if k == name { + return Some(&v[1..]); + } + } + } + None +} + +/// Shared state for the auth middleware: just the signing secret. +#[derive(Debug, Clone)] +pub struct AuthState { + pub token: Arc, +} + +/// Axum middleware that enforces / bootstraps the client auth cookie. +/// +/// - Valid cookie present → pass through, no `Set-Cookie`. +/// - Cookie missing → issue a fresh one and pass through. +/// - Cookie present but tampered/expired → issue a fresh one and pass through. +/// (Spec is intentionally lenient — perimeter is the operator's network.) +pub async fn cookie_middleware( + State(state): State, + request: Request, + next: Next, +) -> Response { + let token = state.token.clone(); + let inbound_cookie = request + .headers() + .get(header::COOKIE) + .and_then(|h| h.to_str().ok()) + .and_then(|v| extract_cookie(v, COOKIE_NAME).map(|s| s.to_string())); + + let needs_new_cookie = match &inbound_cookie { + Some(value) => !verify_cookie(token.as_str(), value), + None => true, + }; + + let mut response = next.run(request).await; + + if needs_new_cookie { + let fresh = issue_cookie(token.as_str()); + match HeaderValue::from_str(&set_cookie_header(&fresh)) { + Ok(hv) => { + response.headers_mut().append(header::SET_COOKIE, hv); + } + Err(_) => { + return (StatusCode::INTERNAL_SERVER_ERROR, "cookie encode failed").into_response(); + } + } + } + + response +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn round_trip() { + let token = "shared-secret-aaaa"; + let value = issue_cookie(token); + assert!(verify_cookie(token, &value)); + assert!(!verify_cookie("different-secret", &value)); + } + + #[test] + fn extract_cookie_finds_named_value() { + let h = "foo=1; maestro_session=abc.def.ghi; bar=2"; + assert_eq!(extract_cookie(h, "maestro_session"), Some("abc.def.ghi")); + assert_eq!(extract_cookie(h, "foo"), Some("1")); + assert_eq!(extract_cookie(h, "missing"), None); + } + + #[test] + fn set_cookie_header_includes_attributes() { + let h = set_cookie_header("abc"); + assert!(h.contains("HttpOnly")); + assert!(h.contains("SameSite=Strict")); + assert!(h.contains("Path=/")); + assert!(h.starts_with("maestro_session=abc;")); + } +} diff --git a/crates/maestro-registry/src/client_api.rs b/crates/maestro-registry/src/client_api.rs new file mode 100644 index 0000000..fa1e946 --- /dev/null +++ b/crates/maestro-registry/src/client_api.rs @@ -0,0 +1,199 @@ +//! Client-facing HTTP API routes (NAP-215). +//! +//! Mounts under the same router as the daemon-facing `WS /daemon`. The cookie +//! middleware in [`crate::auth`] gates everything except the WS upgrade for +//! `/daemon` (daemons authenticate via the shared token in `Hello`). + +use std::sync::Arc; +use std::time::Duration; + +use axum::extract::{ws::WebSocketUpgrade, Path, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::routing::{any, get}; +use axum::{Json, Router}; +use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, Session}; +use uuid::Uuid; + +use crate::events::events_ws; +use crate::jwt::issue_attach_jwt; +use crate::state::{Registry, RegistryError, RPC_TIMEOUT}; + +/// Shared state plumbed into client-facing handlers. +#[derive(Debug, Clone)] +pub struct ClientApiState { + pub registry: Registry, + pub token: Arc, + pub rpc_timeout: Duration, +} + +/// Mount client-facing routes on the given router. Caller is responsible for +/// applying the cookie middleware (we do that in [`crate::server::router`]). +pub fn routes(state: ClientApiState) -> Router { + Router::new() + .route("/", get(index)) + .route("/api/daemons", get(list_daemons)) + .route("/api/sessions", get(list_sessions).post(create_session)) + .route("/api/sessions/:id", axum::routing::delete(delete_session)) + .route("/api/sessions/:id/connect-info", get(session_connect_info)) + .route("/api/events", any(events_ws_handler)) + .with_state(state) +} + +async fn index() -> impl IntoResponse { + // Placeholder until the real web bundle lands in NAP-218. + ( + [("content-type", "text/html; charset=utf-8")], + "maestro

maestro registry

\ +

The web client is not bundled yet — see NAP-218.

", + ) +} + +async fn list_daemons(State(s): State) -> Json> { + let infos: Vec = s + .registry + .daemons() + .await + .into_iter() + .map(Into::into) + .collect(); + Json(infos) +} + +async fn list_sessions(State(s): State) -> Json> { + Json(s.registry.sessions().await) +} + +async fn create_session( + State(s): State, + Json(body): Json, +) -> Response { + let daemon = body.daemon_id.clone(); + let outcome = + tokio::time::timeout(s.rpc_timeout, s.registry.create_session(&daemon, body)).await; + match outcome { + Err(_) => ( + StatusCode::GATEWAY_TIMEOUT, + Json(error_body("timeout", "daemon did not reply in time")), + ) + .into_response(), + Ok(Err(RegistryError::DaemonNotConnected(_))) => ( + StatusCode::NOT_FOUND, + Json(error_body("unknown_daemon", "daemon is not connected")), + ) + .into_response(), + Ok(Err(RegistryError::Timeout(_))) => ( + StatusCode::GATEWAY_TIMEOUT, + Json(error_body("timeout", "daemon did not reply in time")), + ) + .into_response(), + Ok(Err(e)) => ( + StatusCode::BAD_GATEWAY, + Json(error_body("transport", &e.to_string())), + ) + .into_response(), + Ok(Ok(Err(rpc_err))) => ( + StatusCode::BAD_GATEWAY, + Json(error_body(&rpc_err.0.code, &rpc_err.0.message)), + ) + .into_response(), + Ok(Ok(Ok(session))) => (StatusCode::CREATED, Json(session)).into_response(), + } +} + +async fn delete_session(State(s): State, Path(id): Path) -> Response { + let Some(daemon) = s.registry.daemon_for_session(id).await else { + return ( + StatusCode::NOT_FOUND, + Json(error_body("unknown_session", "no such session")), + ) + .into_response(); + }; + + let outcome = tokio::time::timeout(s.rpc_timeout, s.registry.kill_session(&daemon, id)).await; + match outcome { + Err(_) => ( + StatusCode::GATEWAY_TIMEOUT, + Json(error_body("timeout", "daemon did not reply in time")), + ) + .into_response(), + Ok(Err(RegistryError::DaemonNotConnected(_))) => ( + StatusCode::NOT_FOUND, + Json(error_body("unknown_daemon", "daemon is not connected")), + ) + .into_response(), + Ok(Err(RegistryError::Timeout(_))) => ( + StatusCode::GATEWAY_TIMEOUT, + Json(error_body("timeout", "daemon did not reply in time")), + ) + .into_response(), + Ok(Err(e)) => ( + StatusCode::BAD_GATEWAY, + Json(error_body("transport", &e.to_string())), + ) + .into_response(), + Ok(Ok(Err(rpc_err))) => ( + StatusCode::BAD_GATEWAY, + Json(error_body(&rpc_err.0.code, &rpc_err.0.message)), + ) + .into_response(), + Ok(Ok(Ok(()))) => StatusCode::NO_CONTENT.into_response(), + } +} + +async fn session_connect_info(State(s): State, Path(id): Path) -> Response { + let Some(daemon) = s.registry.daemon_for_session(id).await else { + return ( + StatusCode::NOT_FOUND, + Json(error_body("unknown_session", "no such session")), + ) + .into_response(); + }; + let Some(daemon_url) = s.registry.daemon_url(&daemon).await else { + return ( + StatusCode::NOT_FOUND, + Json(error_body("unknown_daemon", "owning daemon is offline")), + ) + .into_response(); + }; + let jwt = match issue_attach_jwt(s.token.as_str(), &daemon, id) { + Ok(t) => t, + Err(e) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(error_body("jwt_failed", &e.to_string())), + ) + .into_response(); + } + }; + Json(ConnectInfo { daemon_url, jwt }).into_response() +} + +async fn events_ws_handler( + State(s): State, + ws: WebSocketUpgrade, +) -> impl IntoResponse { + let registry = s.registry.clone(); + ws.on_upgrade(move |socket| async move { + let _ = events_ws(socket, registry).await; + }) +} + +fn error_body(code: &str, message: &str) -> serde_json::Value { + serde_json::json!({ + "error": { + "code": code, + "message": message, + }, + }) +} + +/// Convenience: produce a `ClientApiState` from a `Registry` and shared token. +/// Uses the production [`RPC_TIMEOUT`]. +pub fn state_for(registry: Registry, token: String) -> ClientApiState { + ClientApiState { + registry, + token: Arc::new(token), + rpc_timeout: RPC_TIMEOUT, + } +} diff --git a/crates/maestro-registry/src/events.rs b/crates/maestro-registry/src/events.rs new file mode 100644 index 0000000..a4bea2e --- /dev/null +++ b/crates/maestro-registry/src/events.rs @@ -0,0 +1,59 @@ +//! `WS /api/events` — registry-side fan-out of catalog + presence deltas. +//! +//! On connect we immediately push a [`RegistryEvent::Snapshot`] of the current +//! state, then forward every broadcast event until the client disconnects. +//! Lagged subscribers receive a [`RegistryEvent::Lagged`] notice with the +//! count of dropped frames — the convention is for the client to re-fetch +//! `GET /api/sessions` on receipt. + +use axum::extract::ws::{Message, WebSocket}; +use futures_util::{SinkExt, StreamExt}; +use maestro_protocol::RegistryEvent; +use tokio::sync::broadcast::error::RecvError; + +use crate::state::Registry; + +pub async fn events_ws(socket: WebSocket, registry: Registry) -> Result<(), axum::Error> { + let (mut sink, mut stream) = socket.split(); + + let (snapshot, mut rx) = registry.subscribe().await; + sink.send(Message::Text( + serde_json::to_string(&snapshot).expect("snapshot"), + )) + .await?; + + loop { + tokio::select! { + ev = rx.recv() => { + match ev { + Ok(event) => { + let text = serde_json::to_string(&event).expect("event encode"); + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + Err(RecvError::Lagged(n)) => { + let text = serde_json::to_string(&RegistryEvent::Lagged { skipped: n }) + .expect("lagged encode"); + if sink.send(Message::Text(text)).await.is_err() { + break; + } + } + Err(RecvError::Closed) => break, + } + } + client_frame = stream.next() => { + // Treat anything from the client as either a close or a ping + // we don't care about; the events WS is server-push only. + match client_frame { + None => break, + Some(Err(_)) => break, + Some(Ok(Message::Close(_))) => break, + Some(Ok(_)) => {} + } + } + } + } + + Ok(()) +} diff --git a/crates/maestro-registry/src/handler.rs b/crates/maestro-registry/src/handler.rs index c47404f..0baa1f1 100644 --- a/crates/maestro-registry/src/handler.rs +++ b/crates/maestro-registry/src/handler.rs @@ -37,13 +37,14 @@ pub async fn handle_socket( None => return None, }; - let (daemon_id, hostname, version, token) = match hello { + let (daemon_id, hostname, version, token, daemon_url) = match hello { DaemonOutbound::Hello { daemon_id, hostname, version, token, - } => (daemon_id, hostname, version, token), + daemon_url, + } => (daemon_id, hostname, version, token, daemon_url), _ => { let _ = send_error(&mut sink, "expected_hello", "first message must be Hello").await; return None; @@ -57,7 +58,7 @@ pub async fn handle_socket( let (out_tx, mut out_rx) = mpsc::channel::(OUTBOUND_BUFFER); registry - .register(daemon_id.clone(), hostname, version, out_tx) + .register(daemon_id.clone(), hostname, version, daemon_url, out_tx) .await; let _ = sink diff --git a/crates/maestro-registry/src/jwt.rs b/crates/maestro-registry/src/jwt.rs new file mode 100644 index 0000000..a92ceed --- /dev/null +++ b/crates/maestro-registry/src/jwt.rs @@ -0,0 +1,54 @@ +//! Registry-side JWT issuance for the per-session attach endpoint. +//! +//! The daemon validates these tokens locally with the same shared secret — +//! the registry doesn't see any callback on the attach hot path. +//! Spec §Auth, §"Wire protocol (Client↔Daemon)". + +use chrono::{Duration, Utc}; +use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; +use maestro_protocol::JwtClaims; +use uuid::Uuid; + +/// Spec §Auth: ~5 minutes. +pub const ATTACH_JWT_TTL: Duration = Duration::minutes(5); + +/// Mint a short-lived attach JWT for `(daemon_id, session_id)`. +pub fn issue_attach_jwt( + secret: &str, + daemon_id: &str, + session_id: Uuid, +) -> Result { + let now = Utc::now(); + let claims = JwtClaims { + session_id, + daemon_id: daemon_id.to_string(), + exp: (now + ATTACH_JWT_TTL).timestamp(), + iat: Some(now.timestamp()), + }; + let header = Header::new(Algorithm::HS256); + let key = EncodingKey::from_secret(secret.as_bytes()); + encode(&header, &claims, &key) +} + +#[cfg(test)] +mod tests { + use super::*; + use jsonwebtoken::{decode, DecodingKey, Validation}; + + #[test] + fn issued_token_verifies_with_shared_secret() { + let secret = "shared-secret"; + let daemon = "laptop"; + let session = Uuid::new_v4(); + let token = issue_attach_jwt(secret, daemon, session).unwrap(); + let decoded = decode::( + &token, + &DecodingKey::from_secret(secret.as_bytes()), + &Validation::new(Algorithm::HS256), + ) + .unwrap(); + assert_eq!(decoded.claims.daemon_id, daemon); + assert_eq!(decoded.claims.session_id, session); + assert!(decoded.claims.exp > Utc::now().timestamp()); + } +} diff --git a/crates/maestro-registry/src/lib.rs b/crates/maestro-registry/src/lib.rs index b341ec6..4829c6d 100644 --- a/crates/maestro-registry/src/lib.rs +++ b/crates/maestro-registry/src/lib.rs @@ -1,16 +1,28 @@ //! The always-on Maestro registry. //! -//! This crate covers the daemon-facing surface only — there is no client HTTP -//! API in NAP-213. See `docs/superpowers/specs/2026-05-24-maestro-design.md` -//! §Registry and §"Wire protocol (D↔R)". +//! Covers two surfaces: +//! +//! - Daemon-facing `WS /daemon` (NAP-213): presence + session catalog. +//! - Client-facing HTTP/WS API (NAP-215): list daemons/sessions, spawn/kill +//! sessions (forwarded to the owning daemon via WS), issue per-session +//! attach JWTs, and broadcast catalog deltas over `WS /api/events`. +//! +//! See `docs/superpowers/specs/2026-05-24-maestro-design.md` §Registry, +//! §"Wire protocol", and §Auth. #![deny(missing_debug_implementations)] +pub mod auth; +pub mod client_api; +pub mod events; pub mod handler; +pub mod jwt; pub mod server; pub mod state; pub mod token; -pub use server::{router, router_with_timeout, RegistryServer}; +pub use auth::{issue_cookie, verify_cookie, COOKIE_NAME}; +pub use jwt::{issue_attach_jwt, ATTACH_JWT_TTL}; +pub use server::{router, router_full, router_with_timeout, RegistryServer}; pub use state::{DaemonView, Registry, RegistryError, RpcError, HEARTBEAT_TIMEOUT, RPC_TIMEOUT}; pub use token::{ensure_token, RegistryConfig, TokenError}; diff --git a/crates/maestro-registry/src/server.rs b/crates/maestro-registry/src/server.rs index 06cac6a..6bfb744 100644 --- a/crates/maestro-registry/src/server.rs +++ b/crates/maestro-registry/src/server.rs @@ -1,7 +1,5 @@ -//! axum router + listener for the daemon-facing WS endpoint. -//! -//! The router exposes only `WS /daemon` in NAP-213. Client-facing HTTP arrives -//! in NAP-215. +//! axum router + listener for the registry, including both the daemon-facing +//! `WS /daemon` endpoint (NAP-213) and the client-facing HTTP API (NAP-215). use std::net::SocketAddr; use std::sync::Arc; @@ -11,41 +9,70 @@ use axum::extract::ws::WebSocketUpgrade; use axum::extract::State; use axum::response::IntoResponse; use axum::routing::any; -use axum::Router; +use axum::{middleware, Router}; use tokio::net::TcpListener; +use crate::auth::{cookie_middleware, AuthState}; +use crate::client_api::{routes as client_routes, ClientApiState}; use crate::handler::handle_socket; -use crate::state::{Registry, HEARTBEAT_TIMEOUT}; +use crate::state::{Registry, HEARTBEAT_TIMEOUT, RPC_TIMEOUT}; #[derive(Debug, Clone)] -struct AppState { +struct DaemonAppState { registry: Registry, token: Arc, heartbeat_timeout: Duration, } -/// Build the daemon-facing router with the production heartbeat timeout. +/// Build the production router (daemon WS + client HTTP API) with default +/// timeouts. pub fn router(registry: Registry, token: String) -> Router { - router_with_timeout(registry, token, HEARTBEAT_TIMEOUT) + router_full(registry, token, HEARTBEAT_TIMEOUT, RPC_TIMEOUT) } -/// Build the daemon-facing router with a custom heartbeat timeout. Mainly for -/// integration tests that want to verify the timeout path without waiting 30s. +/// Variant for integration tests that want a short heartbeat. Keeps the +/// existing 4-arg call sites in `daemon_ws.rs` working. pub fn router_with_timeout( registry: Registry, token: String, heartbeat_timeout: Duration, ) -> Router { - Router::new() - .route("/daemon", any(daemon_ws)) - .with_state(AppState { - registry, - token: Arc::new(token), - heartbeat_timeout, - }) + router_full(registry, token, heartbeat_timeout, RPC_TIMEOUT) } -async fn daemon_ws(State(state): State, ws: WebSocketUpgrade) -> impl IntoResponse { +/// Full builder; lets tests override every knob. +pub fn router_full( + registry: Registry, + token: String, + heartbeat_timeout: Duration, + rpc_timeout: Duration, +) -> Router { + let token_arc = Arc::new(token.clone()); + + let daemon_router = Router::new() + .route("/daemon", any(daemon_ws)) + .with_state(DaemonAppState { + registry: registry.clone(), + token: token_arc.clone(), + heartbeat_timeout, + }); + + let client_router = client_routes(ClientApiState { + registry, + token: token_arc.clone(), + rpc_timeout, + }) + .layer(middleware::from_fn_with_state( + AuthState { + token: token_arc.clone(), + }, + cookie_middleware, + )); + + Router::new().merge(daemon_router).merge(client_router) +} + +async fn daemon_ws(State(state): State, ws: WebSocketUpgrade) -> impl IntoResponse { let registry = state.registry.clone(); let token = state.token.clone(); let timeout = state.heartbeat_timeout; diff --git a/crates/maestro-registry/src/state.rs b/crates/maestro-registry/src/state.rs index e0f8e8a..0b772f0 100644 --- a/crates/maestro-registry/src/state.rs +++ b/crates/maestro-registry/src/state.rs @@ -14,11 +14,17 @@ use std::sync::Arc; use std::time::Duration; use maestro_protocol::{ - CreateSessionRequest, DaemonInbound, ErrorInfo, ReplyTo, RpcResult, Session, + CreateSessionRequest, DaemonInbound, DaemonInfo, ErrorInfo, RegistryEvent, ReplyTo, RpcResult, + Session, }; -use tokio::sync::{mpsc, oneshot, Mutex}; +use tokio::sync::{broadcast, mpsc, oneshot, Mutex}; use uuid::Uuid; +/// Capacity of the registry-wide broadcast channel that feeds +/// `WS /api/events` subscribers. Subscribers that fall this far behind get a +/// `Lagged` event and are expected to resync via `GET /api/sessions`. +const EVENT_CHANNEL_CAPACITY: usize = 256; + /// How long a daemon can go without sending us *any* message before it is /// considered offline and its sessions are dropped from the catalog. pub const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(30); @@ -46,18 +52,32 @@ pub enum RegistryError { #[error("daemon rpc failed: {0:?}")] pub struct RpcError(pub ErrorInfo); -/// Public, read-only view of a single connected daemon. +/// Public, read-only view of a single connected daemon. Convertible to the +/// protocol-level `DaemonInfo` for inclusion in HTTP responses and events. #[derive(Debug, Clone)] pub struct DaemonView { pub daemon_id: String, pub hostname: String, pub version: String, + pub daemon_url: String, +} + +impl From for DaemonInfo { + fn from(v: DaemonView) -> Self { + Self { + daemon_id: v.daemon_id, + hostname: v.hostname, + version: v.version, + daemon_url: v.daemon_url, + } + } } /// Shared, cloneable registry handle. #[derive(Debug, Clone)] pub struct Registry { inner: Arc>, + events: broadcast::Sender, } #[derive(Debug, Default)] @@ -73,16 +93,52 @@ struct State { struct DaemonEntry { hostname: String, version: String, + daemon_url: String, outbound: mpsc::Sender, } +impl DaemonEntry { + fn to_view(&self, daemon_id: &str) -> DaemonView { + DaemonView { + daemon_id: daemon_id.to_string(), + hostname: self.hostname.clone(), + version: self.version.clone(), + daemon_url: self.daemon_url.clone(), + } + } +} + impl Registry { pub fn new() -> Self { + let (events, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY); Self { inner: Arc::new(Mutex::new(State::default())), + events, } } + /// Subscribe to the registry's broadcast event stream. The subscriber gets + /// a one-shot `Snapshot` of the current catalog first, then live deltas. + pub async fn subscribe(&self) -> (RegistryEvent, broadcast::Receiver) { + let state = self.inner.lock().await; + let snapshot = RegistryEvent::Snapshot { + daemons: state + .daemons + .iter() + .map(|(id, e)| e.to_view(id).into()) + .collect(), + sessions: state.sessions.values().cloned().collect(), + }; + let rx = self.events.subscribe(); + (snapshot, rx) + } + + fn broadcast(&self, event: RegistryEvent) { + // Send returns Err only when there are no subscribers; that's not a + // bug, so swallow. + let _ = self.events.send(event); + } + // ----------------------------------------------------------------------- // Connection lifecycle // ----------------------------------------------------------------------- @@ -95,26 +151,58 @@ impl Registry { daemon_id: String, hostname: String, version: String, + daemon_url: String, outbound: mpsc::Sender, ) { let mut state = self.inner.lock().await; - state.drop_daemon_sessions(&daemon_id); + let removed_sessions = state.drop_daemon_sessions(&daemon_id); state.daemons.insert( - daemon_id, + daemon_id.clone(), DaemonEntry { - hostname, - version, + hostname: hostname.clone(), + version: version.clone(), + daemon_url: daemon_url.clone(), outbound, }, ); + drop(state); + for id in removed_sessions { + self.broadcast(RegistryEvent::SessionRemoved { + daemon_id: daemon_id.clone(), + id, + }); + } + self.broadcast(RegistryEvent::DaemonAdded { + daemon: DaemonInfo { + daemon_id, + hostname, + version, + daemon_url, + }, + }); } /// Remove the daemon and any sessions it owned. Also fails any pending /// RPCs that were sent to it. pub async fn deregister(&self, daemon_id: &str) { - let mut state = self.inner.lock().await; - state.daemons.remove(daemon_id); - state.drop_daemon_sessions(daemon_id); + let removed_sessions; + let had_daemon; + { + let mut state = self.inner.lock().await; + had_daemon = state.daemons.remove(daemon_id).is_some(); + removed_sessions = state.drop_daemon_sessions(daemon_id); + } + for id in removed_sessions { + self.broadcast(RegistryEvent::SessionRemoved { + daemon_id: daemon_id.to_string(), + id, + }); + } + if had_daemon { + self.broadcast(RegistryEvent::DaemonRemoved { + daemon_id: daemon_id.to_string(), + }); + } } // ----------------------------------------------------------------------- @@ -122,30 +210,75 @@ impl Registry { // ----------------------------------------------------------------------- pub async fn apply_snapshot(&self, daemon_id: &str, sessions: Vec) { - let mut state = self.inner.lock().await; - state.drop_daemon_sessions(daemon_id); - for s in sessions { - state.sessions.insert((daemon_id.to_string(), s.id), s); + let (removed, added) = { + let mut state = self.inner.lock().await; + let removed = state.drop_daemon_sessions(daemon_id); + let mut added = Vec::with_capacity(sessions.len()); + for s in sessions { + added.push(s.clone()); + state.sessions.insert((daemon_id.to_string(), s.id), s); + } + (removed, added) + }; + for id in removed { + self.broadcast(RegistryEvent::SessionRemoved { + daemon_id: daemon_id.to_string(), + id, + }); + } + for s in added { + self.broadcast(RegistryEvent::SessionAdded { session: s }); } } pub async fn add_session(&self, daemon_id: &str, session: Session) { - let mut state = self.inner.lock().await; - state - .sessions - .insert((daemon_id.to_string(), session.id), session); + { + let mut state = self.inner.lock().await; + state + .sessions + .insert((daemon_id.to_string(), session.id), session.clone()); + } + self.broadcast(RegistryEvent::SessionAdded { session }); } pub async fn remove_session(&self, daemon_id: &str, id: Uuid) { - let mut state = self.inner.lock().await; - state.sessions.remove(&(daemon_id.to_string(), id)); + let removed = { + let mut state = self.inner.lock().await; + state.sessions.remove(&(daemon_id.to_string(), id)) + }; + if removed.is_some() { + self.broadcast(RegistryEvent::SessionRemoved { + daemon_id: daemon_id.to_string(), + id, + }); + } } pub async fn update_session(&self, daemon_id: &str, session: Session) { - let mut state = self.inner.lock().await; + { + let mut state = self.inner.lock().await; + state + .sessions + .insert((daemon_id.to_string(), session.id), session.clone()); + } + self.broadcast(RegistryEvent::SessionUpdated { session }); + } + + /// Look up the daemon that owns the given session id. + pub async fn daemon_for_session(&self, session_id: Uuid) -> Option { + let state = self.inner.lock().await; state .sessions - .insert((daemon_id.to_string(), session.id), session); + .iter() + .find(|((_, id), _)| *id == session_id) + .map(|((d, _), _)| d.clone()) + } + + /// Resolve a daemon's connect URL (the operator-advertised value from its + /// Hello frame). + pub async fn daemon_url(&self, daemon_id: &str) -> Option { + let state = self.inner.lock().await; + state.daemons.get(daemon_id).map(|e| e.daemon_url.clone()) } /// Resolve a CreateSession RPC reply. Returns true if a waiter was found. @@ -238,15 +371,7 @@ impl Registry { pub async fn daemons(&self) -> Vec { let state = self.inner.lock().await; - state - .daemons - .iter() - .map(|(id, e)| DaemonView { - daemon_id: id.clone(), - hostname: e.hostname.clone(), - version: e.version.clone(), - }) - .collect() + state.daemons.iter().map(|(id, e)| e.to_view(id)).collect() } pub async fn sessions(&self) -> Vec { @@ -272,8 +397,19 @@ impl Default for Registry { } impl State { - fn drop_daemon_sessions(&mut self, daemon_id: &str) { - self.sessions.retain(|(d, _), _| d.as_str() != daemon_id); + /// Removes every session belonging to `daemon_id`, returning their ids so + /// the caller can emit `SessionRemoved` events outside the lock. + fn drop_daemon_sessions(&mut self, daemon_id: &str) -> Vec { + let mut removed = Vec::new(); + self.sessions.retain(|(d, id), _| { + if d.as_str() == daemon_id { + removed.push(*id); + false + } else { + true + } + }); + removed } } diff --git a/crates/maestro-registry/tests/daemon_ws.rs b/crates/maestro-registry/tests/daemon_ws.rs index fb46549..e8e3734 100644 --- a/crates/maestro-registry/tests/daemon_ws.rs +++ b/crates/maestro-registry/tests/daemon_ws.rs @@ -82,6 +82,7 @@ fn hello(token: &str) -> DaemonOutbound { hostname: "naps-laptop".into(), version: "0.1.0".into(), token: token.into(), + daemon_url: "ws://127.0.0.1:8081".into(), } } diff --git a/crates/maestro-registry/tests/http_api.rs b/crates/maestro-registry/tests/http_api.rs new file mode 100644 index 0000000..45519a0 --- /dev/null +++ b/crates/maestro-registry/tests/http_api.rs @@ -0,0 +1,722 @@ +//! Integration tests for the client-facing HTTP API (NAP-215). +//! +//! Spins up a registry on `:0`, plus a tiny mock-daemon WS client that exists +//! purely to satisfy `CreateSession` / `KillSession` round-trips. All +//! HTTP/WS calls go over the wire via `reqwest` and `tokio-tungstenite`. + +use std::collections::HashMap; +use std::future::Future; +use std::net::{Ipv4Addr, SocketAddr}; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use futures_util::{SinkExt, StreamExt}; +use jsonwebtoken::{decode, Algorithm, DecodingKey, Validation}; +use maestro_protocol::{ + ConnectInfo, CreateSessionRequest, DaemonInbound, DaemonInfo, DaemonOutbound, JwtClaims, + RegistryEvent, RpcResult, Session, SessionStatus, +}; +use maestro_registry::{router_full, Registry, COOKIE_NAME}; +use reqwest::Client; +use serde_json::Value; +use tokio::net::TcpListener; +use tokio::sync::{oneshot, Mutex}; +use tokio_tungstenite::tungstenite::Message as WsMessage; +use tokio_tungstenite::{ + connect_async, tungstenite::client::IntoClientRequest, MaybeTlsStream, WebSocketStream, +}; +use uuid::Uuid; + +const TOKEN: &str = "http-api-test-token"; +const TEST_HEARTBEAT: Duration = Duration::from_secs(30); + +// --------------------------------------------------------------------------- +// Harness +// --------------------------------------------------------------------------- + +struct Harness { + addr: SocketAddr, + registry: Registry, + _shutdown_tx: oneshot::Sender<()>, +} + +impl Harness { + fn base(&self) -> String { + format!("http://{}", self.addr) + } + + fn ws_base(&self) -> String { + format!("ws://{}", self.addr) + } +} + +async fn spawn(rpc_timeout: Duration) -> Harness { + let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); + let registry = Registry::new(); + let app = router_full( + registry.clone(), + TOKEN.to_string(), + TEST_HEARTBEAT, + rpc_timeout, + ); + let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>(); + tokio::spawn(async move { + let _ = axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = shutdown_rx.await; + }) + .await; + }); + Harness { + addr, + registry, + _shutdown_tx: shutdown_tx, + } +} + +// --------------------------------------------------------------------------- +// Mock daemon +// --------------------------------------------------------------------------- + +type WsClient = WebSocketStream>; + +#[derive(Clone, Copy)] +enum DaemonBehavior { + /// Replies immediately with the session the request asked for. + ReplyOk, + /// Never replies — exercises the registry's 504 timeout path. + Hang, +} + +#[derive(Clone)] +struct MockDaemon { + daemon_id: String, + daemon_url: String, + captured: Arc>>, +} + +impl MockDaemon { + fn new(daemon_id: &str, daemon_url: &str) -> Self { + Self { + daemon_id: daemon_id.into(), + daemon_url: daemon_url.into(), + captured: Arc::new(Mutex::new(Vec::new())), + } + } + + async fn spawn(self, addr: SocketAddr, behavior: DaemonBehavior) { + let url = format!("ws://{addr}/daemon"); + let (mut ws, _) = connect_async(url).await.expect("ws connect"); + let hello = DaemonOutbound::Hello { + daemon_id: self.daemon_id.clone(), + hostname: "mock-host".into(), + version: "0.1.0".into(), + token: TOKEN.into(), + daemon_url: self.daemon_url.clone(), + }; + ws.send(WsMessage::Text(serde_json::to_string(&hello).unwrap())) + .await + .unwrap(); + // consume the Ack + let _ = ws.next().await; + // send an empty initial snapshot so the registry's view is consistent + ws.send(WsMessage::Text( + serde_json::to_string(&DaemonOutbound::SessionsSnapshot { sessions: vec![] }).unwrap(), + )) + .await + .unwrap(); + + let captured = self.captured.clone(); + let daemon_id = self.daemon_id.clone(); + tokio::spawn(async move { + run_loop(ws, behavior, captured, daemon_id).await; + }); + } + + async fn captured(&self) -> Vec { + self.captured.lock().await.clone() + } +} + +async fn run_loop( + mut ws: WsClient, + behavior: DaemonBehavior, + captured: Arc>>, + daemon_id: String, +) { + while let Some(Ok(WsMessage::Text(text))) = ws.next().await { + let msg: DaemonInbound = match serde_json::from_str(&text) { + Ok(m) => m, + Err(_) => continue, + }; + captured.lock().await.push(msg.clone()); + + match (behavior, msg) { + (DaemonBehavior::ReplyOk, DaemonInbound::CreateSession { reply_to, request }) => { + let session = mock_session(&daemon_id, &request); + // Send SessionAdded BEFORE the RPC result so the registry's + // catalog is updated by the time the HTTP caller resumes. + // The registry processes inbound frames serially in order, + // so this ordering guarantee holds end-to-end. + let _ = ws + .send(WsMessage::Text( + serde_json::to_string(&DaemonOutbound::SessionAdded { + session: session.clone(), + }) + .unwrap(), + )) + .await; + let reply = DaemonOutbound::CreateSessionResult { + reply_to, + result: RpcResult::Ok { value: session }, + }; + let _ = ws + .send(WsMessage::Text(serde_json::to_string(&reply).unwrap())) + .await; + } + (DaemonBehavior::ReplyOk, DaemonInbound::KillSession { reply_to, id }) => { + let _ = ws + .send(WsMessage::Text( + serde_json::to_string(&DaemonOutbound::SessionRemoved { id }).unwrap(), + )) + .await; + let reply = DaemonOutbound::KillSessionResult { + reply_to, + result: RpcResult::Ok { value: () }, + }; + let _ = ws + .send(WsMessage::Text(serde_json::to_string(&reply).unwrap())) + .await; + } + (DaemonBehavior::Hang, _) => { + // intentionally do nothing + } + _ => {} + } + } +} + +fn mock_session(daemon_id: &str, request: &CreateSessionRequest) -> Session { + // If a worktree spec was attached, pretend the daemon "resolved" the cwd + // to /.worktrees/ so the test can assert on it. + let cwd = request + .worktree + .as_ref() + .map(|w| w.base_repo.join(".worktrees").join(&w.branch)) + .unwrap_or_else(|| request.cwd.clone()); + Session { + id: Uuid::new_v4(), + daemon_id: daemon_id.into(), + cwd, + command: request.command.clone(), + env: request.env.clone(), + metadata: request.metadata.clone(), + status: SessionStatus::Running, + created_at: Utc::now(), + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +async fn wait_for(timeout: Duration, mut cond: F) +where + F: FnMut() -> Fut, + Fut: Future, +{ + let deadline = Instant::now() + timeout; + while Instant::now() < deadline { + if cond().await { + return; + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + panic!("timed out after {timeout:?} waiting for condition"); +} + +fn client_with_cookies() -> Client { + reqwest::Client::builder() + .cookie_store(true) + .build() + .unwrap() +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn first_request_sets_cookie_and_subsequent_reuses_it() { + let h = spawn(Duration::from_secs(5)).await; + let client = client_with_cookies(); + + let r1 = client + .get(format!("{}/api/daemons", h.base())) + .send() + .await + .unwrap(); + assert!(r1.status().is_success()); + let cookies: Vec<_> = r1 + .headers() + .get_all(reqwest::header::SET_COOKIE) + .iter() + .collect(); + assert!( + !cookies.is_empty(), + "first request must include a Set-Cookie" + ); + let raw = cookies[0].to_str().unwrap(); + assert!(raw.contains(COOKIE_NAME)); + assert!(raw.contains("HttpOnly")); + assert!(raw.contains("SameSite=Strict")); + + let r2 = client + .get(format!("{}/api/daemons", h.base())) + .send() + .await + .unwrap(); + assert!(r2.status().is_success()); + let new_cookies: Vec<_> = r2 + .headers() + .get_all(reqwest::header::SET_COOKIE) + .iter() + .collect(); + assert!( + new_cookies.is_empty(), + "valid existing cookie should not be reissued, got {new_cookies:?}" + ); +} + +#[tokio::test] +async fn tampered_cookie_is_replaced_silently() { + let h = spawn(Duration::from_secs(5)).await; + let client = reqwest::Client::new(); + + let resp = client + .get(format!("{}/api/daemons", h.base())) + .header( + reqwest::header::COOKIE, + format!("{}=not-a-jwt", COOKIE_NAME), + ) + .send() + .await + .unwrap(); + assert!(resp.status().is_success()); + let cookies: Vec<_> = resp + .headers() + .get_all(reqwest::header::SET_COOKIE) + .iter() + .collect(); + assert!( + !cookies.is_empty(), + "tampered cookie must be replaced with a freshly-signed one" + ); +} + +#[tokio::test] +async fn list_daemons_includes_url() { + let h = spawn(Duration::from_secs(5)).await; + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::ReplyOk).await; + + let registry = h.registry.clone(); + wait_for(Duration::from_secs(2), || { + let r = registry.clone(); + async move { !r.daemons().await.is_empty() } + }) + .await; + + let client = client_with_cookies(); + let resp = client + .get(format!("{}/api/daemons", h.base())) + .send() + .await + .unwrap(); + assert!(resp.status().is_success()); + let body: Vec = resp.json().await.unwrap(); + assert_eq!(body.len(), 1); + assert_eq!(body[0].daemon_id, "laptop"); + assert_eq!(body[0].daemon_url, "ws://10.0.0.1:8081"); +} + +#[tokio::test] +async fn post_sessions_round_trips_to_daemon() { + let h = spawn(Duration::from_secs(5)).await; + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::ReplyOk).await; + + let registry = h.registry.clone(); + wait_for(Duration::from_secs(2), || { + let r = registry.clone(); + async move { !r.daemons().await.is_empty() } + }) + .await; + + let client = client_with_cookies(); + let request = CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + let resp = client + .post(format!("{}/api/sessions", h.base())) + .json(&request) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 201); + let session: Session = resp.json().await.unwrap(); + assert_eq!(session.daemon_id, "laptop"); + assert_eq!(session.command, "bash"); + + // GET /api/sessions should now see it. + let resp = client + .get(format!("{}/api/sessions", h.base())) + .send() + .await + .unwrap(); + let listed: Vec = resp.json().await.unwrap(); + assert!(listed.iter().any(|s| s.id == session.id)); +} + +#[tokio::test] +async fn post_sessions_returns_504_on_daemon_timeout() { + // Short RPC timeout so the 504 path is fast. + let h = spawn(Duration::from_millis(200)).await; + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::Hang).await; + + let registry = h.registry.clone(); + wait_for(Duration::from_secs(2), || { + let r = registry.clone(); + async move { !r.daemons().await.is_empty() } + }) + .await; + + let client = client_with_cookies(); + let request = CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + let resp = client + .post(format!("{}/api/sessions", h.base())) + .json(&request) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 504); + + // The daemon should have received the CreateSession even though it never + // replied. + let captured = mock.captured().await; + assert!( + captured + .iter() + .any(|m| matches!(m, DaemonInbound::CreateSession { .. })), + "daemon should have seen CreateSession, got {captured:?}" + ); +} + +#[tokio::test] +async fn post_sessions_with_worktree_forwards_spec_and_returns_resolved_cwd() { + let h = spawn(Duration::from_secs(5)).await; + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::ReplyOk).await; + + let registry = h.registry.clone(); + wait_for(Duration::from_secs(2), || { + let r = registry.clone(); + async move { !r.daemons().await.is_empty() } + }) + .await; + + let client = client_with_cookies(); + let body = serde_json::json!({ + "daemon_id": "laptop", + "cwd": "/home/naps62/projects/maestro", + "command": "claude", + "worktree": { + "base_repo": "/home/naps62/projects/maestro", + "branch": "feat/test", + "cleanup_on_kill": true + } + }); + let resp = client + .post(format!("{}/api/sessions", h.base())) + .json(&body) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 201); + let session: Session = resp.json().await.unwrap(); + assert_eq!( + session.cwd, + PathBuf::from("/home/naps62/projects/maestro/.worktrees/feat/test") + ); + + let captured = mock.captured().await; + let forwarded = captured + .iter() + .find_map(|m| match m { + DaemonInbound::CreateSession { request, .. } => Some(request.clone()), + _ => None, + }) + .expect("CreateSession was not forwarded"); + let spec = forwarded.worktree.expect("worktree spec dropped"); + assert_eq!(spec.branch, "feat/test"); + assert!(spec.cleanup_on_kill); +} + +#[tokio::test] +async fn delete_session_round_trips() { + let h = spawn(Duration::from_secs(5)).await; + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::ReplyOk).await; + + let registry = h.registry.clone(); + wait_for(Duration::from_secs(2), || { + let r = registry.clone(); + async move { !r.daemons().await.is_empty() } + }) + .await; + + let client = client_with_cookies(); + let request = CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }; + let session: Session = client + .post(format!("{}/api/sessions", h.base())) + .json(&request) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + + // The mock daemon doesn't actually update its catalog after kill — but the + // registry's catalog reflects whatever the daemon broadcasts. We're + // asserting only on the HTTP response. + let resp = client + .delete(format!("{}/api/sessions/{}", h.base(), session.id)) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 204); + + let captured = mock.captured().await; + assert!( + captured + .iter() + .any(|m| matches!(m, DaemonInbound::KillSession { id, .. } if *id == session.id)), + "daemon should have seen KillSession, got {captured:?}" + ); +} + +#[tokio::test] +async fn delete_session_returns_404_for_unknown_session() { + let h = spawn(Duration::from_secs(5)).await; + let client = client_with_cookies(); + let resp = client + .delete(format!("{}/api/sessions/{}", h.base(), Uuid::new_v4())) + .send() + .await + .unwrap(); + assert_eq!(resp.status().as_u16(), 404); +} + +#[tokio::test] +async fn connect_info_issues_valid_jwt() { + let h = spawn(Duration::from_secs(5)).await; + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::ReplyOk).await; + + let registry = h.registry.clone(); + wait_for(Duration::from_secs(2), || { + let r = registry.clone(); + async move { !r.daemons().await.is_empty() } + }) + .await; + + let client = client_with_cookies(); + let session: Session = client + .post(format!("{}/api/sessions", h.base())) + .json(&CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + + let resp = client + .get(format!( + "{}/api/sessions/{}/connect-info", + h.base(), + session.id + )) + .send() + .await + .unwrap(); + assert!(resp.status().is_success()); + let info: ConnectInfo = resp.json().await.unwrap(); + assert_eq!(info.daemon_url, "ws://10.0.0.1:8081"); + + // Verify the JWT against the same shared secret a daemon would use. + let decoded = decode::( + &info.jwt, + &DecodingKey::from_secret(TOKEN.as_bytes()), + &Validation::new(Algorithm::HS256), + ) + .expect("jwt decodes"); + assert_eq!(decoded.claims.daemon_id, "laptop"); + assert_eq!(decoded.claims.session_id, session.id); + assert!(decoded.claims.exp > Utc::now().timestamp()); +} + +#[tokio::test] +async fn events_ws_pushes_snapshot_then_deltas() { + let h = spawn(Duration::from_secs(5)).await; + + // First obtain a valid cookie via a plain HTTP call. + let client = client_with_cookies(); + let _ = client + .get(format!("{}/api/daemons", h.base())) + .send() + .await + .unwrap(); + let cookie_value = harvest_cookie(&client, &h.base()).await; + + // Now open the events WS with that cookie. + let url = format!("{}/api/events", h.ws_base()); + let mut req = url.into_client_request().unwrap(); + req.headers_mut().insert( + reqwest::header::COOKIE, + format!("{}={}", COOKIE_NAME, cookie_value).parse().unwrap(), + ); + let (mut ws, _) = connect_async(req).await.expect("connect events ws"); + + let snapshot = recv_event(&mut ws).await; + match snapshot { + RegistryEvent::Snapshot { daemons, sessions } => { + assert!(daemons.is_empty()); + assert!(sessions.is_empty()); + } + other => panic!("expected Snapshot first, got {other:?}"), + } + + // Now connect a daemon — subscriber should see DaemonAdded. + let mock = MockDaemon::new("laptop", "ws://10.0.0.1:8081"); + mock.clone().spawn(h.addr, DaemonBehavior::ReplyOk).await; + + let event = recv_event(&mut ws).await; + match event { + RegistryEvent::DaemonAdded { daemon } => assert_eq!(daemon.daemon_id, "laptop"), + other => panic!("expected DaemonAdded, got {other:?}"), + } + + // Spawn a session via HTTP; subscriber should see SessionAdded. + let _: Session = client + .post(format!("{}/api/sessions", h.base())) + .json(&CreateSessionRequest { + daemon_id: "laptop".into(), + cwd: PathBuf::from("/tmp"), + command: "bash".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .send() + .await + .unwrap() + .json() + .await + .unwrap(); + + let event = recv_event(&mut ws).await; + assert!( + matches!(event, RegistryEvent::SessionAdded { .. }), + "expected SessionAdded, got {event:?}" + ); +} + +#[tokio::test] +async fn index_serves_placeholder() { + let h = spawn(Duration::from_secs(5)).await; + let client = client_with_cookies(); + let resp = client.get(h.base()).send().await.unwrap(); + assert!(resp.status().is_success()); + let body = resp.text().await.unwrap(); + assert!(body.contains("maestro")); +} + +// --------------------------------------------------------------------------- +// Internal helpers +// --------------------------------------------------------------------------- + +async fn harvest_cookie(client: &Client, base: &str) -> String { + // Reqwest's cookie store keeps the cookie, but doesn't expose values + // directly. Easiest: re-issue a request without the store and read the + // Set-Cookie that comes back. + let bare = reqwest::Client::new(); + let resp = bare + .get(format!("{base}/api/daemons")) + .send() + .await + .unwrap(); + let raw = resp + .headers() + .get(reqwest::header::SET_COOKIE) + .expect("set-cookie") + .to_str() + .unwrap(); + // raw looks like "maestro_session=; HttpOnly; ..." + let prefix = format!("{COOKIE_NAME}="); + let body = raw.strip_prefix(&prefix).expect("cookie prefix"); + let value = body.split(';').next().unwrap(); + let _ = client; // unused — kept in signature for symmetry + value.to_string() +} + +async fn recv_event(ws: &mut WsClient) -> RegistryEvent { + let frame = tokio::time::timeout(Duration::from_secs(2), ws.next()) + .await + .expect("event arrived") + .expect("ws open") + .expect("ws frame"); + match frame { + WsMessage::Text(t) => match serde_json::from_str::(&t) { + Ok(e) => e, + Err(e) => { + let parsed: Value = serde_json::from_str(&t).unwrap(); + panic!("could not parse event {parsed:?}: {e}"); + } + }, + other => panic!("expected text frame, got {other:?}"), + } +}