diff --git a/Cargo.lock b/Cargo.lock index 5af7432..baf943e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -129,6 +129,45 @@ dependencies = [ "zbus", ] +[[package]] +name = "asn1-rs" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5493c3bedbacf7fd7382c6346bbd66687d12bbaad3a89a2d2c303ee6cf20b048" +dependencies = [ + "asn1-rs-derive", + "asn1-rs-impl", + "displaydoc", + "nom 7.1.3", + "num-traits", + "rusticata-macros", + "thiserror 1.0.69", + "time", +] + +[[package]] +name = "asn1-rs-derive" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "965c2d33e53cb6b267e148a4cb0760bc01f4904c1cd4bb4002a085bb016d1490" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "synstructure", +] + +[[package]] +name = "asn1-rs-impl" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b18050c2cd6fe86c3a76584ef5e0baf286d038cda203eb6223df2cc413565f7" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "async-broadcast" version = "0.7.2" @@ -402,7 +441,7 @@ checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" dependencies = [ "cfg-if", "cpufeatures 0.3.0", - "rand_core", + "rand_core 0.10.1", ] [[package]] @@ -488,6 +527,35 @@ dependencies = [ "crossbeam-utils", ] +[[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" @@ -549,6 +617,20 @@ version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be1e0bca6c3637f992fc1cc7cbc52a78c1ef6db076dbf1059c4323d6a2048376" +[[package]] +name = "der-parser" +version = "9.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5cd0a5c643689626bec213c4d8bd4d96acc8ffdb4ad4bb6bc16abf27d5f4b553" +dependencies = [ + "asn1-rs", + "displaydoc", + "nom 7.1.3", + "num-bigint", + "num-traits", + "rusticata-macros", +] + [[package]] name = "deranged" version = "0.5.8" @@ -563,6 +645,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" dependencies = [ "block-buffer", "crypto-common", + "subtle", ] [[package]] @@ -586,6 +669,15 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "document-features" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4b8a88685455ed29a21542a33abd9cb6510b6b129abadabdcef0f4c55bc8f61" +dependencies = [ + "litrs", +] + [[package]] name = "downcast-rs" version = "1.2.1" @@ -946,7 +1038,7 @@ dependencies = [ "js-sys", "libc", "r-efi 6.0.0", - "rand_core", + "rand_core 0.10.1", "wasm-bindgen", ] @@ -1039,7 +1131,7 @@ dependencies = [ "pastey", "pin-project-lite", "smallvec", - "thiserror", + "thiserror 2.0.18", ] [[package]] @@ -1084,6 +1176,24 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" +[[package]] +name = "hkdf" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" +dependencies = [ + "hmac", +] + +[[package]] +name = "hmac" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" +dependencies = [ + "digest", +] + [[package]] name = "http" version = "1.4.2" @@ -1548,6 +1658,12 @@ 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 = "lock_api" version = "0.4.14" @@ -1631,6 +1747,12 @@ dependencies = [ "unicase", ] +[[package]] +name = "minimal-lexical" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" + [[package]] name = "miniz_oxide" version = "0.8.9" @@ -1658,6 +1780,16 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "956787520e75e9bd233246045d19f42fb73242759cc57fba9611d940ae96d4b0" +[[package]] +name = "nom" +version = "7.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a" +dependencies = [ + "memchr", + "minimal-lexical", +] + [[package]] name = "nom" version = "8.0.0" @@ -1824,6 +1956,15 @@ dependencies = [ "objc2-core-foundation", ] +[[package]] +name = "oid-registry" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8d8034d9489cdaf79228eb9f6a3b8d7bb32ba00d6645ebd48eef4077ceb5bd9" +dependencies = [ + "asn1-rs", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -1906,6 +2047,16 @@ version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4" +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64", + "serde_core", +] + [[package]] name = "percent-encoding" version = "2.3.2" @@ -1992,6 +2143,15 @@ version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391" +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + [[package]] name = "proc-macro-crate" version = "3.5.0" @@ -2010,6 +2170,22 @@ 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 = "quick-xml" version = "0.39.4" @@ -2033,7 +2209,7 @@ dependencies = [ "rustc-hash", "rustls", "socket2", - "thiserror", + "thiserror 2.0.18", "tokio", "tracing", "web-time", @@ -2048,14 +2224,14 @@ dependencies = [ "bytes", "getrandom 0.4.3", "lru-slab", - "rand", + "rand 0.10.2", "rand_pcg", "ring", "rustc-hash", "rustls", "rustls-pki-types", "slab", - "thiserror", + "thiserror 2.0.18", "tinyvec", "tracing", "web-time", @@ -2096,6 +2272,16 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rand" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9ef1d0d795eb7d84685bca4f72f3649f064e6641543d3a8c415898726a57b41" +dependencies = [ + "rand_chacha", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.2" @@ -2104,7 +2290,26 @@ checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ "chacha20", "getrandom 0.4.3", - "rand_core", + "rand_core 0.10.1", +] + +[[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]] +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]] @@ -2119,7 +2324,21 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" dependencies = [ - "rand_core", + "rand_core 0.10.1", +] + +[[package]] +name = "rcgen" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75e669e5202259b5314d1ea5397316ad400819437857b90861765f24c4cf80a2" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", ] [[package]] @@ -2205,6 +2424,8 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64", "bytes", + "cookie", + "cookie_store", "futures-core", "futures-util", "http", @@ -2300,14 +2521,14 @@ dependencies = [ "itertools 0.14.0", "log", "memchr", - "nom", + "nom 8.0.0", "num", "num-derive", "num-traits", "sha1", "sha2", "sha3", - "thiserror", + "thiserror 2.0.18", "zeroize", "zstd", ] @@ -2339,6 +2560,15 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" +[[package]] +name = "rusticata-macros" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf0c4a6ece9950b9abdb62b1cfcf2a68b3b67a10ba445b3bb85be2a293d0632" +dependencies = [ + "nom 7.1.3", +] + [[package]] name = "rustix" version = "1.1.4" @@ -2358,6 +2588,7 @@ version = "0.23.42" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" dependencies = [ + "log", "once_cell", "ring", "rustls-pki-types", @@ -2608,17 +2839,22 @@ dependencies = [ "notify-rust", "open", "proc-macro2", + "rcgen", "reqwest", "resvg", "roxmltree", "rustix", + "rustls", "sd-notify", "serde", "serde_json", "sha2", + "spl-core", + "spl-transport", "syn 2.0.119", "tempfile", "tokio", + "tokio-rustls", "tokio-util", "toml", "tracing", @@ -2629,6 +2865,40 @@ dependencies = [ "zbus", ] +[[package]] +name = "spl-core" +version = "0.1.0" +source = "git+https://github.com/solpbc/spl-rust?rev=742bc9dc789c5a75658844849a04d75033aeb6e3#742bc9dc789c5a75658844849a04d75033aeb6e3" +dependencies = [ + "base64", + "hkdf", + "serde", + "serde_json", + "sha2", + "thiserror 2.0.18", +] + +[[package]] +name = "spl-transport" +version = "0.1.0" +source = "git+https://github.com/solpbc/spl-rust?rev=742bc9dc789c5a75658844849a04d75033aeb6e3#742bc9dc789c5a75658844849a04d75033aeb6e3" +dependencies = [ + "base64", + "futures-util", + "rcgen", + "rustls", + "rustls-webpki", + "serde", + "serde_json", + "spl-core", + "thiserror 2.0.18", + "tokio", + "tokio-rustls", + "tokio-tungstenite", + "tracing", + "webpki-roots", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -2761,7 +3031,7 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ed071c670382e85fc2f48ae706492d8c338f4f89bf72520d32f8abfe880aade" dependencies = [ - "thiserror", + "thiserror 2.0.18", "windows", "windows-version", ] @@ -2779,13 +3049,33 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "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", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", ] [[package]] @@ -2819,6 +3109,7 @@ dependencies = [ "powerfmt", "serde_core", "time-core", + "time-macros", ] [[package]] @@ -2827,6 +3118,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" +[[package]] +name = "time-macros" +version = "0.2.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f" +dependencies = [ + "num-conv", + "time-core", +] + [[package]] name = "tiny-skia" version = "0.12.0" @@ -2916,6 +3217,21 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -3089,6 +3405,25 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.5", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror 2.0.18", + "utf-8", +] + [[package]] name = "typenum" version = "1.20.1" @@ -3164,6 +3499,12 @@ dependencies = [ "xmlwriter", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -3699,6 +4040,24 @@ version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ea6fc2961e4ef194dcbfe56bb845534d0dc8098940c7e5c012a258bfec6701bd" +[[package]] +name = "x509-parser" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcbc162f30700d6f3f82a24bf7cc62ffe7caea42c0b2cba8bf7f3ae50cf51f69" +dependencies = [ + "asn1-rs", + "data-encoding", + "der-parser", + "lazy_static", + "nom 7.1.3", + "oid-registry", + "ring", + "rusticata-macros", + "thiserror 1.0.69", + "time", +] + [[package]] name = "xattr" version = "1.6.1" @@ -3724,6 +4083,15 @@ dependencies = [ "lzma-sys", ] +[[package]] +name = "yasna" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd" +dependencies = [ + "time", +] + [[package]] name = "yoke" version = "0.8.3" diff --git a/crates/solstone-linux/Cargo.toml b/crates/solstone-linux/Cargo.toml index c337736..b6240c6 100644 --- a/crates/solstone-linux/Cargo.toml +++ b/crates/solstone-linux/Cargo.toml @@ -23,7 +23,9 @@ flac-bound = { version = "0.5.0", default-features = false, features = ["libflac libpulse-binding = "2.30.1" chrono = "0.4" rustix = { version = "1", features = ["event", "fs", "process", "time"] } -reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls", "stream"] } +reqwest = { version = "0.12", default-features = false, features = ["cookies", "json", "multipart", "rustls-tls", "stream"] } +spl-core = { version = "0.1.0", git = "https://github.com/solpbc/spl-rust", rev = "742bc9dc789c5a75658844849a04d75033aeb6e3" } +spl-transport = { version = "0.1.0", git = "https://github.com/solpbc/spl-rust", rev = "742bc9dc789c5a75658844849a04d75033aeb6e3" } tokio = { workspace = true, features = ["signal", "sync"] } tokio-util = { version = "0.7", features = ["io", "rt"] } ashpd = { version = "=0.13.12", features = ["screencast"] } @@ -48,7 +50,10 @@ tempfile = "3" http-body-util = "0.1" hyper = { version = "1", features = ["http1", "server"] } hyper-util = { version = "0.1", features = ["tokio"] } -tokio = { workspace = true, features = ["net", "test-util"] } +tokio = { workspace = true, features = ["io-util", "net", "rt-multi-thread", "sync", "test-util"] } +rcgen = { version = "0.13", default-features = false, features = ["ring", "pem", "x509-parser"] } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] } +tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12"] } roxmltree = "0.21" toml = "1" diff --git a/crates/solstone-linux/src/cli.rs b/crates/solstone-linux/src/cli.rs index 3b86563..d7fc667 100644 --- a/crates/solstone-linux/src/cli.rs +++ b/crates/solstone-linux/src/cli.rs @@ -629,8 +629,11 @@ fn cmd_run(interval: Option) -> i32 { #[cfg(test)] mod tests { use super::*; - use crate::config::save_identity; + use crate::{ + config::save_identity, private_link_test_peer::PrivateLinkPeer, test_support::MockServer, + }; use clap::CommandFactory; + use serde_json::json; use std::{cell::Cell, path::Path}; struct FakeRunner { @@ -1108,6 +1111,54 @@ mod tests { assert!(!out.contains("Config saved")); } + #[test] + fn setup_registration_remains_on_production_http_not_private_link() { + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap(); + let server = runtime.block_on(MockServer::new(vec![( + 200, + json!({"key":"K123456789", "name":"host-a"}), + )])); + let peer = runtime.block_on(PrivateLinkPeer::start()); + let temp = tempfile::tempdir().unwrap(); + let options = SetupOptions { + server_url: Some(server.url.clone()), + token: None, + stream_name: Some("host-a".into()), + non_interactive: true, + }; + let mut output = Vec::new(); + let mut errors = Vec::new(); + assert_eq!( + cmd_setup( + options, + paths(&temp), + None, + &mut RealRegistrar, + &mut output, + &mut errors, + ), + 0 + ); + let requests = server.requests(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].method, "POST"); + assert_eq!(requests[0].uri, "/app/observer/register"); + assert!(requests[0].headers.get("authorization").is_none()); + let body: serde_json::Value = serde_json::from_slice(&requests[0].body).unwrap(); + assert!( + body["hostname"] + .as_str() + .is_some_and(|value| !value.is_empty()) + ); + assert_eq!(body["label"], "host-a"); + assert_eq!(body["stream_type"], "desktop"); + assert!(peer.requests().is_empty()); + runtime.block_on(peer.shutdown()); + } + struct ScriptedPrompt { inputs: std::collections::VecDeque>, output: String, diff --git a/crates/solstone-linux/src/lib.rs b/crates/solstone-linux/src/lib.rs index c961285..39a2716 100644 --- a/crates/solstone-linux/src/lib.rs +++ b/crates/solstone-linux/src/lib.rs @@ -18,6 +18,10 @@ pub mod matching; pub mod observer; pub mod pipeline; pub mod positions; +#[allow(dead_code)] +mod private_file; +#[allow(dead_code)] +mod private_link; pub mod recovery; pub mod restore_token; pub mod rotation; @@ -39,6 +43,9 @@ pub mod video; #[cfg(test)] mod observer_contract_tests; #[cfg(test)] +#[allow(dead_code)] +mod private_link_test_peer; +#[cfg(test)] mod release_rail_tests; #[cfg(test)] mod test_support; diff --git a/crates/solstone-linux/src/private_file.rs b/crates/solstone-linux/src/private_file.rs new file mode 100644 index 0000000..926be3b --- /dev/null +++ b/crates/solstone-linux/src/private_file.rs @@ -0,0 +1,348 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +use std::{ + fmt, + fs::{self, File}, + io::{self, Write}, + os::unix::fs::PermissionsExt, + path::{Path, PathBuf}, + sync::atomic::{AtomicU64, Ordering}, +}; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum DurableWriteStage { + Create, + Write, + Fsync, + Rename, + DirSync, +} + +pub(crate) trait DurableWriteFault: Send + Sync { + fn before(&self, stage: DurableWriteStage) -> io::Result<()>; +} + +pub(crate) struct NoWriteFault; + +impl DurableWriteFault for NoWriteFault { + fn before(&self, _stage: DurableWriteStage) -> io::Result<()> { + Ok(()) + } +} + +pub(crate) enum PrivateFileError { + InvalidTarget(&'static str), + Io { + target: &'static str, + operation: &'static str, + kind: io::ErrorKind, + }, +} + +impl PrivateFileError { + fn io(target: &'static str, operation: &'static str, error: io::Error) -> Self { + Self::Io { + target, + operation, + kind: error.kind(), + } + } +} + +impl fmt::Debug for PrivateFileError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, formatter) + } +} + +impl fmt::Display for PrivateFileError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidTarget(target) => write!(formatter, "InvalidTarget({target})"), + Self::Io { + target, + operation, + kind, + } => write!(formatter, "Io({target}, {operation}, {kind:?})"), + } + } +} + +impl std::error::Error for PrivateFileError {} + +pub(crate) fn ensure_private_directory(path: &Path) -> Result<(), PrivateFileError> { + if path.as_os_str().is_empty() || path.parent().is_none() { + return Err(PrivateFileError::InvalidTarget("directory")); + } + let mut missing = Vec::new(); + let mut current = path; + loop { + match fs::symlink_metadata(current) { + Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_dir() => { + return Err(PrivateFileError::InvalidTarget("directory")); + } + Ok(_) => break, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + missing.push(current.to_path_buf()); + current = current + .parent() + .filter(|parent| !parent.as_os_str().is_empty()) + .ok_or(PrivateFileError::InvalidTarget("directory"))?; + } + Err(error) => return Err(PrivateFileError::io("directory", "inspect", error)), + } + } + for directory in missing.iter().rev() { + fs::create_dir(directory) + .map_err(|error| PrivateFileError::io("directory", "create", error))?; + set_and_verify_mode(directory, 0o700, true)?; + } + set_and_verify_mode(path, 0o700, true) +} + +fn set_and_verify_mode(path: &Path, mode: u32, directory: bool) -> Result<(), PrivateFileError> { + fs::set_permissions(path, fs::Permissions::from_mode(mode)) + .map_err(|error| PrivateFileError::io("target", "chmod", error))?; + let metadata = fs::symlink_metadata(path) + .map_err(|error| PrivateFileError::io("target", "inspect", error))?; + let valid_kind = if directory { + metadata.is_dir() + } else { + metadata.is_file() + }; + if metadata.file_type().is_symlink() + || !valid_kind + || metadata.permissions().mode() & 0o777 != mode + { + return Err(PrivateFileError::InvalidTarget("target")); + } + Ok(()) +} + +pub(crate) fn open_regular_readonly(path: &Path) -> Result { + let descriptor = rustix::fs::openat( + rustix::fs::CWD, + path, + rustix::fs::OFlags::RDONLY | rustix::fs::OFlags::CLOEXEC | rustix::fs::OFlags::NOFOLLOW, + rustix::fs::Mode::empty(), + ) + .map_err(|error| { + if error == rustix::io::Errno::LOOP { + PrivateFileError::InvalidTarget("file") + } else { + PrivateFileError::io("file", "open", error.into()) + } + })?; + let file = File::from(descriptor); + if !file + .metadata() + .map_err(|error| PrivateFileError::io("file", "inspect", error))? + .is_file() + { + return Err(PrivateFileError::InvalidTarget("file")); + } + Ok(file) +} + +static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0); + +pub(crate) fn atomic_write_bytes(path: &Path, bytes: &[u8]) -> Result<(), PrivateFileError> { + atomic_write_bytes_with_fault(path, bytes, &NoWriteFault) +} + +pub(crate) fn atomic_write_bytes_with_fault( + path: &Path, + bytes: &[u8], + fault: &dyn DurableWriteFault, +) -> Result<(), PrivateFileError> { + let previous = match fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => { + return Err(PrivateFileError::InvalidTarget("file")); + } + Ok(_) => Some(fs::read(path).map_err(|error| PrivateFileError::io("file", "read", error))?), + Err(error) if error.kind() == io::ErrorKind::NotFound => None, + Err(error) => return Err(PrivateFileError::io("file", "inspect", error)), + }; + let parent = path + .parent() + .ok_or(PrivateFileError::InvalidTarget("file"))?; + let name = path + .file_name() + .ok_or(PrivateFileError::InvalidTarget("file"))? + .to_string_lossy(); + let temporary = parent.join(format!( + ".{name}.{}.{}.tmp", + std::process::id(), + TEMP_COUNTER.fetch_add(1, Ordering::Relaxed) + )); + let result = write_temporary(path, &temporary, parent, bytes, fault); + if result.is_err() { + let _ = fs::remove_file(&temporary); + if let Some(previous) = previous { + let _ = fs::write(path, previous); + } + } + result +} + +fn write_temporary( + path: &Path, + temporary: &PathBuf, + parent: &Path, + bytes: &[u8], + fault: &dyn DurableWriteFault, +) -> Result<(), PrivateFileError> { + fault + .before(DurableWriteStage::Create) + .map_err(|error| PrivateFileError::io("file", "create", error))?; + let descriptor = rustix::fs::open( + temporary, + rustix::fs::OFlags::CREATE + | rustix::fs::OFlags::EXCL + | rustix::fs::OFlags::WRONLY + | rustix::fs::OFlags::CLOEXEC + | rustix::fs::OFlags::NOFOLLOW, + rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR, + ) + .map_err(|error| PrivateFileError::io("file", "create", error.into()))?; + let mut file = File::from(descriptor); + file.set_permissions(fs::Permissions::from_mode(0o600)) + .map_err(|error| PrivateFileError::io("file", "chmod", error))?; + fault + .before(DurableWriteStage::Write) + .map_err(|error| PrivateFileError::io("file", "write", error))?; + file.write_all(bytes) + .and_then(|()| file.flush()) + .map_err(|error| PrivateFileError::io("file", "write", error))?; + fault + .before(DurableWriteStage::Fsync) + .map_err(|error| PrivateFileError::io("file", "fsync", error))?; + file.sync_all() + .map_err(|error| PrivateFileError::io("file", "fsync", error))?; + fault + .before(DurableWriteStage::Rename) + .map_err(|error| PrivateFileError::io("file", "rename", error))?; + fs::rename(temporary, path).map_err(|error| PrivateFileError::io("file", "rename", error))?; + fault + .before(DurableWriteStage::DirSync) + .map_err(|error| PrivateFileError::io("directory", "fsync", error))?; + File::open(parent) + .and_then(|directory| directory.sync_all()) + .map_err(|error| PrivateFileError::io("directory", "fsync", error)) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::fs::{MetadataExt, symlink}; + + struct FailAt(DurableWriteStage); + impl DurableWriteFault for FailAt { + fn before(&self, stage: DurableWriteStage) -> io::Result<()> { + if stage == self.0 { + Err(io::Error::other("injected")) + } else { + Ok(()) + } + } + } + + #[test] + fn ensure_private_directory_creates_private_tree() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("a/b"); + ensure_private_directory(&path).unwrap(); + assert_eq!(fs::metadata(&path).unwrap().mode() & 0o777, 0o700); + assert_eq!( + fs::metadata(path.parent().unwrap()).unwrap().mode() & 0o777, + 0o700 + ); + } + + #[test] + fn ensure_private_directory_rejects_wrong_kinds_without_following() { + for leaf in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let referent = temp.path().join("referent"); + fs::create_dir(&referent).unwrap(); + fs::set_permissions(&referent, fs::Permissions::from_mode(0o755)).unwrap(); + let link = temp.path().join("link"); + symlink(&referent, &link).unwrap(); + let target = if leaf { link } else { link.join("child") }; + assert!(ensure_private_directory(&target).is_err()); + assert_eq!(fs::metadata(&referent).unwrap().mode() & 0o777, 0o755); + } + let temp = tempfile::tempdir().unwrap(); + let file = temp.path().join("file"); + fs::write(&file, b"x").unwrap(); + assert!(ensure_private_directory(&file.join("child")).is_err()); + } + + #[test] + fn open_regular_readonly_accepts_only_regular_files() { + let temp = tempfile::tempdir().unwrap(); + let file = temp.path().join("file"); + fs::write(&file, b"x").unwrap(); + assert!(open_regular_readonly(&file).is_ok()); + assert!(open_regular_readonly(temp.path()).is_err()); + let link = temp.path().join("link"); + symlink(&file, &link).unwrap(); + assert!(open_regular_readonly(&link).is_err()); + } + + #[test] + fn atomic_write_is_exact_private_and_regular() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("state"); + atomic_write_bytes(&path, b"complete").unwrap(); + let metadata = fs::symlink_metadata(&path).unwrap(); + assert!(metadata.is_file()); + assert_eq!(metadata.mode() & 0o777, 0o600); + assert_eq!(fs::read(path).unwrap(), b"complete"); + } + + #[test] + fn every_injected_stage_preserves_previous_complete_file() { + for stage in [ + DurableWriteStage::Create, + DurableWriteStage::Write, + DurableWriteStage::Fsync, + DurableWriteStage::Rename, + DurableWriteStage::DirSync, + ] { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("state"); + atomic_write_bytes(&path, b"previous").unwrap(); + assert!(atomic_write_bytes_with_fault(&path, b"partial", &FailAt(stage)).is_err()); + assert_eq!(fs::read(&path).unwrap(), b"previous", "{stage:?}"); + assert!(fs::read_dir(temp.path()).unwrap().all(|entry| { + !entry + .unwrap() + .file_name() + .to_string_lossy() + .ends_with(".tmp") + })); + } + } + + #[test] + fn failed_initial_write_never_leaves_partial_target_or_temporary() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("state"); + assert!( + atomic_write_bytes_with_fault(&path, b"partial", &FailAt(DurableWriteStage::Write)) + .is_err() + ); + assert!(!path.exists()); + assert_eq!(fs::read_dir(temp.path()).unwrap().count(), 0); + } + + #[test] + fn error_format_never_contains_paths() { + let secret = "/secret/owner/path"; + let error = open_regular_readonly(Path::new(secret)).unwrap_err(); + assert!(!format!("{error}").contains(secret)); + assert!(!format!("{error:?}").contains(secret)); + } +} diff --git a/crates/solstone-linux/src/private_link.rs b/crates/solstone-linux/src/private_link.rs new file mode 100644 index 0000000..b3faf89 --- /dev/null +++ b/crates/solstone-linux/src/private_link.rs @@ -0,0 +1,1740 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +use std::{ + fmt, + fs::{self, File}, + future::Future, + io::{self, Read}, + os::unix::fs::PermissionsExt, + path::{Path, PathBuf}, + pin::Pin, + sync::{Arc, Mutex, RwLock}, + time::Duration, +}; + +use reqwest::{Method, RequestBuilder, StatusCode, Url}; +use serde::{Deserialize, Serialize}; +use spl_core::bridge::{BridgeNames, RequestHeaderPolicy}; +use spl_transport::credential::Credential; +use spl_transport::{ + TransportError, + client::{DialedCarrier, TokenPersistHook, TransportClient}, + journal_bridge::{ + BridgePolicy, CapabilityGate, CarrierOpener, JournalBridgeConfig, JournalBridgeHandle, + }, +}; + +use crate::private_file::{ + DurableWriteFault, PrivateFileError, atomic_write_bytes, atomic_write_bytes_with_fault, + ensure_private_directory, open_regular_readonly, +}; + +pub(crate) const CREDENTIALS_FILENAME: &str = "credentials.json"; +pub(crate) const OBSERVER_FILENAME: &str = "observer.json"; +const PRIVATE_STATE_LOCK_FILENAME: &str = ".solstone-linux.private-state.lock"; +const MAX_PAIR_LINK_BYTES: u64 = 4096; +const MAX_REQUEST_BODY_BYTES: usize = 16 * 1024 * 1024; +const LOOPBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); +const LOOPBACK_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); +pub(crate) const OBSERVER_HEADER_NAME: &str = "x-solstone-observer"; +pub(crate) const PROTOCOL_VERSION_HEADER_NAME: &str = "x-solstone-protocol-version"; + +#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct ObserverState { + pub(crate) credential_instance_id: String, + pub(crate) key: String, + pub(crate) prefix: String, + pub(crate) name: String, + pub(crate) ingest_url: String, + pub(crate) protocol_version: u64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum PrivateTargetKind { + ConfigDirectory, + Credential, + Observer, + Lock, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum PrivateIoOperation { + EnsureDirectory, + Canonicalize, + Open, + Inspect, + Chmod, + Lock, + Read, + Serialize, + Persist, +} + +pub(crate) enum PrivateStateError { + MalformedCredential, + MalformedObserver, + InvalidTarget { + kind: PrivateTargetKind, + }, + Io { + operation: PrivateIoOperation, + source: io::Error, + }, + LockContended, + PairInputInvalid, + PairingFailed, + BridgeUnavailable, + BootstrapFailed, + RegistrationInvalid, + TokenPersistenceFailed, + ShutdownFailed, +} + +impl fmt::Display for PrivateStateError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::MalformedCredential => formatter.write_str("MalformedCredential"), + Self::MalformedObserver => formatter.write_str("MalformedObserver"), + Self::InvalidTarget { kind } => write!(formatter, "InvalidTarget({kind:?})"), + Self::Io { operation, source } => { + write!(formatter, "Io({operation:?}, {:?})", source.kind()) + } + Self::LockContended => formatter.write_str("LockContended"), + Self::PairInputInvalid => formatter.write_str("PairInputInvalid"), + Self::PairingFailed => formatter.write_str("PairingFailed"), + Self::BridgeUnavailable => formatter.write_str("BridgeUnavailable"), + Self::BootstrapFailed => formatter.write_str("BootstrapFailed"), + Self::RegistrationInvalid => formatter.write_str("RegistrationInvalid"), + Self::TokenPersistenceFailed => formatter.write_str("TokenPersistenceFailed"), + Self::ShutdownFailed => formatter.write_str("ShutdownFailed"), + } + } +} + +impl fmt::Debug for PrivateStateError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + fmt::Display::fmt(self, formatter) + } +} + +impl std::error::Error for PrivateStateError {} + +fn map_private_file( + error: PrivateFileError, + kind: PrivateTargetKind, + operation: PrivateIoOperation, +) -> PrivateStateError { + match error { + PrivateFileError::InvalidTarget(_) => PrivateStateError::InvalidTarget { kind }, + PrivateFileError::Io { kind, .. } => PrivateStateError::Io { + operation, + source: io::Error::from(kind), + }, + } +} + +pub(crate) fn confine_path(origin: &Url, path: &str) -> Result { + if !path.starts_with('/') + || path.starts_with("//") + || path + .bytes() + .any(|byte| matches!(byte, b'?' | b'#' | b'\\' | b'\r' | b'\n' | 0)) + { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + }); + } + for segment in path.split('/') { + // Reject encoded percent so double encoding has one bounded failure rule. + if segment.to_ascii_lowercase().contains("%25") { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + }); + } + let decoded = percent_decode(segment)?; + if decoded == b"." + || decoded == b".." + || decoded + .iter() + .any(|byte| matches!(byte, b'/' | b'\\' | b'\r' | b'\n' | 0)) + { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + }); + } + } + let url = origin + .join(path) + .map_err(|_| PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + })?; + if url.scheme() != origin.scheme() + || url.host_str() != origin.host_str() + || url.port_or_known_default() != origin.port_or_known_default() + || url.query().is_some() + || url.fragment().is_some() + { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + }); + } + Ok(url) +} + +fn percent_decode(value: &str) -> Result, PrivateStateError> { + let bytes = value.as_bytes(); + let mut output = Vec::with_capacity(bytes.len()); + let mut index = 0; + while index < bytes.len() { + if bytes[index] != b'%' { + output.push(bytes[index]); + index += 1; + continue; + } + if index + 2 >= bytes.len() { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + }); + } + let high = hex(bytes[index + 1])?; + let low = hex(bytes[index + 2])?; + output.push(high << 4 | low); + index += 3; + } + Ok(output) +} + +fn hex(byte: u8) -> Result { + match byte { + b'0'..=b'9' => Ok(byte - b'0'), + b'a'..=b'f' => Ok(byte - b'a' + 10), + b'A'..=b'F' => Ok(byte - b'A' + 10), + _ => Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Observer, + }), + } +} + +pub(crate) struct PrivateStateLock { + _file: File, + canonical_root: PathBuf, +} + +impl Drop for PrivateStateLock { + fn drop(&mut self) { + let _ = rustix::fs::flock(&self._file, rustix::fs::FlockOperation::Unlock); + } +} + +impl PrivateStateLock { + pub(crate) fn acquire(config_root: &Path) -> Result { + ensure_private_directory(config_root).map_err(|error| { + map_private_file( + error, + PrivateTargetKind::ConfigDirectory, + PrivateIoOperation::EnsureDirectory, + ) + })?; + let canonical_root = + fs::canonicalize(config_root).map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Canonicalize, + source, + })?; + let descriptor = rustix::fs::openat( + rustix::fs::CWD, + canonical_root.join(PRIVATE_STATE_LOCK_FILENAME), + rustix::fs::OFlags::RDWR + | rustix::fs::OFlags::CLOEXEC + | rustix::fs::OFlags::NOFOLLOW + | rustix::fs::OFlags::CREATE, + rustix::fs::Mode::RUSR | rustix::fs::Mode::WUSR, + ) + .map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Open, + source: source.into(), + })?; + let file = File::from(descriptor); + if !file + .metadata() + .map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Inspect, + source, + })? + .is_file() + { + return Err(PrivateStateError::InvalidTarget { + kind: PrivateTargetKind::Lock, + }); + } + file.set_permissions(fs::Permissions::from_mode(0o600)) + .map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Chmod, + source, + })?; + match rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive) { + Ok(()) => {} + Err(rustix::io::Errno::WOULDBLOCK) => return Err(PrivateStateError::LockContended), + Err(source) => { + return Err(PrivateStateError::Io { + operation: PrivateIoOperation::Lock, + source: source.into(), + }); + } + } + Ok(Self { + _file: file, + canonical_root, + }) + } + + pub(crate) fn root(&self) -> &Path { + &self.canonical_root + } +} + +pub(crate) fn read_pair_link(input: R) -> Result { + let mut bytes = Vec::new(); + input + .take(MAX_PAIR_LINK_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Read, + source, + })?; + if bytes.len() as u64 > MAX_PAIR_LINK_BYTES { + return Err(PrivateStateError::PairInputInvalid); + } + let text = std::str::from_utf8(&bytes).map_err(|_| PrivateStateError::PairInputInvalid)?; + let link = text.trim_end_matches(char::is_whitespace); + if link.is_empty() || link.chars().any(char::is_whitespace) { + return Err(PrivateStateError::PairInputInvalid); + } + Ok(link.to_owned()) +} + +pub(crate) trait Pairer: Send + Sync { + fn pair<'a>( + &'a self, + link: &'a str, + device_label: &'a str, + additional_fields: &'a serde_json::Map, + ) -> Pin> + Send + 'a>>; +} + +pub(crate) struct SplPairer; + +impl Pairer for SplPairer { + fn pair<'a>( + &'a self, + link: &'a str, + device_label: &'a str, + additional_fields: &'a serde_json::Map, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + spl_transport::pairing::pair_from_link(link, device_label, additional_fields) + .await + .map_err(|_| PrivateStateError::PairingFailed) + }) + } +} + +pub(crate) async fn setup( + config_root: &Path, + device_label: &str, + input: R, +) -> Result<(), PrivateStateError> { + setup_with_pairer(&SplPairer, config_root, device_label, input).await +} + +pub(crate) async fn setup_with_pairer( + pairer: &dyn Pairer, + config_root: &Path, + device_label: &str, + input: R, +) -> Result<(), PrivateStateError> { + let state_lock = PrivateStateLock::acquire(config_root)?; + let link = read_pair_link(input)?; + let credential = pairer + .pair(&link, device_label, &serde_json::Map::new()) + .await?; + persist_credential(state_lock.root(), &credential) +} + +fn read_private_file( + path: &Path, + kind: PrivateTargetKind, +) -> Result>, PrivateStateError> { + let mut file = match open_regular_readonly(path) { + Ok(file) => file, + Err(PrivateFileError::Io { + kind: io::ErrorKind::NotFound, + .. + }) => return Ok(None), + Err(error) => { + return Err(map_private_file(error, kind, PrivateIoOperation::Open)); + } + }; + let mut bytes = Vec::new(); + file.read_to_end(&mut bytes) + .map_err(|source| PrivateStateError::Io { + operation: PrivateIoOperation::Read, + source, + })?; + Ok(Some(bytes)) +} + +pub(crate) fn load_credential(config_root: &Path) -> Result, PrivateStateError> { + let Some(bytes) = read_private_file( + &config_root.join(CREDENTIALS_FILENAME), + PrivateTargetKind::Credential, + )? + else { + return Ok(None); + }; + serde_json::from_slice(&bytes) + .map(Some) + .map_err(|_| PrivateStateError::MalformedCredential) +} + +pub(crate) fn persist_credential( + config_root: &Path, + credential: &Credential, +) -> Result<(), PrivateStateError> { + let bytes = + serde_json::to_vec(credential).map_err(|_| PrivateStateError::MalformedCredential)?; + atomic_write_bytes(&config_root.join(CREDENTIALS_FILENAME), &bytes).map_err(|error| { + map_private_file( + error, + PrivateTargetKind::Credential, + PrivateIoOperation::Persist, + ) + }) +} + +pub(crate) fn load_observer( + config_root: &Path, + credential_instance_id: &str, + expected_name: &str, + origin: &Url, +) -> Result, PrivateStateError> { + let Some(bytes) = read_private_file( + &config_root.join(OBSERVER_FILENAME), + PrivateTargetKind::Observer, + )? + else { + return Ok(None); + }; + let observer = serde_json::from_slice::(&bytes) + .map_err(|_| PrivateStateError::MalformedObserver)?; + if observer.credential_instance_id != credential_instance_id + || observer.name != expected_name + || observer.protocol_version != 2 + || observer.key.is_empty() + || observer.prefix.is_empty() + || observer.name.is_empty() + || observer.ingest_url.is_empty() + || contains_invalid_header_value(&observer.key) + || confine_path(origin, &observer.ingest_url).is_err() + { + return Ok(None); + } + Ok(Some(observer)) +} + +pub(crate) fn persist_observer( + config_root: &Path, + observer: &ObserverState, +) -> Result<(), PrivateStateError> { + let bytes = serde_json::to_vec(observer).map_err(|_| PrivateStateError::MalformedObserver)?; + atomic_write_bytes(&config_root.join(OBSERVER_FILENAME), &bytes).map_err(|error| { + map_private_file( + error, + PrivateTargetKind::Observer, + PrivateIoOperation::Persist, + ) + }) +} + +fn contains_invalid_header_value(value: &str) -> bool { + value.bytes().any(|byte| matches!(byte, b'\r' | b'\n' | 0)) +} + +fn persist_and_publish_observer( + config_root: &Path, + observer: &ObserverState, + opener: &PrivateLinkOpener, + fault: &dyn DurableWriteFault, +) -> Result<(), PrivateStateError> { + let bytes = serde_json::to_vec(observer).map_err(|_| PrivateStateError::MalformedObserver)?; + atomic_write_bytes_with_fault(&config_root.join(OBSERVER_FILENAME), &bytes, fault).map_err( + |error| { + map_private_file( + error, + PrivateTargetKind::Observer, + PrivateIoOperation::Persist, + ) + }, + )?; + opener.set_registered(observer) +} + +#[derive(Clone)] +enum OpenerAuth { + Unregistered, + Registered { key: String }, +} + +struct PrivateLinkOpener { + transport: Arc, + auth: RwLock, + expected_name: String, +} + +impl PrivateLinkOpener { + fn new(transport: TransportClient, expected_name: String) -> Self { + Self { + transport: Arc::new(transport), + auth: RwLock::new(OpenerAuth::Unregistered), + expected_name, + } + } + + fn set_registered(&self, observer: &ObserverState) -> Result<(), PrivateStateError> { + if observer.key.is_empty() + || contains_invalid_header_value(&observer.key) + || observer.protocol_version != 2 + || observer.name != self.expected_name + { + return Err(PrivateStateError::RegistrationInvalid); + } + let mut auth = self + .auth + .write() + .map_err(|_| PrivateStateError::RegistrationInvalid)?; + *auth = OpenerAuth::Registered { + key: observer.key.clone(), + }; + Ok(()) + } +} + +impl CarrierOpener for PrivateLinkOpener { + fn proxy_headers( + &self, + upstream_headers: &[(String, String)], + ) -> Result, TransportError> { + let auth = self + .auth + .read() + .map_err(|_| TransportError::Pairing("opener state unavailable".into()))?; + Ok(proxy_headers_for_auth(upstream_headers, &auth)) + } + + fn dial_carrier( + &self, + ) -> Pin> + Send + '_>> { + Box::pin(self.transport.dial_carrier()) + } +} + +fn proxy_headers_for_auth( + upstream_headers: &[(String, String)], + auth: &OpenerAuth, +) -> Vec<(String, String)> { + let mut headers = upstream_headers.to_vec(); + match auth { + OpenerAuth::Unregistered => { + headers.push((PROTOCOL_VERSION_HEADER_NAME.to_owned(), "2".to_owned())); + } + OpenerAuth::Registered { key } => { + headers.push((OBSERVER_HEADER_NAME.to_owned(), key.clone())); + headers.push(("authorization".to_owned(), format!("Bearer {key}"))); + headers.push((PROTOCOL_VERSION_HEADER_NAME.to_owned(), "2".to_owned())); + } + } + headers +} + +pub(crate) struct PrivateLinkSession { + client: reqwest::Client, + origin: Url, + opener: Arc, + handle: JournalBridgeHandle, + token_persistence: Option>, +} + +impl PrivateLinkSession { + pub(crate) fn request( + &self, + method: Method, + relative_path: &str, + ) -> Result { + let url = confine_path(&self.origin, relative_path)?; + Ok(self + .client + .request(method, url) + .timeout(LOOPBACK_REQUEST_TIMEOUT)) + } + + pub(crate) async fn shutdown(self) -> Result<(), PrivateStateError> { + let status = self.handle.shutdown_and_wait().await; + if self + .token_persistence + .as_ref() + .is_some_and(|state| state.failed()) + { + return Err(PrivateStateError::TokenPersistenceFailed); + } + if status.listener_active || status.active_requests != 0 { + return Err(PrivateStateError::ShutdownFailed); + } + Ok(()) + } +} + +struct TokenPersistence { + config_root: PathBuf, + credential: Mutex, + failed: Mutex, + fault: Arc, +} + +impl TokenPersistence { + fn new( + config_root: PathBuf, + credential: Credential, + fault: Arc, + ) -> (Arc, TokenPersistHook) { + let state = Arc::new(Self { + config_root, + credential: Mutex::new(credential), + failed: Mutex::new(false), + fault, + }); + let hook_state = state.clone(); + let hook: TokenPersistHook = Arc::new(move |token, expires_at| { + // SPL makes the token live before this synchronous hook; a concurrent + // request can observe it before durability, an upstream race we cannot close here. + hook_state.persist(token, expires_at); + }); + (state, hook) + } + + fn persist(&self, token: &str, expires_at: i64) { + let mut current = self.credential.lock().unwrap_or_else(|p| p.into_inner()); + let mut updated = current.clone(); + updated.device_token = Some(token.to_owned()); + updated.device_token_expires_at = Some(expires_at); + let durable = serde_json::to_vec(&updated).ok().is_some_and(|bytes| { + atomic_write_bytes_with_fault( + &self.config_root.join(CREDENTIALS_FILENAME), + &bytes, + self.fault.as_ref(), + ) + .is_ok() + }); + if durable { + *current = updated; + } else { + *self.failed.lock().unwrap_or_else(|p| p.into_inner()) = true; + } + } + + fn failed(&self) -> bool { + *self.failed.lock().unwrap_or_else(|p| p.into_inner()) + } +} + +pub(crate) async fn start_private_link_session( + credential: Credential, + expected_name: &str, +) -> Result { + start_private_link_session_inner(credential, expected_name, None, None).await +} + +async fn start_private_link_session_inner( + credential: Credential, + expected_name: &str, + persistence: Option<(PathBuf, Arc)>, + capability_capture: Option<&Mutex>>, +) -> Result { + let endpoint_hosts = credential + .endpoints + .iter() + .map(|endpoint| endpoint.host.clone()) + .collect(); + let (token_persistence, hook) = persistence.map_or((None, None), |(root, fault)| { + let (state, hook) = TokenPersistence::new(root, credential.clone(), fault); + (Some(state), Some(hook)) + }); + let transport = + TransportClient::new(credential, hook).map_err(|_| PrivateStateError::BridgeUnavailable)?; + let opener = Arc::new(PrivateLinkOpener::new(transport, expected_name.to_owned())); + let bridge_names = BridgeNames { + capability_cookie_name: "solstone_linux_cap".to_owned(), + upstream_cookie_prefix: "solstone_linux_".to_owned(), + observer_header_name: OBSERVER_HEADER_NAME.to_owned(), + protocol_version_header_name: PROTOCOL_VERSION_HEADER_NAME.to_owned(), + }; + let policy = BridgePolicy { + port: 0, + capability_gate: CapabilityGate::Enabled, + stream_response: BridgePolicy::default().stream_response, + local_response: Arc::new(|_, _| None), + attribution_headers: Arc::new(|_| Vec::new()), + request_headers: RequestHeaderPolicy::Allow( + [ + "accept", + "accept-language", + "content-type", + "cache-control", + "if-none-match", + "if-modified-since", + "range", + "user-agent", + ] + .into_iter() + .map(str::to_owned) + .collect(), + ), + max_request_body_bytes: MAX_REQUEST_BODY_BYTES, + }; + let handle = spl_transport::journal_bridge::start(JournalBridgeConfig { + opener: opener.clone(), + bridge_names, + endpoint_hosts, + policy, + }) + .await + .map_err(|_| PrivateStateError::BridgeUnavailable)?; + let bootstrap_url = handle + .bootstrap_url() + .ok_or(PrivateStateError::BootstrapFailed)?; + if let Some(capture) = capability_capture { + let capability = Url::parse(&bootstrap_url) + .ok() + .and_then(|url| { + url.query_pairs() + .find(|(name, _)| name == "cap") + .map(|(_, value)| value.into_owned()) + }) + .ok_or(PrivateStateError::BootstrapFailed)?; + *capture.lock().unwrap_or_else(|p| p.into_inner()) = Some(capability); + } + let origin = Url::parse(&format!("http://127.0.0.1:{}", handle.port())) + .map_err(|_| PrivateStateError::BridgeUnavailable)?; + let client = reqwest::Client::builder() + .cookie_store(true) + .redirect(reqwest::redirect::Policy::none()) + .no_proxy() + .connect_timeout(LOOPBACK_CONNECT_TIMEOUT) + .build() + .map_err(|_| PrivateStateError::BridgeUnavailable)?; + let response = client + .get(bootstrap_url) + .timeout(LOOPBACK_REQUEST_TIMEOUT) + .send() + .await + .map_err(|_| PrivateStateError::BootstrapFailed)?; + if response.status() != StatusCode::FOUND { + handle.begin_shutdown(); + return Err(PrivateStateError::BootstrapFailed); + } + Ok(PrivateLinkSession { + client, + origin, + opener, + handle, + token_persistence, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::private_file::DurableWriteStage; + use crate::private_link_test_peer::PrivateLinkPeer; + use spl_transport::credential::EndpointAddr; + use std::{ + io::Cursor, + net::TcpListener, + os::unix::fs::{MetadataExt, symlink}, + process::Command, + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + }; + + fn credential() -> Credential { + Credential { + client_key_pem: "client-key".into(), + client_cert_pem: "client-cert".into(), + ca_chain_pem: vec!["ca".into()], + ca_fp_prefix: vec![1, 2, 3], + instance_id: "instance".into(), + home_label: "home".into(), + endpoints: vec![EndpointAddr { + host: "127.0.0.1".into(), + port: 7657, + }], + home_attestation: Some("attestation".into()), + local_endpoints: Some(serde_json::json!([{"ip":"127.0.0.1","port":7657}])), + relay_origin: Some("https://relay.invalid".into()), + device_token: Some("device-token".into()), + device_token_expires_at: Some(123), + } + } + + fn observer(path: &str) -> ObserverState { + ObserverState { + credential_instance_id: "instance".into(), + key: "observer-key".into(), + prefix: "prefix".into(), + name: "stream".into(), + ingest_url: path.into(), + protocol_version: 2, + } + } + + fn assert_load_rejection_keeps_opener_unregistered(state: ObserverState) { + let temp = tempfile::tempdir().unwrap(); + persist_observer(temp.path(), &state).unwrap(); + let loaded = load_observer( + temp.path(), + "instance", + "stream", + &Url::parse("http://127.0.0.1:1").unwrap(), + ) + .unwrap(); + assert!(loaded.is_none()); + let headers = proxy_headers_for_auth(&[], &OpenerAuth::Unregistered); + assert_eq!( + headers, + vec![(PROTOCOL_VERSION_HEADER_NAME.to_owned(), "2".to_owned())] + ); + } + + macro_rules! opener_rejection_test { + ($name:ident, $state:expr) => { + #[test] + fn $name() { + assert_load_rejection_keeps_opener_unregistered($state); + } + }; + } + + opener_rejection_test!( + opener_stays_unregistered_for_credential_mismatch, + ObserverState { + credential_instance_id: "other".into(), + ..observer("/ingest") + } + ); + opener_rejection_test!( + opener_stays_unregistered_for_name_mismatch, + ObserverState { + name: "other".into(), + ..observer("/ingest") + } + ); + opener_rejection_test!( + opener_stays_unregistered_for_protocol_mismatch, + ObserverState { + protocol_version: 3, + ..observer("/ingest") + } + ); + opener_rejection_test!( + opener_stays_unregistered_for_unsafe_key, + ObserverState { + key: "bad\rkey".into(), + ..observer("/ingest") + } + ); + opener_rejection_test!( + opener_stays_unregistered_for_relative_path, + observer("relative") + ); + opener_rejection_test!( + opener_stays_unregistered_for_scheme_relative_path, + observer("//host/x") + ); + opener_rejection_test!( + opener_stays_unregistered_for_raw_traversal, + observer("/a/../b") + ); + opener_rejection_test!( + opener_stays_unregistered_for_encoded_traversal, + observer("/a/%2e%2e/b") + ); + opener_rejection_test!( + opener_stays_unregistered_for_mixed_encoded_traversal, + observer("/a/%2E./b") + ); + opener_rejection_test!( + opener_stays_unregistered_for_encoded_slash, + observer("/a/%2f/b") + ); + opener_rejection_test!( + opener_stays_unregistered_for_encoded_backslash, + observer("/a/%5c/b") + ); + opener_rejection_test!( + opener_stays_unregistered_for_double_encoding, + observer("/a/%252e%252e/b") + ); + opener_rejection_test!(opener_stays_unregistered_for_query, observer("/a?q")); + opener_rejection_test!(opener_stays_unregistered_for_fragment, observer("/a#f")); + opener_rejection_test!(opener_stays_unregistered_for_backslash, observer("/a\\b")); + + fn assert_redacted_rejection(bytes: &[u8]) { + let temp = tempfile::tempdir().unwrap(); + let error = read_pair_link(Cursor::new(bytes)).unwrap_err(); + let material = String::from_utf8_lossy(bytes); + if !material.is_empty() { + assert!(!format!("{error}").contains(material.as_ref())); + assert!(!format!("{error:?}").contains(material.as_ref())); + } + assert!(!temp.path().join(CREDENTIALS_FILENAME).exists()); + assert!(!temp.path().join(OBSERVER_FILENAME).exists()); + } + + #[test] + fn pair_input_empty_is_rejected() { + assert_redacted_rejection(b""); + } + #[test] + fn pair_input_invalid_utf8_is_rejected() { + assert_redacted_rejection(b"\xff"); + } + #[test] + fn pair_input_embedded_whitespace_is_rejected() { + assert_redacted_rejection(b"pair link"); + } + #[test] + fn pair_input_leading_whitespace_is_rejected() { + assert_redacted_rejection(b" pair"); + } + #[test] + fn pair_input_trailing_spaces_and_tabs_are_accepted_after_trim() { + assert_eq!(read_pair_link(Cursor::new(b"pair \t")).unwrap(), "pair"); + } + #[test] + fn pair_input_multiple_line_endings_are_rejected() { + assert_redacted_rejection(b"pair\nother\n"); + } + #[test] + fn pair_input_without_terminator_is_accepted() { + assert_eq!(read_pair_link(Cursor::new(b"pair")).unwrap(), "pair"); + } + #[test] + fn pair_input_lf_termination_is_accepted() { + assert_eq!(read_pair_link(Cursor::new(b"pair\n")).unwrap(), "pair"); + } + #[test] + fn pair_input_crlf_termination_is_accepted() { + assert_eq!(read_pair_link(Cursor::new(b"pair\r\n")).unwrap(), "pair"); + } + #[test] + fn pair_input_exactly_4096_bytes_is_accepted() { + assert_eq!( + read_pair_link(Cursor::new(vec![b'a'; 4096])).unwrap().len(), + 4096 + ); + } + #[test] + fn pair_input_4097_bytes_is_rejected() { + assert_redacted_rejection(&vec![b'a'; 4097]); + } + + struct FakePairer { + calls: Arc, + result: Option, + } + + struct FailStage(DurableWriteStage); + impl DurableWriteFault for FailStage { + fn before(&self, stage: DurableWriteStage) -> io::Result<()> { + if stage == self.0 { + Err(io::Error::other("injected")) + } else { + Ok(()) + } + } + } + + struct RecordingFault { + stages: Arc>>, + fail: Option, + } + impl DurableWriteFault for RecordingFault { + fn before(&self, stage: DurableWriteStage) -> io::Result<()> { + self.stages.lock().unwrap().push(stage); + if self.fail == Some(stage) { + Err(io::Error::other("injected")) + } else { + Ok(()) + } + } + } + impl Pairer for FakePairer { + fn pair<'a>( + &'a self, + _link: &'a str, + _device_label: &'a str, + _additional_fields: &'a serde_json::Map, + ) -> Pin> + Send + 'a>> + { + self.calls.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { self.result.clone().ok_or(PrivateStateError::PairingFailed) }) + } + } + + #[tokio::test] + async fn injected_pairer_persists_only_success() { + let temp = tempfile::tempdir().unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let pairer = FakePairer { + calls: calls.clone(), + result: None, + }; + let failed = setup_with_pairer(&pairer, temp.path(), "device", Cursor::new(b"pair")).await; + assert!(failed.is_err()); + drop(failed); + assert!(!temp.path().join(CREDENTIALS_FILENAME).exists()); + let pairer = FakePairer { + calls: calls.clone(), + result: Some(credential()), + }; + setup_with_pairer(&pairer, temp.path(), "device", Cursor::new(b"pair")) + .await + .unwrap(); + assert_eq!(load_credential(temp.path()).unwrap(), Some(credential())); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn spl_pairer_surfaces_typed_failure() { + let error = SplPairer + .pair("not-a-pair-link", "device", &serde_json::Map::new()) + .await + .unwrap_err(); + assert!(matches!(error, PrivateStateError::PairingFailed)); + } + + #[test] + fn credential_and_observer_round_trip_all_fields() { + let temp = tempfile::tempdir().unwrap(); + ensure_private_directory(temp.path()).unwrap(); + persist_credential(temp.path(), &credential()).unwrap(); + persist_observer(temp.path(), &observer("/app/observer/ingest")).unwrap(); + assert_eq!(load_credential(temp.path()).unwrap(), Some(credential())); + let origin = Url::parse("http://127.0.0.1:1234").unwrap(); + assert!( + load_observer(temp.path(), "instance", "stream", &origin) + .unwrap() + .is_some() + ); + for name in [CREDENTIALS_FILENAME, OBSERVER_FILENAME] { + let metadata = fs::symlink_metadata(temp.path().join(name)).unwrap(); + assert!(metadata.is_file()); + assert_eq!(metadata.permissions().mode() & 0o777, 0o600); + } + assert_eq!( + fs::metadata(temp.path()).unwrap().permissions().mode() & 0o777, + 0o700 + ); + } + + #[test] + fn malformed_state_errors_are_distinct() { + let temp = tempfile::tempdir().unwrap(); + fs::write(temp.path().join(CREDENTIALS_FILENAME), b"{").unwrap(); + assert!(matches!( + load_credential(temp.path()), + Err(PrivateStateError::MalformedCredential) + )); + fs::write(temp.path().join(OBSERVER_FILENAME), b"{").unwrap(); + let origin = Url::parse("http://127.0.0.1:1").unwrap(); + assert!(matches!( + load_observer(temp.path(), "instance", "stream", &origin), + Err(PrivateStateError::MalformedObserver) + )); + } + + #[test] + fn observer_semantic_rejections_are_read_only() { + let origin = Url::parse("http://127.0.0.1:1234").unwrap(); + let cases = [ + ObserverState { + credential_instance_id: "other".into(), + ..observer("/ingest") + }, + ObserverState { + name: "other".into(), + ..observer("/ingest") + }, + ObserverState { + protocol_version: 3, + ..observer("/ingest") + }, + ObserverState { + key: "bad\rkey".into(), + ..observer("/ingest") + }, + ObserverState { + key: String::new(), + ..observer("/ingest") + }, + ObserverState { + prefix: String::new(), + ..observer("/ingest") + }, + ]; + for state in cases { + let temp = tempfile::tempdir().unwrap(); + let bytes = serde_json::to_vec(&state).unwrap(); + fs::write(temp.path().join(OBSERVER_FILENAME), &bytes).unwrap(); + assert!( + load_observer(temp.path(), "instance", "stream", &origin) + .unwrap() + .is_none() + ); + assert_eq!( + fs::read(temp.path().join(OBSERVER_FILENAME)).unwrap(), + bytes + ); + } + } + + #[test] + fn observer_ingest_path_rejections_are_read_only() { + let origin = Url::parse("http://127.0.0.1:1234").unwrap(); + for path in [ + "relative", + "//other/path", + "/a/../b", + "/%2e%2e/b", + "/%252e%252e/b", + "/a?query", + "/a#fragment", + "/a\\b", + "/a/%2f/b", + "/a/%5c/b", + "/a/%zz", + ] { + let temp = tempfile::tempdir().unwrap(); + let bytes = serde_json::to_vec(&observer(path)).unwrap(); + fs::write(temp.path().join(OBSERVER_FILENAME), &bytes).unwrap(); + assert!( + load_observer(temp.path(), "instance", "stream", &origin) + .unwrap() + .is_none(), + "{path}" + ); + assert_eq!( + fs::read(temp.path().join(OBSERVER_FILENAME)).unwrap(), + bytes + ); + } + } + + #[test] + fn private_state_files_reject_symlinks_without_touching_referent() { + let temp = tempfile::tempdir().unwrap(); + let referent = temp.path().join("referent"); + fs::write(&referent, b"external").unwrap(); + fs::set_permissions(&referent, fs::Permissions::from_mode(0o644)).unwrap(); + for name in [CREDENTIALS_FILENAME, OBSERVER_FILENAME] { + let link = temp.path().join(name); + symlink(&referent, &link).unwrap(); + let result = if name == CREDENTIALS_FILENAME { + load_credential(temp.path()).map(|_| ()) + } else { + load_observer( + temp.path(), + "instance", + "stream", + &Url::parse("http://127.0.0.1:1").unwrap(), + ) + .map(|_| ()) + }; + assert!(matches!( + result, + Err(PrivateStateError::InvalidTarget { .. }) + )); + fs::remove_file(link).unwrap(); + } + assert_eq!(fs::read(&referent).unwrap(), b"external"); + assert_eq!( + fs::metadata(referent).unwrap().permissions().mode() & 0o777, + 0o644 + ); + } + + #[test] + fn lock_release_preserves_inode_and_nonsymlink_aliases_contend() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("root"); + let first = PrivateStateLock::acquire(&root).unwrap(); + let lock_path = root.join(PRIVATE_STATE_LOCK_FILENAME); + let before = fs::metadata(&lock_path).unwrap(); + let alias = root.join("."); + let contender = PrivateStateLock::acquire(&alias); + assert!(matches!(contender, Err(PrivateStateError::LockContended))); + drop(contender); + drop(first); + assert!(lock_path.exists()); + let second = PrivateStateLock::acquire(&alias).unwrap(); + let after = fs::metadata(lock_path).unwrap(); + assert_eq!((before.dev(), before.ino()), (after.dev(), after.ino())); + drop(second); + } + + #[test] + fn lock_rejects_symlinked_config_root_without_touching_referent() { + let temp = tempfile::tempdir().unwrap(); + let referent = temp.path().join("referent"); + fs::create_dir(&referent).unwrap(); + fs::set_permissions(&referent, fs::Permissions::from_mode(0o755)).unwrap(); + let alias = temp.path().join("alias"); + symlink(&referent, &alias).unwrap(); + assert!(matches!( + PrivateStateLock::acquire(&alias), + Err(PrivateStateError::InvalidTarget { .. }) + )); + assert_eq!( + fs::metadata(&referent).unwrap().permissions().mode() & 0o777, + 0o755 + ); + assert!(!referent.join(PRIVATE_STATE_LOCK_FILENAME).exists()); + } + + struct CountingReader(Arc); + impl Read for CountingReader { + fn read(&mut self, _buffer: &mut [u8]) -> io::Result { + self.0.fetch_add(1, Ordering::SeqCst); + Ok(0) + } + } + + #[tokio::test] + async fn lock_loser_fails_before_consuming_pair_input_or_state() { + let temp = tempfile::tempdir().unwrap(); + let held = PrivateStateLock::acquire(temp.path()).unwrap(); + let reads = Arc::new(AtomicUsize::new(0)); + assert!(matches!( + setup_with_pairer( + &FakePairer { + calls: Arc::new(AtomicUsize::new(0)), + result: Some(credential()), + }, + temp.path(), + "device", + CountingReader(reads.clone()), + ) + .await, + Err(PrivateStateError::LockContended) + )); + assert_eq!(reads.load(Ordering::SeqCst), 0); + assert!(!temp.path().join(CREDENTIALS_FILENAME).exists()); + assert!(!temp.path().join(OBSERVER_FILENAME).exists()); + drop(held); + } + + #[tokio::test] + async fn bridge_reuses_one_carrier_across_registration_transition() { + let peer = PrivateLinkPeer::start().await; + peer.enqueue_response(200, b"{}".to_vec()); + peer.enqueue_response(200, b"{}".to_vec()); + let session = start_private_link_session(peer.credential(), "stream") + .await + .unwrap(); + assert_eq!( + session + .request(Method::GET, "/unregistered") + .unwrap() + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); + session + .opener + .set_registered(&observer("/app/observer/ingest")) + .unwrap(); + assert_eq!( + session + .request(Method::GET, "/registered") + .unwrap() + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); + let requests = peer.requests(); + assert_eq!(requests.len(), 2); + assert_eq!( + requests[0] + .headers + .iter() + .find(|(name, _)| name.eq_ignore_ascii_case(PROTOCOL_VERSION_HEADER_NAME)) + .map(|(_, value)| value.as_str()), + Some("2") + ); + assert!( + !requests[0] + .headers + .iter() + .any(|(name, _)| name.eq_ignore_ascii_case(OBSERVER_HEADER_NAME) + || name.eq_ignore_ascii_case("authorization")) + ); + for name in [ + OBSERVER_HEADER_NAME, + PROTOCOL_VERSION_HEADER_NAME, + "authorization", + ] { + assert!( + requests[1] + .headers + .iter() + .any(|(candidate, _)| candidate.eq_ignore_ascii_case(name)) + ); + } + assert_eq!(peer.accepted_carriers(), 1); + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[test] + fn confined_requests_reject_every_unsafe_target_locally() { + let origin = Url::parse("http://127.0.0.1:1234").unwrap(); + for target in [ + "http://other/x", + "//other/x", + "/a/../b", + "/a/%2e%2e/b", + "/a/%2E./b", + "/a/%2f/b", + "/a/%2F/b", + "/a/%5c/b", + "/a/%5C/b", + "/a\\b", + "/a/%252e%252e/b", + "/a?q", + "/a#f", + ] { + assert!(confine_path(&origin, target).is_err(), "{target}"); + } + } + + #[tokio::test] + async fn bridge_rejects_untrusted_local_authority_and_auth_without_upstream() { + let peer = PrivateLinkPeer::start().await; + let session = start_private_link_session(peer.credential(), "stream") + .await + .unwrap(); + for (name, value) in [ + (OBSERVER_HEADER_NAME, "forged"), + (PROTOCOL_VERSION_HEADER_NAME, "2"), + ("authorization", "Bearer forged"), + ] { + let response = session + .request(Method::GET, "/blocked") + .unwrap() + .header(name, value) + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + let bare = reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(); + for request in [ + bare.get(session.origin.join("/missing").unwrap()), + bare.get(session.origin.join("/wrong-cookie").unwrap()) + .header("cookie", "solstone_linux_cap=wrong"), + bare.get(session.origin.join("/wrong-host").unwrap()) + .header("host", "example.invalid"), + ] { + assert_eq!( + request.send().await.unwrap().status(), + StatusCode::FORBIDDEN + ); + } + assert!(peer.requests().is_empty()); + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[tokio::test] + async fn loopback_client_does_not_follow_upstream_redirects() { + let peer = PrivateLinkPeer::start().await; + peer.enqueue_response(302, Vec::new()); + let session = start_private_link_session(peer.credential(), "stream") + .await + .unwrap(); + let response = session + .request(Method::GET, "/redirect") + .unwrap() + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::FOUND); + assert_eq!(peer.requests().len(), 1); + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[tokio::test] + async fn paired_peer_resumes_large_response_after_window_credit() { + let peer = PrivateLinkPeer::start().await; + let body = vec![b'x'; spl_core::mux::INITIAL_WINDOW + 131_072]; + peer.enqueue_response(200, body.clone()); + let session = start_private_link_session(peer.credential(), "stream") + .await + .unwrap(); + let received = session + .request(Method::GET, "/large") + .unwrap() + .send() + .await + .unwrap() + .bytes() + .await + .unwrap(); + assert_eq!(received.as_ref(), body); + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[test] + fn loopback_client_ignores_proxy_environment() { + let trap = TcpListener::bind("127.0.0.1:0").unwrap(); + trap.set_nonblocking(true).unwrap(); + let proxy = format!("http://{}", trap.local_addr().unwrap()); + let child = Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "private_link::tests::loopback_client_ignores_proxy_environment_child", + "--ignored", + "--nocapture", + ]) + .env("SOLSTONE_PROXY_TEST_CHILD", "1") + .env("HTTP_PROXY", &proxy) + .env("HTTPS_PROXY", &proxy) + .env("ALL_PROXY", &proxy) + .env("http_proxy", &proxy) + .env("https_proxy", &proxy) + .env("all_proxy", &proxy) + .status() + .unwrap(); + assert!(child.success()); + assert!(matches!(trap.accept(), Err(error) if error.kind() == io::ErrorKind::WouldBlock)); + } + + #[tokio::test] + #[ignore = "executed in a child with isolated proxy environment"] + async fn loopback_client_ignores_proxy_environment_child() { + assert_eq!( + std::env::var("SOLSTONE_PROXY_TEST_CHILD").as_deref(), + Ok("1") + ); + let peer = PrivateLinkPeer::start().await; + peer.enqueue_response(200, b"{}".to_vec()); + let session = start_private_link_session(peer.credential(), "stream") + .await + .unwrap(); + assert_eq!( + session + .request(Method::GET, "/proxy-proof") + .unwrap() + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[tokio::test] + async fn observer_publication_is_durable_before_registered_auth() { + let temp = tempfile::tempdir().unwrap(); + let prior = observer("/prior"); + persist_observer(temp.path(), &prior).unwrap(); + let prior_bytes = fs::read(temp.path().join(OBSERVER_FILENAME)).unwrap(); + let peer = PrivateLinkPeer::start().await; + for _ in 0..6 { + peer.enqueue_response(200, b"{}".to_vec()); + } + let session = start_private_link_session(peer.credential(), "stream") + .await + .unwrap(); + let next = ObserverState { + key: "new-observer-key".into(), + ingest_url: "/new".into(), + ..observer("/new") + }; + for stage in [ + DurableWriteStage::Create, + DurableWriteStage::Write, + DurableWriteStage::Fsync, + DurableWriteStage::Rename, + DurableWriteStage::DirSync, + ] { + assert!( + persist_and_publish_observer( + temp.path(), + &next, + &session.opener, + &FailStage(stage) + ) + .is_err() + ); + assert_eq!( + fs::read(temp.path().join(OBSERVER_FILENAME)).unwrap(), + prior_bytes + ); + let loaded = load_observer( + temp.path(), + "instance", + "stream", + &Url::parse("http://127.0.0.1:1").unwrap(), + ) + .unwrap(); + assert!(loaded.as_ref() == Some(&prior)); + session + .request(Method::GET, "/still-unregistered") + .unwrap() + .send() + .await + .unwrap(); + } + persist_and_publish_observer( + temp.path(), + &next, + &session.opener, + &crate::private_file::NoWriteFault, + ) + .unwrap(); + session + .request(Method::GET, "/registered-after-durable") + .unwrap() + .send() + .await + .unwrap(); + let requests = peer.requests(); + assert_eq!(requests.len(), 6); + for request in &requests[..5] { + assert!(request.headers.iter().any(|(name, value)| { + name.eq_ignore_ascii_case(PROTOCOL_VERSION_HEADER_NAME) && value == "2" + })); + assert!(!request.headers.iter().any(|(name, _)| { + name.eq_ignore_ascii_case(OBSERVER_HEADER_NAME) + || name.eq_ignore_ascii_case("authorization") + })); + } + for name in [ + OBSERVER_HEADER_NAME, + PROTOCOL_VERSION_HEADER_NAME, + "authorization", + ] { + assert!( + requests[5] + .headers + .iter() + .any(|(candidate, _)| candidate.eq_ignore_ascii_case(name)) + ); + } + session.shutdown().await.unwrap(); + peer.shutdown().await; + } + + #[test] + fn token_hook_returns_only_after_directory_sync_and_reload_sees_refresh() { + let temp = tempfile::tempdir().unwrap(); + persist_credential(temp.path(), &credential()).unwrap(); + let stages = Arc::new(Mutex::new(Vec::new())); + let (_state, hook) = TokenPersistence::new( + temp.path().to_path_buf(), + credential(), + Arc::new(RecordingFault { + stages: stages.clone(), + fail: None, + }), + ); + hook("refreshed-token", 456); + assert_eq!( + *stages.lock().unwrap(), + [ + DurableWriteStage::Create, + DurableWriteStage::Write, + DurableWriteStage::Fsync, + DurableWriteStage::Rename, + DurableWriteStage::DirSync, + ] + ); + let loaded = load_credential(temp.path()).unwrap().unwrap(); + assert_eq!(loaded.device_token.as_deref(), Some("refreshed-token")); + assert_eq!(loaded.device_token_expires_at, Some(456)); + } + + async fn assert_token_failure(stage: DurableWriteStage) { + let temp = tempfile::tempdir().unwrap(); + let prior = credential(); + persist_credential(temp.path(), &prior).unwrap(); + let prior_bytes = fs::read(temp.path().join(CREDENTIALS_FILENAME)).unwrap(); + let peer = PrivateLinkPeer::start().await; + let session = start_private_link_session_inner( + peer.credential(), + "stream", + Some(( + temp.path().to_path_buf(), + Arc::new(RecordingFault { + stages: Arc::new(Mutex::new(Vec::new())), + fail: Some(stage), + }), + )), + None, + ) + .await + .unwrap(); + session + .token_persistence + .as_ref() + .unwrap() + .persist("failed-refresh", 999); + assert_eq!( + fs::read(temp.path().join(CREDENTIALS_FILENAME)).unwrap(), + prior_bytes + ); + assert!(matches!( + session.shutdown().await, + Err(PrivateStateError::TokenPersistenceFailed) + )); + peer.shutdown().await; + } + + #[tokio::test] + async fn token_write_failure_is_latched_and_preserves_prior_credential() { + assert_token_failure(DurableWriteStage::Write).await; + } + + #[tokio::test] + async fn token_fsync_failure_is_latched_and_preserves_prior_credential() { + assert_token_failure(DurableWriteStage::Fsync).await; + } + + #[tokio::test] + async fn executed_session_surfaces_do_not_disclose_secrets() { + let peer = PrivateLinkPeer::start().await; + let mut paired = peer.credential(); + let client_key = paired.client_key_pem.clone(); + let key_interior = client_key + .lines() + .find(|line| !line.starts_with('-')) + .unwrap() + .to_owned(); + let device_token = "device-token-sentinel"; + paired.device_token = Some(device_token.into()); + let capability = Mutex::new(None); + let session = start_private_link_session_inner(paired, "stream", None, Some(&capability)) + .await + .unwrap(); + let capability = capability.into_inner().unwrap().unwrap(); + let observer_key = "observer-key-sentinel"; + let registered = ObserverState { + key: observer_key.into(), + ..observer("/ingest") + }; + session.opener.set_registered(®istered).unwrap(); + let request_debug = format!( + "{:?}", + session + .request(Method::GET, "/safe") + .unwrap() + .build() + .unwrap() + ); + let error = session.request(Method::GET, "/%252e%252e").unwrap_err(); + let outputs = [ + request_debug, + format!("{error}"), + format!("{error:?}"), + format!("{:?}", session.shutdown().await), + ]; + for output in outputs { + for secret in [ + capability.as_str(), + device_token, + client_key.as_str(), + key_interior.as_str(), + observer_key, + ] { + assert!(!output.contains(secret)); + } + } + peer.shutdown().await; + } + + #[test] + fn session_implements_no_debug_clone_or_serialize_traits() { + use core::marker::PhantomData; + + struct DebugProbe(PhantomData); + trait DebugFallback { + fn probe(&self) -> bool { + false + } + } + impl DebugFallback for DebugProbe {} + impl DebugProbe { + fn probe(&self) -> bool { + true + } + } + + struct CloneProbe(PhantomData); + trait CloneFallback { + fn probe(&self) -> bool { + false + } + } + impl CloneFallback for CloneProbe {} + impl CloneProbe { + fn probe(&self) -> bool { + true + } + } + + struct SerializeProbe(PhantomData); + trait SerializeFallback { + fn probe(&self) -> bool { + false + } + } + impl SerializeFallback for SerializeProbe {} + impl SerializeProbe { + fn probe(&self) -> bool { + true + } + } + + assert!(DebugProbe::(PhantomData).probe()); + assert!(CloneProbe::(PhantomData).probe()); + assert!(SerializeProbe::(PhantomData).probe()); + assert!(!DebugProbe::(PhantomData).probe()); + assert!(!CloneProbe::(PhantomData).probe()); + assert!(!SerializeProbe::(PhantomData).probe()); + } +} diff --git a/crates/solstone-linux/src/private_link_test_peer.rs b/crates/solstone-linux/src/private_link_test_peer.rs new file mode 100644 index 0000000..30370ea --- /dev/null +++ b/crates/solstone-linux/src/private_link_test_peer.rs @@ -0,0 +1,323 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// Copyright (c) 2026 sol pbc + +use std::{ + collections::{HashMap, VecDeque}, + io, + sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, + }, +}; + +use rcgen::{ + BasicConstraints, CertificateParams, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose, + PKCS_ECDSA_P256_SHA256, +}; +use rustls::{ + RootCertStore, ServerConfig, + pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}, + server::WebPkiClientVerifier, +}; +use spl_core::{ + frame::{ + FLAG_CLOSE, FLAG_DATA, FLAG_OPEN, FLAG_RESET, FLAG_WINDOW, Frame, FrameDecoder, + RECOMMENDED_CHUNK, + }, + mux::INITIAL_WINDOW, +}; +use spl_transport::credential::{Credential, EndpointAddr}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::{TcpListener, TcpStream}, + task::JoinHandle, +}; +use tokio_rustls::{TlsAcceptor, server::TlsStream}; + +#[derive(Clone)] +pub(crate) struct PeerRequest { + pub(crate) method: String, + pub(crate) path: String, + pub(crate) headers: Vec<(String, String)>, + pub(crate) body: Vec, +} + +#[derive(Clone)] +struct PeerResponse { + status: u16, + body: Vec, +} + +struct OutboundResponse { + bytes: Vec, + offset: usize, + credit: usize, +} + +#[derive(Clone)] +struct PeerState { + responses: Arc>>, + requests: Arc>>, + accepted: Arc, + uploaded: Arc, +} + +pub(crate) struct PrivateLinkPeer { + credential: Credential, + state: PeerState, + task: JoinHandle<()>, +} + +impl PrivateLinkPeer { + pub(crate) async fn start() -> Self { + let listener = TcpListener::bind(("127.0.0.1", 0)).await.unwrap(); + let (credential, acceptor) = credential_and_acceptor(listener.local_addr().unwrap().port()); + let state = PeerState { + responses: Arc::new(Mutex::new(VecDeque::new())), + requests: Arc::new(Mutex::new(Vec::new())), + accepted: Arc::new(AtomicUsize::new(0)), + uploaded: Arc::new(AtomicUsize::new(0)), + }; + let task_state = state.clone(); + let task = tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + task_state.accepted.fetch_add(1, Ordering::SeqCst); + let Ok(tls) = acceptor.accept(stream).await else { + continue; + }; + let _ = serve_carrier(tls, &task_state).await; + } + }); + Self { + credential, + state, + task, + } + } + + pub(crate) fn credential(&self) -> Credential { + self.credential.clone() + } + pub(crate) fn enqueue_response(&self, status: u16, body: impl Into>) { + self.state + .responses + .lock() + .unwrap() + .push_back(PeerResponse { + status, + body: body.into(), + }); + } + pub(crate) fn requests(&self) -> Vec { + self.state.requests.lock().unwrap().clone() + } + pub(crate) fn accepted_carriers(&self) -> usize { + self.state.accepted.load(Ordering::SeqCst) + } + pub(crate) async fn shutdown(self) { + self.task.abort(); + let _ = self.task.await; + } +} + +fn credential_and_acceptor(port: u16) -> (Credential, TlsAcceptor) { + let ca_key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(); + let mut ca_params = CertificateParams::new(Vec::::new()).unwrap(); + ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained); + ca_params.key_usages = vec![ + KeyUsagePurpose::DigitalSignature, + KeyUsagePurpose::KeyCertSign, + ]; + let ca = ca_params.self_signed(&ca_key).unwrap(); + let server_key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(); + let mut server_params = CertificateParams::new(vec!["spl.local".into()]).unwrap(); + server_params + .extended_key_usages + .push(ExtendedKeyUsagePurpose::ServerAuth); + let server = server_params.signed_by(&server_key, &ca, &ca_key).unwrap(); + let client_key = KeyPair::generate_for(&PKCS_ECDSA_P256_SHA256).unwrap(); + let mut client_params = CertificateParams::new(vec!["observer.test".into()]).unwrap(); + client_params + .extended_key_usages + .push(ExtendedKeyUsagePurpose::ClientAuth); + let client = client_params.signed_by(&client_key, &ca, &ca_key).unwrap(); + let ca_der = CertificateDer::from(ca.der().to_vec()); + let mut roots = RootCertStore::empty(); + roots.add(ca_der.clone()).unwrap(); + let verifier = WebPkiClientVerifier::builder(Arc::new(roots)) + .build() + .unwrap(); + let config = + ServerConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())) + .with_safe_default_protocol_versions() + .unwrap() + .with_client_cert_verifier(verifier) + .with_single_cert( + vec![CertificateDer::from(server.der().to_vec()), ca_der.clone()], + PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(server_key.serialize_der())), + ) + .unwrap(); + ( + Credential { + client_key_pem: client_key.serialize_pem(), + client_cert_pem: client.pem(), + ca_chain_pem: vec![ca.pem()], + ca_fp_prefix: spl_core::ca::sha256(ca_der.as_ref())[..16].to_vec(), + instance_id: "test-instance".into(), + home_label: "test home".into(), + endpoints: vec![EndpointAddr { + host: "127.0.0.1".into(), + port, + }], + home_attestation: None, + local_endpoints: None, + relay_origin: None, + device_token: None, + device_token_expires_at: None, + }, + TlsAcceptor::from(Arc::new(config)), + ) +} + +async fn serve_carrier(mut tls: TlsStream, state: &PeerState) -> io::Result<()> { + let mut decoder = FrameDecoder::new(); + let mut requests: HashMap> = HashMap::new(); + let mut outbound: HashMap = HashMap::new(); + let mut buffer = [0; 16 * 1024]; + loop { + let count = tls.read(&mut buffer).await?; + if count == 0 { + return Ok(()); + } + decoder.feed(&buffer[..count]); + for frame in decoder + .drain() + .map_err(|_| io::Error::other("frame decode"))? + { + if let Some(pong) = frame.control_pong() { + write_frame(&mut tls, pong).await?; + continue; + } + if frame.flags & FLAG_OPEN != 0 { + requests.entry(frame.stream_id).or_default(); + } + if frame.flags & FLAG_DATA != 0 { + state + .uploaded + .fetch_add(frame.payload.len(), Ordering::SeqCst); + requests + .entry(frame.stream_id) + .or_default() + .extend_from_slice(&frame.payload); + write_frame( + &mut tls, + Frame::window(frame.stream_id, frame.payload.len() as u32), + ) + .await?; + } + if frame.flags & FLAG_CLOSE != 0 { + let raw = requests.remove(&frame.stream_id).unwrap_or_default(); + if let Some(request) = parse_request(&raw) { + state.requests.lock().unwrap().push(request); + } + let response = + state + .responses + .lock() + .unwrap() + .pop_front() + .unwrap_or(PeerResponse { + status: 500, + body: Vec::new(), + }); + let mut response = encode_response(response); + flush_response(&mut tls, frame.stream_id, &mut response).await?; + if response.offset != response.bytes.len() { + outbound.insert(frame.stream_id, response); + } + } + if frame.flags & FLAG_RESET != 0 { + requests.remove(&frame.stream_id); + } + if frame.flags & FLAG_WINDOW != 0 + && let (Some(credit), Some(response)) = + (frame.window_credit(), outbound.get_mut(&frame.stream_id)) + { + response.credit = response.credit.saturating_add(credit as usize); + flush_response(&mut tls, frame.stream_id, response).await?; + if response.offset == response.bytes.len() { + outbound.remove(&frame.stream_id); + } + } + } + } +} + +async fn write_frame(tls: &mut TlsStream, frame: Frame) -> io::Result<()> { + tls.write_all( + &frame + .encode() + .map_err(|_| io::Error::other("frame encode"))?, + ) + .await +} + +fn encode_response(response: PeerResponse) -> OutboundResponse { + let mut bytes = format!( + "HTTP/1.1 {} OK\r\ncontent-length: {}\r\n\r\n", + response.status, + response.body.len() + ) + .into_bytes(); + bytes.extend(response.body); + OutboundResponse { + bytes, + offset: 0, + credit: INITIAL_WINDOW, + } +} + +async fn flush_response( + tls: &mut TlsStream, + stream: u32, + response: &mut OutboundResponse, +) -> io::Result<()> { + while response.offset < response.bytes.len() && response.credit > 0 { + let count = (response.bytes.len() - response.offset) + .min(RECOMMENDED_CHUNK) + .min(response.credit); + let end = response.offset + count; + let last = end == response.bytes.len(); + write_frame( + tls, + Frame::new( + stream, + FLAG_DATA | if last { FLAG_CLOSE } else { 0 }, + response.bytes[response.offset..end].to_vec(), + ), + ) + .await?; + response.offset = end; + response.credit -= count; + } + Ok(()) +} + +fn parse_request(raw: &[u8]) -> Option { + let split = raw.windows(4).position(|part| part == b"\r\n\r\n")?; + let head = std::str::from_utf8(&raw[..split]).ok()?; + let mut lines = head.split("\r\n"); + let mut request = lines.next()?.split_whitespace(); + let method = request.next()?.to_owned(); + let path = request.next()?.to_owned(); + let headers = lines + .filter_map(|line| line.split_once(':')) + .map(|(name, value)| (name.to_owned(), value.trim().to_owned())) + .collect(); + Some(PeerRequest { + method, + path, + headers, + body: raw[split + 4..].to_vec(), + }) +} diff --git a/crates/solstone-linux/src/toolchain_policy_tests.rs b/crates/solstone-linux/src/toolchain_policy_tests.rs index cecb525..084f00e 100644 --- a/crates/solstone-linux/src/toolchain_policy_tests.rs +++ b/crates/solstone-linux/src/toolchain_policy_tests.rs @@ -704,6 +704,50 @@ fn dependency_policy_denies_wildcards_and_unknown_sources() { } } +#[test] +fn dependency_policy_pins_the_single_spl_git_source() { + let root = workspace_root(); + let deny = read_toml(&root.join("deny.toml")); + let strings = |value: &toml::Value| { + value + .as_array() + .unwrap() + .iter() + .map(|item| item.as_str().unwrap().to_owned()) + .collect::>() + }; + assert_eq!( + strings(&deny["sources"]["allow-registry"]), + ["https://github.com/rust-lang/crates.io-index"] + ); + assert_eq!( + strings(&deny["sources"]["allow-git"]), + ["https://github.com/solpbc/spl-rust"] + ); + assert_eq!(deny["sources"]["unknown-registry"].as_str(), Some("deny")); + assert_eq!(deny["sources"]["unknown-git"].as_str(), Some("deny")); + assert_eq!(deny["bans"]["wildcards"].as_str(), Some("deny")); + + let manifest = read_toml(&root.join("crates/solstone-linux/Cargo.toml")); + for name in ["spl-core", "spl-transport"] { + let dependency = &manifest["dependencies"][name]; + assert_eq!(dependency["version"].as_str(), Some("0.1.0")); + assert_eq!( + dependency["git"].as_str(), + Some("https://github.com/solpbc/spl-rust") + ); + assert_eq!( + dependency["rev"].as_str(), + Some("742bc9dc789c5a75658844849a04d75033aeb6e3") + ); + } + for (name, dependency) in manifest["dependencies"].as_table().unwrap() { + if dependency.get("git").is_some() { + assert!(matches!(name.as_str(), "spl-core" | "spl-transport")); + } + } +} + #[test] fn observer_contract_gate_is_locked_offline_and_named() { let makefile = fs::read_to_string(workspace_root().join("Makefile")).unwrap(); diff --git a/crates/solstone-linux/src/upload.rs b/crates/solstone-linux/src/upload.rs index 023c8b4..023b128 100644 --- a/crates/solstone-linux/src/upload.rs +++ b/crates/solstone-linux/src/upload.rs @@ -762,6 +762,7 @@ mod tests { use super::*; use crate::{ config::{ConfigPaths, load_config}, + private_link_test_peer::PrivateLinkPeer, test_support::{Action, MockServer, MutableClock, wait_for_requests}, }; use tempfile::TempDir; @@ -836,6 +837,86 @@ mod tests { assert_eq!(loaded.config.stream, "fedora"); } + #[tokio::test] + async fn daemon_requests_remain_on_production_http_not_private_link() { + let server = MockServer::new(vec![ + (200, json!({"key":"K123456789", "name":"host-a"})), + (200, json!({"status":"ok", "key":"stored"})), + (200, json!({"segments":[]})), + ]) + .await; + let peer = PrivateLinkPeer::start().await; + let temp = TempDir::new().unwrap(); + let mut config = config(&server, &temp); + config.key.clear(); + config.stream = "host-a".into(); + let client = client(&config); + assert!(client.ensure_registered(&mut config).await); + let capture = write_file(&temp, "capture.jsonl", b"{\"event\":1}\n"); + assert!( + client + .upload_segment("20260101", "120000", &[capture]) + .await + .success + ); + assert!( + client + .get_server_segments("20260101") + .await + .segments + .is_some() + ); + + let requests = server.requests(); + assert_eq!(requests.len(), 3); + assert_eq!( + (requests[0].method.as_str(), requests[0].uri.as_str()), + ("POST", "/app/observer/register") + ); + assert_eq!( + (requests[1].method.as_str(), requests[1].uri.as_str()), + ("POST", "/app/observer/ingest") + ); + assert!( + requests[1] + .headers + .get("authorization") + .unwrap() + .to_str() + .unwrap() + .starts_with("Bearer K123456789") + ); + let upload = String::from_utf8_lossy(&requests[1].body); + assert!(upload.contains("name=\"day\"")); + assert!(upload.contains("20260101")); + assert!(upload.contains("name=\"segment\"")); + assert!(upload.contains("120000")); + assert!(upload.contains("name=\"files\"")); + assert!(upload.contains("{\"event\":1}")); + assert_eq!( + (requests[2].method.as_str(), requests[2].uri.as_str()), + ("GET", "/app/observer/ingest/segments/20260101") + ); + assert_eq!( + requests[2] + .headers + .get(OBSERVER_PROTOCOL_VERSION_HEADER) + .unwrap(), + "2" + ); + assert_eq!( + requests[2] + .headers + .get("authorization") + .unwrap() + .to_str() + .unwrap(), + "Bearer K123456789" + ); + assert!(peer.requests().is_empty()); + peer.shutdown().await; + } + // tests/test_upload.py::test_ensure_registered_skips_when_key_present // AC 18: constructor clock plumbing deliberately preserves the existing-key short circuit. #[tokio::test] diff --git a/deny.toml b/deny.toml index e660d44..b652624 100644 --- a/deny.toml +++ b/deny.toml @@ -32,3 +32,5 @@ wildcards = "deny" [sources] unknown-registry = "deny" unknown-git = "deny" +allow-registry = ["https://github.com/rust-lang/crates.io-index"] +allow-git = ["https://github.com/solpbc/spl-rust"]