diff --git a/.github/workflows/run_tests.yml b/.github/workflows/run_tests.yml index 27c0819d7c..0841b4c748 100644 --- a/.github/workflows/run_tests.yml +++ b/.github/workflows/run_tests.yml @@ -660,7 +660,7 @@ jobs: R2_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }} SCCACHE_BUCKET: sccache-zed - name: run_tests::check_wasm::cargo_check_wasm - run: cargo -Zbuild-std=std,panic_abort check --target wasm32-unknown-unknown -p gpui_platform + run: cargo -Zbuild-std=std,panic_abort check --target wasm32-unknown-unknown -p gpui_platform -p cloud_api_client env: CARGO_TARGET_WASM32_UNKNOWN_UNKNOWN_RUSTFLAGS: -C target-feature=+atomics,+bulk-memory,+mutable-globals RUSTC_BOOTSTRAP: '1' diff --git a/Cargo.lock b/Cargo.lock index 68e97949e0..97e65a3d0c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17486,7 +17486,7 @@ dependencies = [ "tracing", "url", "uuid", - "webpki-roots", + "webpki-roots 0.26.8", ] [[package]] @@ -21160,6 +21160,15 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "webpki-roots" +version = "1.0.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf85cb06032201fa7c6f829d7db5a7e5aa45bcc0655327713065f6f0576731bf" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "webrtc-sys" version = "0.3.23" @@ -22978,23 +22987,24 @@ checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" [[package]] name = "yawc" -version = "0.2.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19a5d82922135b4ae73a079a4ffb5501e9aadb4d785b8c660eaa0a8b899028c5" +version = "0.3.3" +source = "git+https://github.com/zed-industries/yawc?rev=71a452f551cac178367eaac5d7418a09afa1f3a2#71a452f551cac178367eaac5d7418a09afa1f3a2" dependencies = [ "base64 0.22.1", "bytes 1.11.1", "flate2", "futures 0.3.32", + "getrandom 0.2.16", "http-body-util", "hyper 1.7.0", "hyper-util", "js-sys", + "log", "nom 8.0.0", "pin-project", "rand 0.8.6", "sha1", - "thiserror 1.0.69", + "thiserror 2.0.17", "tokio", "tokio-rustls 0.26.4", "tokio-util", @@ -23002,7 +23012,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", + "webpki-roots 1.0.8", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 551596b61c..16cbb76609 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -888,7 +888,7 @@ webrtc-sys = "0.3.23" wgpu = "29.0.4" windows-core = "0.61" yaml-rust2 = "0.8" -yawc = "0.2.5" +yawc = { git = "https://github.com/zed-industries/yawc", rev = "71a452f551cac178367eaac5d7418a09afa1f3a2", version = "0.3.3" } zeroize = "1.8" zstd = "0.11" diff --git a/crates/cloud_api_client/Cargo.toml b/crates/cloud_api_client/Cargo.toml index baca3c1ce6..632cb9d5f3 100644 --- a/crates/cloud_api_client/Cargo.toml +++ b/crates/cloud_api_client/Cargo.toml @@ -16,7 +16,6 @@ anyhow.workspace = true cloud_api_types.workspace = true futures.workspace = true gpui.workspace = true -gpui_tokio.workspace = true http_client.workspace = true parking_lot.workspace = true serde.workspace = true @@ -24,3 +23,6 @@ serde_json.workspace = true async-lock.workspace = true thiserror.workspace = true yawc.workspace = true + +[target.'cfg(not(target_family = "wasm"))'.dependencies] +gpui_tokio.workspace = true diff --git a/crates/cloud_api_client/src/cloud_api_client.rs b/crates/cloud_api_client/src/cloud_api_client.rs index d9e584e313..b98b6c60c8 100644 --- a/crates/cloud_api_client/src/cloud_api_client.rs +++ b/crates/cloud_api_client/src/cloud_api_client.rs @@ -3,12 +3,9 @@ mod websocket; use std::sync::Arc; -use anyhow::{Context, Result, anyhow}; -use cloud_api_types::websocket_protocol::{PROTOCOL_VERSION, PROTOCOL_VERSION_HEADER_NAME}; +use anyhow::{Result, anyhow}; pub use cloud_api_types::*; use futures::AsyncReadExt as _; -use gpui::{App, Task}; -use gpui_tokio::Tokio; use http_client::http::request; use http_client::{ AsyncBody, HttpClientWithUrl, HttpRequestExt, Json, Method, Request, Response, StatusCode, @@ -16,9 +13,6 @@ use http_client::{ use parking_lot::RwLock; use serde::de::DeserializeOwned; use thiserror::Error; -use yawc::WebSocket; - -use crate::websocket::Connection; pub use llm_token::LlmApiToken; @@ -27,6 +21,12 @@ struct Credentials { access_token: String, } +#[derive(Clone, Copy)] +enum Authentication { + Credentials, + Session, +} + #[derive(Debug, Error)] pub enum ClientApiError { /// 401 — credentials are invalid or expired. @@ -62,6 +62,7 @@ pub enum ClientApiError { pub struct CloudApiClient { credentials: RwLock>, http_client: Arc, + authentication: Authentication, } impl CloudApiClient { @@ -69,6 +70,15 @@ impl CloudApiClient { Self { credentials: RwLock::new(None), http_client, + authentication: Authentication::Credentials, + } + } + + pub fn new_with_session_authentication(http_client: Arc) -> Self { + Self { + credentials: RwLock::new(None), + http_client, + authentication: Authentication::Session, } } @@ -115,35 +125,6 @@ impl CloudApiClient { .await } - pub fn connect(&self, cx: &App) -> Result>> { - let mut connect_url = self - .http_client - .build_zed_cloud_url("/client/users/connect")?; - connect_url - .set_scheme(match connect_url.scheme() { - "https" => "wss", - "http" => "ws", - scheme => Err(anyhow!("invalid URL scheme: {scheme}"))?, - }) - .map_err(|_| anyhow!("failed to set URL scheme"))?; - - let credentials = self.credentials.read(); - let credentials = credentials.as_ref().context("no credentials provided")?; - let authorization_header = format!("{} {}", credentials.user_id, credentials.access_token); - - Ok(Tokio::spawn_result(cx, async move { - let ws = WebSocket::connect(connect_url) - .with_request( - request::Builder::new() - .header("Authorization", authorization_header) - .header(PROTOCOL_VERSION_HEADER_NAME, PROTOCOL_VERSION.to_string()), - ) - .await?; - - Ok(Connection::new(ws)) - })) - } - async fn create_llm_token( &self, system_id: Option, @@ -203,10 +184,13 @@ impl CloudApiClient { request_builder: request::Builder, body: impl Into, ) -> Result, ClientApiError> { - let request = { + let request = if matches!(self.authentication, Authentication::Session) { + build_request(request_builder, body, None) + .map_err(ClientApiError::RequestBuildFailed)? + } else { let credentials = self.credentials.read(); let credentials = credentials.as_ref().ok_or(ClientApiError::NotSignedIn)?; - build_request(request_builder, body, credentials) + build_request(request_builder, body, Some(credentials)) .map_err(ClientApiError::RequestBuildFailed)? }; @@ -261,10 +245,10 @@ impl CloudApiClient { .as_ref(), ), AsyncBody::default(), - &Credentials { + Some(&Credentials { user_id, access_token: access_token.into(), - }, + }), )?; let mut response = self.http_client.send(request).await?; @@ -331,13 +315,60 @@ impl CloudApiClient { fn build_request( req: request::Builder, body: impl Into, - credentials: &Credentials, + credentials: Option<&Credentials>, ) -> Result> { Ok(req .header("Content-Type", "application/json") - .header( - "Authorization", - format!("{} {}", credentials.user_id, credentials.access_token), - ) + .when_some(credentials, |request, credentials| { + request.header( + "Authorization", + format!("{} {}", credentials.user_id, credentials.access_token), + ) + }) .body(body.into())?) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn build_session_authenticated_request_without_authorization_header() -> Result<()> { + let request = build_request( + Request::builder().uri("https://cloud.zed.dev/client/users/me"), + AsyncBody::default(), + None, + )?; + + assert_eq!( + request + .headers() + .get("Content-Type") + .and_then(|value| value.to_str().ok()), + Some("application/json") + ); + assert!(!request.headers().contains_key("Authorization")); + Ok(()) + } + + #[test] + fn build_credentials_authenticated_request_with_authorization_header() -> Result<()> { + let request = build_request( + Request::builder().uri("https://cloud.zed.dev/client/users/me"), + AsyncBody::default(), + Some(&Credentials { + user_id: 123, + access_token: "token".into(), + }), + )?; + + assert_eq!( + request + .headers() + .get("Authorization") + .and_then(|value| value.to_str().ok()), + Some("123 token") + ); + Ok(()) + } +} diff --git a/crates/cloud_api_client/src/websocket.rs b/crates/cloud_api_client/src/websocket.rs index 48a628db78..9f4c200ed7 100644 --- a/crates/cloud_api_client/src/websocket.rs +++ b/crates/cloud_api_client/src/websocket.rs @@ -1,73 +1,24 @@ use std::pin::Pin; -use std::time::Duration; use anyhow::Result; use cloud_api_types::websocket_protocol::MessageToClient; -use futures::channel::mpsc::unbounded; -use futures::stream::{SplitSink, SplitStream}; -use futures::{FutureExt as _, SinkExt as _, Stream, StreamExt as _, TryStreamExt as _, pin_mut}; -use gpui::{App, BackgroundExecutor, Task}; -use yawc::WebSocket; -use yawc::frame::{FrameView, OpCode}; +use futures::Stream; +use futures::channel::mpsc::UnboundedSender; +use yawc::frame::{Frame, OpCode}; -const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1); +#[cfg(not(target_family = "wasm"))] +mod native; +#[cfg(target_family = "wasm")] +mod web; pub type MessageStream = Pin>>>; -pub struct Connection { - tx: SplitSink, - rx: SplitStream, -} - -impl Connection { - pub fn new(ws: WebSocket) -> Self { - let (tx, rx) = ws.split(); - - Self { tx, rx } - } - - pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) { - let (mut tx, rx) = (self.tx, self.rx); - - let (message_tx, message_rx) = unbounded(); - - let handle_io = |executor: BackgroundExecutor| async move { - // Send messages on this frequency so the connection isn't closed. - let keepalive_timer = executor.timer(KEEPALIVE_INTERVAL).fuse(); - futures::pin_mut!(keepalive_timer); - - let rx = rx.fuse(); - pin_mut!(rx); - - loop { - futures::select_biased! { - _ = keepalive_timer => { - let _ = tx.send(FrameView::ping(Vec::new())).await; - - keepalive_timer.set(executor.timer(KEEPALIVE_INTERVAL).fuse()); - } - frame = rx.next() => { - let Some(frame) = frame else { - break; - }; - - match frame.opcode { - OpCode::Binary => { - let message_result = MessageToClient::deserialize(&frame.payload); - message_tx.unbounded_send(message_result).ok(); - } - OpCode::Close => { - break; - } - _ => {} - } - } - } - } - }; - - let task = cx.spawn(async move |cx| handle_io(cx.background_executor().clone()).await); - - (message_rx.into_stream().boxed(), task) +fn forward_frame(frame: Frame, message_tx: &UnboundedSender>) -> bool { + match frame.opcode() { + OpCode::Binary => message_tx + .unbounded_send(MessageToClient::deserialize(frame.payload())) + .is_ok(), + OpCode::Close => false, + OpCode::Continuation | OpCode::Text | OpCode::Ping | OpCode::Pong => true, } } diff --git a/crates/cloud_api_client/src/websocket/native.rs b/crates/cloud_api_client/src/websocket/native.rs new file mode 100644 index 0000000000..d19da9f002 --- /dev/null +++ b/crates/cloud_api_client/src/websocket/native.rs @@ -0,0 +1,91 @@ +use std::time::Duration; + +use anyhow::{Context as _, Result, anyhow}; +use cloud_api_types::websocket_protocol::{PROTOCOL_VERSION, PROTOCOL_VERSION_HEADER_NAME}; +use futures::channel::mpsc::unbounded; +use futures::stream::{SplitSink, SplitStream}; +use futures::{FutureExt as _, SinkExt as _, StreamExt as _, TryStreamExt as _}; +use gpui::{App, Task}; +use http_client::http::request; +use yawc::frame::Frame; +use yawc::{TcpWebSocket, WebSocket}; + +use super::{MessageStream, forward_frame}; +use crate::CloudApiClient; + +const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(1); + +pub struct Connection { + tx: SplitSink, + rx: SplitStream, +} + +impl Connection { + fn new(websocket: TcpWebSocket) -> Self { + let (tx, rx) = websocket.split(); + Self { tx, rx } + } + + pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) { + let mut tx = self.tx; + let rx = self.rx.fuse(); + let (message_tx, message_rx) = unbounded(); + let executor = cx.background_executor().clone(); + let task = cx.spawn(async move |_cx| { + let keepalive_timer = executor.timer(KEEPALIVE_INTERVAL).fuse(); + futures::pin_mut!(keepalive_timer, rx); + + loop { + futures::select_biased! { + _ = keepalive_timer => { + if tx.send(Frame::ping(Vec::new())).await.is_err() { + break; + } + keepalive_timer.set(executor.timer(KEEPALIVE_INTERVAL).fuse()); + } + frame = rx.next() => { + let Some(frame) = frame else { + break; + }; + if !forward_frame(frame, &message_tx) { + break; + } + } + } + } + }); + + (message_rx.into_stream().boxed(), task) + } +} + +impl CloudApiClient { + pub fn connect(self: &std::sync::Arc, cx: &App) -> Result>> { + let mut connect_url = self + .http_client + .build_zed_cloud_url("/client/users/connect")?; + connect_url + .set_scheme(match connect_url.scheme() { + "https" => "wss", + "http" => "ws", + scheme => Err(anyhow!("invalid URL scheme: {scheme}"))?, + }) + .map_err(|_| anyhow!("failed to set URL scheme"))?; + + let credentials = self.credentials.read(); + let credentials = credentials.as_ref().context("no credentials provided")?; + let authorization_header = format!("{} {}", credentials.user_id, credentials.access_token); + + Ok(gpui_tokio::Tokio::spawn_result(cx, async move { + let websocket = WebSocket::connect(connect_url) + .with_request( + request::Builder::new() + .header("Authorization", authorization_header) + .header(PROTOCOL_VERSION_HEADER_NAME, PROTOCOL_VERSION.to_string()), + ) + .await?; + + Ok(Connection::new(websocket)) + })) + } +} diff --git a/crates/cloud_api_client/src/websocket/web.rs b/crates/cloud_api_client/src/websocket/web.rs new file mode 100644 index 0000000000..223791cb9a --- /dev/null +++ b/crates/cloud_api_client/src/websocket/web.rs @@ -0,0 +1,79 @@ +use std::time::Duration; + +use anyhow::{Result, anyhow}; + +use futures::channel::mpsc::unbounded; +use futures::stream::SplitStream; +use futures::{FutureExt as _, StreamExt as _, TryStreamExt as _}; +use gpui::{App, Task}; +use yawc::WebSocket; + +use super::{MessageStream, forward_frame}; +use crate::CloudApiClient; + +const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); + +pub struct Connection { + rx: SplitStream, +} + +impl Connection { + fn new(websocket: WebSocket) -> Self { + let (_, rx) = websocket.split(); + Self { rx } + } + + pub fn spawn(self, cx: &App) -> (MessageStream, Task<()>) { + let mut rx = self.rx; + let (message_tx, message_rx) = unbounded(); + let task = cx.spawn(async move |_cx| { + while let Some(frame) = rx.next().await { + let frame = match frame { + Ok(frame) => frame, + Err(error) => { + let error = anyhow!("Cloud WebSocket error: {error}"); + if message_tx.unbounded_send(Err(error)).is_err() { + break; + } + continue; + } + }; + if !forward_frame(frame, &message_tx) { + break; + } + } + }); + + (message_rx.into_stream().boxed(), task) + } +} + +impl CloudApiClient { + pub fn connect(self: &std::sync::Arc, cx: &App) -> Result>> { + let client = self.clone(); + let executor = cx.background_executor().clone(); + Ok(cx.spawn(async move |_cx| { + let mut connect_url = client + .http_client + .build_zed_cloud_url("/client/users/connect")?; + connect_url + .set_scheme(match connect_url.scheme() { + "https" => "wss", + "http" => "ws", + scheme => return Err(anyhow!("invalid URL scheme: {scheme}")), + }) + .map_err(|_| anyhow!("failed to set URL scheme"))?; + + + let connect = WebSocket::connect(connect_url).fuse(); + let timeout = executor.timer(CONNECT_TIMEOUT).fuse(); + futures::pin_mut!(connect, timeout); + let websocket = futures::select_biased! { + result = connect => result.map_err(|error| anyhow!("failed to connect to Cloud WebSocket: {error}"))?, + _ = timeout => return Err(anyhow!("timed out connecting to Cloud WebSocket")), + }; + + Ok(Connection::new(websocket)) + })) + } +} diff --git a/script/licenses/zed-licenses.toml b/script/licenses/zed-licenses.toml index 17ebc15954..7f9c865616 100644 --- a/script/licenses/zed-licenses.toml +++ b/script/licenses/zed-licenses.toml @@ -20,6 +20,7 @@ accepted = [ "Zlib", "BSL-1.0", "bzip2-1.0.6", + "CDLA-Permissive-2.0", ] [procinfo.clarify] diff --git a/tooling/xtask/src/tasks/workflows/run_tests.rs b/tooling/xtask/src/tasks/workflows/run_tests.rs index 62043c545a..2707034eee 100644 --- a/tooling/xtask/src/tasks/workflows/run_tests.rs +++ b/tooling/xtask/src/tasks/workflows/run_tests.rs @@ -499,7 +499,7 @@ fn check_wasm() -> NamedJob { fn cargo_check_wasm() -> Step { named::bash(concat!( "cargo -Zbuild-std=std,panic_abort ", - "check --target wasm32-unknown-unknown -p gpui_platform", + "check --target wasm32-unknown-unknown -p gpui_platform -p cloud_api_client", )) .add_env(( "CARGO_TARGET_WASM32_UNKNOWN_UNKNOWN_RUSTFLAGS",