From 8e6575e9d0bc68dce13b30c04c1e81e331106275 Mon Sep 17 00:00:00 2001 From: Trezy Date: Mon, 16 Mar 2026 22:32:56 -0500 Subject: [PATCH] fix: resolve proper endpoints for labelers --- Cargo.lock | 104 +++++++++++++++++++++++++++++++++++++++++++++++++ Cargo.toml | 4 +- src/labeler.rs | 76 +++++++++++++++++++++++++++++------- src/profile.rs | 18 +++++++++ 4 files changed, 187 insertions(+), 15 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b608bf9..1e7fed9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -83,6 +83,28 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-lc-rs" +version = "1.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94bffc006df10ac2a68c83692d734a465f8ee6c5b384d8545a636f81d858f4bf" +dependencies = [ + "aws-lc-sys", + "zeroize", +] + +[[package]] +name = "aws-lc-sys" +version = "0.38.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4321e568ed89bb5a7d291a7f37997c2c0df89809d7b6d12062c81ddb54aa782e" +dependencies = [ + "cc", + "cmake", + "dunce", + "fs_extra", +] + [[package]] name = "axum" version = "0.8.8" @@ -206,6 +228,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "47b26a0954ae34af09b50f0de26458fa95369a0d478d8236d3f93082b219bd29" dependencies = [ "find-msvc-tools", + "jobserver", + "libc", "shlex", ] @@ -229,6 +253,42 @@ dependencies = [ "windows-link", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + +[[package]] +name = "cmake" +version = "0.1.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "75443c44cd6b379beb8c5b45d85d0773baf31cce901fe7bb252f4eff3008ef7d" +dependencies = [ + "cc", +] + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -323,6 +383,12 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-bigint" version = "0.5.5" @@ -432,6 +498,12 @@ version = "0.15.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aaf95b3e5c8f23aa320147307562d361db0ae0d51242340f558153b4eb2439b" +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "ecdsa" version = "0.16.9" @@ -620,6 +692,12 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures" version = "0.3.31" @@ -799,6 +877,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + [[package]] name = "happyview" version = "0.1.0" @@ -808,6 +897,7 @@ dependencies = [ "base64", "bytes", "chrono", + "ciborium", "dashmap", "dotenvy", "futures-util", @@ -827,6 +917,7 @@ dependencies = [ "sha2", "sqlx", "tokio", + "tokio-rustls", "tokio-tungstenite", "tower", "tower-http", @@ -1270,6 +1361,16 @@ version = "1.0.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92ecc6618181def0457392ccd0ee51198e065e016d1d527a7ac1b6dc7c1f09d2" +[[package]] +name = "jobserver" +version = "0.1.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" +dependencies = [ + "getrandom 0.3.4", + "libc", +] + [[package]] name = "js-sys" version = "0.3.85" @@ -2058,6 +2159,8 @@ version = "0.23.36" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c665f33d38cea657d9614f766881e4d510e0eda4239891eea56b4cadcf01801b" dependencies = [ + "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -2081,6 +2184,7 @@ version = "0.103.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53" dependencies = [ + "aws-lc-rs", "ring", "rustls-pki-types", "untrusted", diff --git a/Cargo.toml b/Cargo.toml index 067c9cf..9eacb6f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,16 +14,18 @@ futures-util = "0.3" jsonwebtoken = "9" bytes = "1" chrono = { version = "0.4", features = ["serde"] } +ciborium = "0.2" p256 = { version = "0.13", features = ["pkcs8"] } uuid = { version = "1", features = ["v4"] } rand = "0.9" reqwest = { version = "0.12", features = ["json"] } -rustls = { version = "0.23", default-features = false } +rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12"] } serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.10" sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "json", "chrono", "migrate"] } tokio = { version = "1", features = ["full"] } +tokio-rustls = "0.26" tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] } tower = { version = "0.5", features = ["util"] } tower-http = { version = "0.6", features = ["cors", "fs", "trace"] } diff --git a/src/labeler.rs b/src/labeler.rs index 0902f59..751f248 100644 --- a/src/labeler.rs +++ b/src/labeler.rs @@ -3,9 +3,9 @@ use std::sync::Arc; use futures_util::StreamExt; use serde::Deserialize; +use tokio::net::TcpStream; use tokio::sync::watch; use tokio::task::JoinHandle; -use tokio_tungstenite::Connector; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::client::IntoClientRequest; @@ -17,6 +17,14 @@ use crate::profile; // Types // --------------------------------------------------------------------------- +/// AT Protocol event stream frame header (DAG-CBOR). +#[derive(Deserialize)] +struct FrameHeader { + op: i64, + #[serde(default)] + t: Option, +} + #[derive(Deserialize)] struct SubscribeLabelsMessage { seq: i64, @@ -129,10 +137,11 @@ async fn run_subscription_once( state: &AppState, did: &str, ) -> Result<(), Box> { - // Resolve the labeler's PDS endpoint from its DID. - let pds_endpoint = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did) + // Resolve the labeler's service endpoint from its DID document. + // Prefers #atproto_labeler, falls back to #atproto_pds. + let pds_endpoint = profile::resolve_labeler_endpoint(&state.http, &state.config.plc_url, did) .await - .map_err(|e| format!("failed to resolve PDS for {did}: {e:?}"))?; + .map_err(|e| format!("failed to resolve labeler endpoint for {did}: {e:?}"))?; // Convert HTTP URL to WebSocket URL. let ws_url = http_to_ws(&pds_endpoint); @@ -156,9 +165,19 @@ async fn run_subscription_once( let request = url.into_client_request()?; - // Build a rustls config that only advertises HTTP/1.1 in ALPN. - // Without this, rustls negotiates h2 and the WebSocket upgrade fails - // ("HTTP version must be 1.1 or higher"). + // Manually establish TCP + TLS with HTTP/1.1 ALPN, then do the + // WebSocket handshake over the established stream. This avoids + // tokio-tungstenite's default TLS which may negotiate h2 via ALPN. + let host = request + .uri() + .host() + .ok_or("missing host in WebSocket URL")? + .to_string(); + let port = request.uri().port_u16().unwrap_or(443); + + let tcp = TcpStream::connect((&*host, port)).await?; + + let _ = rustls::crypto::ring::default_provider().install_default(); let root_store = rustls::RootCertStore::from_iter(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); let mut tls_config = rustls::ClientConfig::builder() @@ -166,11 +185,11 @@ async fn run_subscription_once( .with_no_client_auth(); tls_config.alpn_protocols = vec![b"http/1.1".to_vec()]; - let connector = Connector::Rustls(Arc::new(tls_config)); + let connector = tokio_rustls::TlsConnector::from(Arc::new(tls_config)); + let domain = rustls::pki_types::ServerName::try_from(host)?; + let tls_stream = connector.connect(domain, tcp).await?; - let (ws, _) = - tokio_tungstenite::connect_async_tls_with_config(request, None, false, Some(connector)) - .await?; + let (ws, _) = tokio_tungstenite::client_async(request, tls_stream).await?; tracing::info!(did = %did, "connected to labeler"); @@ -206,7 +225,6 @@ async fn run_subscription_once( }; let bytes = match msg { - Message::Text(t) => t.as_bytes().to_vec(), Message::Binary(b) => b.to_vec(), Message::Close(_) => { tracing::info!(did = %did, "labeler websocket received close frame"); @@ -215,8 +233,12 @@ async fn run_subscription_once( _ => continue, }; - let message: SubscribeLabelsMessage = match serde_json::from_slice(&bytes) { - Ok(m) => m, + // AT Protocol event streams use two concatenated DAG-CBOR objects: + // 1. Frame header: { op: int, t: string? } + // 2. Frame body: the actual message payload + let message: SubscribeLabelsMessage = match parse_event_frame(&bytes) { + Ok(Some(m)) => m, + Ok(None) => continue, // non-message frame (error, info, etc.) Err(e) => { tracing::warn!(did = %did, "skipping unparseable labeler message: {e}"); continue; @@ -258,6 +280,32 @@ async fn run_subscription_once( // Helpers // --------------------------------------------------------------------------- +/// Parse an AT Protocol event stream frame (two concatenated DAG-CBOR objects). +/// Returns `Ok(Some(msg))` for label messages, `Ok(None)` for other frame types. +fn parse_event_frame( + bytes: &[u8], +) -> Result, Box> { + let mut cursor = std::io::Cursor::new(bytes); + + // Decode frame header. + let header: FrameHeader = ciborium::from_reader(&mut cursor)?; + + // op=1 is a regular message, op=-1 is an error frame. + if header.op != 1 { + return Ok(None); + } + + // Only process #labels messages. + match header.t.as_deref() { + Some("#labels") => {} + _ => return Ok(None), + } + + // Decode the body (remaining bytes after header). + let message: SubscribeLabelsMessage = ciborium::from_reader(&mut cursor)?; + Ok(Some(message)) +} + fn http_to_ws(url: &str) -> String { let base = url.trim_end_matches('/'); if let Some(rest) = base.strip_prefix("https://") { diff --git a/src/profile.rs b/src/profile.rs index d970ac2..5fffbcc 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -98,6 +98,24 @@ pub async fn resolve_pds_endpoint( .ok_or_else(|| AppError::NotFound("no PDS endpoint in DID document".into())) } +/// Resolve the labeler service endpoint for a DID. +/// Tries `#atproto_labeler` first, falls back to `#atproto_pds`. +pub async fn resolve_labeler_endpoint( + http: &reqwest::Client, + plc_url: &str, + did: &str, +) -> Result { + let did_doc = resolve_did_document(http, plc_url, did).await?; + + did_doc + .service + .iter() + .find(|s| s.id == "#atproto_labeler") + .or_else(|| did_doc.service.iter().find(|s| s.id == "#atproto_pds")) + .map(|s| s.service_endpoint.clone()) + .ok_or_else(|| AppError::NotFound("no labeler or PDS endpoint in DID document".into())) +} + /// Fetch a DID document from the PLC directory or via `did:web` resolution. async fn resolve_did_document( http: &reqwest::Client, -- 2.51.2