diff --git a/Cargo.lock b/Cargo.lock index dc028071..bb94f504 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -619,6 +619,8 @@ dependencies = [ "http", "http-body", "http-body-util", + "hyper", + "hyper-util", "itoa", "matchit", "memchr", @@ -627,10 +629,15 @@ dependencies = [ "pin-project-lite", "rustversion", "serde", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", "sync_wrapper", + "tokio", "tower 0.5.2", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -651,6 +658,7 @@ dependencies = [ "sync_wrapper", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -6673,6 +6681,12 @@ dependencies = [ "regex-syntax 0.8.5", ] +[[package]] +name = "regex-lite" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973" + [[package]] name = "regex-syntax" version = "0.5.6" @@ -9531,6 +9545,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -9548,6 +9563,7 @@ dependencies = [ "tower 0.5.2", "tower-layer", "tower-service", + "tracing", "url", ] @@ -9569,6 +9585,7 @@ version = "0.1.41" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" dependencies = [ + "log 0.4.29", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -11490,7 +11507,6 @@ dependencies = [ "log 0.4.29", "mime_guess", "native-tls", - "regex 1.11.1", "reqwest 0.12.20", "serde", "serde_json", @@ -11550,6 +11566,7 @@ dependencies = [ "nanoid", "r2d2", "r2d2_sqlite", + "regex-lite", "rusqlite", "schemars 0.8.22", "sea-query", @@ -11559,8 +11576,10 @@ dependencies = [ "sha2", "thiserror 2.0.17", "ts-rs", + "urlencoding", "yaak-core", "yaak-database", + "yaak-templates", ] [[package]] @@ -11654,6 +11673,28 @@ dependencies = [ "yaak-ws", ] +[[package]] +name = "yaak-send-proxy" +version = "0.1.0" +dependencies = [ + "async-trait", + "axum", + "base64 0.22.1", + "bytes", + "clap", + "env_logger", + "futures-util", + "log 0.4.29", + "serde", + "serde_json", + "tokio", + "tower-http", + "url", + "uuid", + "yaak-http", + "yaak-models", +] + [[package]] name = "yaak-sse" version = "0.1.0" @@ -11746,6 +11787,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "yaak-models", + "yaak-templates", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index d1542d74..9c609bf6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,6 +26,8 @@ members = [ "crates/yaak-proxy", # Proxy-specific crates "crates-proxy/yaak-proxy-lib", + # Server crates (the browser tier's hosted send executor) + "crates-server/yaak-send-proxy", # CLI crates "crates-cli/yaak-cli", # Tauri-specific crates diff --git a/crates-server/yaak-send-proxy/Cargo.toml b/crates-server/yaak-send-proxy/Cargo.toml new file mode 100644 index 00000000..2d8f5c57 --- /dev/null +++ b/crates-server/yaak-send-proxy/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "yaak-send-proxy" +version = "0.1.0" +edition = "2024" +publish = false +description = "Stateless HTTP send executor for Yaak in the browser" + +# The send engine (yaak-http) and the model types it speaks (yaak-models, for +# HttpRequest / Cookie / HttpResponseEventData). Deliberately NOT yaak (the +# render + storage orchestration), yaak-plugins, or the RPC router: this binary +# opens no database, runs no plugins, and renders nothing. yaak-models comes +# along only because yaak-http's types are its types; nothing here calls into +# its query layer. + +[[bin]] +name = "yaak-send-proxy" +path = "src/main.rs" + +[dependencies] +async-trait = "0.1" +axum = { version = "0.7", features = ["http1", "http2", "json", "tokio"] } +base64 = "0.22.1" +bytes = "1.11.1" +clap = { version = "4.5", features = ["derive", "env"] } +env_logger = "0.11" +futures-util = "0.3" +log = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "signal", "sync", "io-util", "time", "net"] } +tower-http = { version = "0.6", features = ["cors", "trace"] } +url = "2" +uuid = { version = "1", features = ["v4"] } +yaak-http = { workspace = true } +yaak-models = { workspace = true } diff --git a/crates-server/yaak-send-proxy/README.md b/crates-server/yaak-send-proxy/README.md new file mode 100644 index 00000000..62c3d454 --- /dev/null +++ b/crates-server/yaak-send-proxy/README.md @@ -0,0 +1,134 @@ +# yaak-send-proxy + +The network half of Yaak in a browser. + +A tab can't see an HTTP response the way a desktop app can: CORS hides most +headers (2 of 8 in a typical response), redirects are followed silently, and +there is no timeline. So the tab renders the request and posts it here, and this +process puts it on the network with the desktop's own engine (`yaak-http`) and +streams back everything that happened — every header, every redirect hop, DNS +timing, the body — for the tab to store. + +It is a **stateless executor**. It keeps nothing: no database, no files, no +sessions, no cookies between calls. Every byte it sees comes from the tab in the +request, and every byte it returns is stored by the tab. Restart it any time. + +## Running it + +```shell +cargo run -p yaak-send-proxy +``` + +Listens on `127.0.0.1:9227`. Then run the web build against it: + +```shell +YAAK_TARGET=web npm run dev --workspace @yaakapp/yaak-client +``` + +The tab looks for the proxy at `http://127.0.0.1:9227` unless +`VITE_YAAK_SEND_PROXY_URL` says otherwise at build time. + +Every flag has a `YAAK_PROXY_*` environment variable, so a container needs no +arguments; `--help` lists them all. + +| Flag | Default | What | +| --- | --- | --- | +| `--bind` | `127.0.0.1:9227` | Listen address. `0.0.0.0:9227` inside a container. | +| `--allowed-origins` | `*` | CORS origins, comma-separated. A hosted instance should name its web origin. | +| `--token` | unset | Require `Authorization: Bearer `. Unset means anonymous, which is what the hosted funnel wants alongside the rate limit. | +| `--allow-private-networks` | off | Let sends reach private, loopback and link-local addresses. **Off by default; see below.** | +| `--allow-hosts` | empty | Only these hosts (`api.example.com`, `*.example.com`). Empty means any host not denied. | +| `--deny-hosts` | empty | Never these hosts. Checked before the allow list. | +| `--max-request-bytes` | 16 MiB | Largest rendered request accepted from the tab. | +| `--max-response-bytes` | 64 MiB | Largest upstream body relayed before the send is cut off. | +| `--max-timeout-secs` | 60 | Ceiling on a send's timeout; a request asking for more (or none) gets this. | +| `--rate-limit-per-minute` | 120 | Sends per client IP per minute; 0 disables. | +| `--max-concurrent` | 256 | Sends in flight at once. | +| `--trust-forwarded-for` | off | Take the client IP from `X-Forwarded-For`. Only behind a load balancer that sets it. | + +## What it refuses, and why + +A hosted proxy is, by construction, a machine that makes HTTP requests on +behalf of strangers. Left alone that is an open relay into whatever network it +sits on. So by default it refuses to connect to: + +- loopback (`127/8`, `::1`), private (`10/8`, `172.16/12`, `192.168/16`, + `fc00::/7`), link-local (`169.254/16` — where cloud metadata lives — and + `fe80::/10`), carrier-grade NAT, multicast, reserved and unspecified ranges, + and IPv4 addresses tunnelled inside IPv6 forms (`::ffff:a.b.c.d`, NAT64); +- anything not `http://` or `https://`. + +The check runs **on the resolved addresses, after DNS**, for every hop of a +redirect chain, so a public hostname that points at an internal address is +caught, and so is a `Location:` header that points at one. It also refuses body +types that would read files on the proxy's disk (`binary`, multipart file +fields), since no browser tab could legitimately mean those. + +Refusals are logged with the reason. A self-hosted instance on a private network +that legitimately needs to reach the services next to it turns the range check +off with `--allow-private-networks`, and can narrow that with `--allow-hosts`. + +## Self-hosting + +One binary, no dependencies. Build it and run it wherever you like: + +```shell +cargo build --release -p yaak-send-proxy +YAAK_PROXY_BIND=0.0.0.0:9227 \ +YAAK_PROXY_ALLOWED_ORIGINS=https://yaak.example.com \ +YAAK_PROXY_TOKEN=change-me \ +./target/release/yaak-send-proxy +``` + +Put TLS in front of it (a reverse proxy) — the token travels as a header. If the +reverse proxy buffers responses, tell it not to: the reply is a stream and the +`X-Accel-Buffering: no` header it sets is honoured by nginx-shaped ones. + +## The wire + +`POST /v1/http/send` with a JSON body: + +```json +{ + "request": { "url": "https://…", "method": "GET", "headers": […], "body": {…}, "bodyType": null, "urlParameters": […] }, + "settings": { "validateCertificates": true, "followRedirects": true, "timeoutMs": 0, "sendCookies": true, "storeCookies": true }, + "cookies": [ … ] +} +``` + +`request` is a Yaak `HttpRequest` in the desktop's own model shape with every +template already rendered by the tab; the proxy builds the URL, headers and +body from it exactly the way the desktop does after rendering. `cookies` is the +jar's contents (or `null` for no jar). + +The reply is `application/x-ndjson`, one JSON frame per line, in the order things +happened: + +| `type` | When | Carries | +| --- | --- | --- | +| `event` | as the engine produces them | one timeline event, in the desktop's `http_response_event.event` shape | +| `response` | once, when the final hop's headers arrive | status, all headers, request headers as sent, remote address, HTTP version, timing | +| `body` | as the body is read | a decompressed chunk, base64 | +| `done` | last, on success | elapsed, byte counts, and the cookie jar as the send left it | +| `error` | last, on failure | the reason, and any cookies collected before the failure | + +Refusals that happen before anything is sent (a blocked destination, a bad body, +rate limit, missing token) are plain HTTP errors (`403`, `400`, `429`, `401`) +with `{"error": "…"}`, not streams. + +Why a streamed HTTP response and not a WebSocket: one `POST` is stateless by +construction, cancellable by closing the connection, readable with `curl`, and +needs no upgrade handling on either side. A WebSocket only earns its keep when +traffic is bidirectional, which a single send is not. + +`GET /v1/health` reports the version and the effective limits. + +## What comes later + +Not built, by design, but the router is shaped for it: a WebSocket relay +(`/v1/ws/relay`) and a gRPC relay (`/v1/grpc/relay`) would be long-lived, +bidirectional endpoints on the same binary, behind the same destination policy, +limits and token. They differ from this endpoint in holding per-connection +in-memory state while a connection is open (never persisted), which brings +connection limits and a larger abuse surface — the reason they are separate +work. diff --git a/crates-server/yaak-send-proxy/src/config.rs b/crates-server/yaak-send-proxy/src/config.rs new file mode 100644 index 00000000..52aab400 --- /dev/null +++ b/crates-server/yaak-send-proxy/src/config.rs @@ -0,0 +1,78 @@ +use clap::Parser; +use std::net::SocketAddr; + +/// A stateless HTTP send executor for Yaak running in a browser. +/// +/// The tab renders the request and owns the data; this binary only puts bytes on the +/// network and streams back what came back. Nothing is written to disk or a database. +#[derive(Parser, Debug, Clone)] +#[command(name = "yaak-send-proxy", version, about, long_about = None)] +pub struct Config { + /// Address to listen on. 127.0.0.1 for a local instance; 0.0.0.0 inside a container. + #[arg(long, env = "YAAK_PROXY_BIND", default_value = "127.0.0.1:9227")] + pub bind: SocketAddr, + + /// Browser origins allowed to call this proxy (CORS), comma-separated. `*` allows any. + /// A local dev instance wants the Vite origin; a hosted instance wants its own web origin. + #[arg( + long, + env = "YAAK_PROXY_ALLOWED_ORIGINS", + default_value = "*", + value_delimiter = ',' + )] + pub allowed_origins: Vec, + + /// Require `Authorization: Bearer ` on every send. Unset means anonymous access, + /// which is what the hosted funnel wants alongside the rate limit. + #[arg(long, env = "YAAK_PROXY_TOKEN")] + pub token: Option, + + /// Allow sends to private, loopback, link-local and other non-public addresses. + /// + /// Off by default: a hosted instance must not become a relay into its own network. A + /// self-hosted instance on a private network legitimately needs this on to reach the + /// services next to it. + #[arg( + long, + env = "YAAK_PROXY_ALLOW_PRIVATE_NETWORKS", + default_value_t = false + )] + pub allow_private_networks: bool, + + /// Only allow sends to these hosts (exact host, or `*.suffix`), comma-separated. Empty means + /// every host not on the deny list. + #[arg(long, env = "YAAK_PROXY_ALLOW_HOSTS", value_delimiter = ',')] + pub allow_hosts: Vec, + + /// Never send to these hosts (exact host, or `*.suffix`), comma-separated. Checked before the + /// allow list. + #[arg(long, env = "YAAK_PROXY_DENY_HOSTS", value_delimiter = ',')] + pub deny_hosts: Vec, + + /// Largest request the proxy accepts from the tab (the rendered request JSON, body included). + #[arg(long, env = "YAAK_PROXY_MAX_REQUEST_BYTES", default_value_t = 16 * 1024 * 1024)] + pub max_request_bytes: usize, + + /// Largest upstream response body the proxy will relay before cutting the send off. + #[arg(long, env = "YAAK_PROXY_MAX_RESPONSE_BYTES", default_value_t = 64 * 1024 * 1024)] + pub max_response_bytes: usize, + + /// Ceiling on a send's timeout, in seconds. A request asking for longer (or for no timeout) + /// gets this instead. + #[arg(long, env = "YAAK_PROXY_MAX_TIMEOUT_SECS", default_value_t = 60)] + pub max_timeout_secs: u64, + + /// Sends allowed per client IP per minute. 0 disables the limit. + #[arg(long, env = "YAAK_PROXY_RATE_LIMIT_PER_MINUTE", default_value_t = 120)] + pub rate_limit_per_minute: u32, + + /// Sends in flight at once across all clients. + #[arg(long, env = "YAAK_PROXY_MAX_CONCURRENT", default_value_t = 256)] + pub max_concurrent: usize, + + /// Take the client IP from `X-Forwarded-For` (first hop) instead of the socket. Only turn + /// this on behind a load balancer that sets the header; otherwise anyone can spoof their way + /// past the rate limit. + #[arg(long, env = "YAAK_PROXY_TRUST_FORWARDED_FOR", default_value_t = false)] + pub trust_forwarded_for: bool, +} diff --git a/crates-server/yaak-send-proxy/src/guard.rs b/crates-server/yaak-send-proxy/src/guard.rs new file mode 100644 index 00000000..412ab31b --- /dev/null +++ b/crates-server/yaak-send-proxy/src/guard.rs @@ -0,0 +1,310 @@ +//! Where a send may go. +//! +//! A hosted sender is, by construction, a machine that makes HTTP requests on +//! behalf of strangers. Left alone that is an open relay into whatever network +//! it sits on: cloud metadata endpoints, internal admin panels, the database +//! next door. So every destination is checked twice — once on the URL before a +//! hop is attempted (literal IPs, host allow/deny lists) and once on the +//! addresses a hostname actually resolves to, right before the connection is +//! made. The second check is the one that matters for a hostname pointing at +//! an internal address, and it runs on every redirect hop because the engine +//! resolves every hop. + +use async_trait::async_trait; +use log::warn; +use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; +use std::sync::Arc; +use tokio::sync::mpsc; +use url::Url; +use yaak_http::dns::AddressFilter; +use yaak_http::sender::{HttpResponse, HttpResponseEvent, HttpSender}; +use yaak_http::types::SendableHttpRequest; + +/// The destination policy, built once from config and shared by every send. +#[derive(Clone)] +pub struct DestinationPolicy { + allow_private_networks: bool, + allow_hosts: Vec, + deny_hosts: Vec, +} + +#[derive(Clone, Debug)] +enum HostPattern { + Exact(String), + /// `*.example.com`: any subdomain, and the bare domain too. + Suffix(String), +} + +impl HostPattern { + fn parse(raw: &str) -> Option { + let raw = raw.trim().trim_end_matches('.').to_ascii_lowercase(); + if raw.is_empty() { + return None; + } + Some(match raw.strip_prefix("*.") { + Some(suffix) => Self::Suffix(suffix.to_string()), + None => Self::Exact(raw), + }) + } + + fn matches(&self, host: &str) -> bool { + match self { + Self::Exact(h) => host == h, + Self::Suffix(s) => host == s || host.strip_suffix(s).is_some_and(|p| p.ends_with('.')), + } + } +} + +impl DestinationPolicy { + pub fn new( + allow_private_networks: bool, + allow_hosts: &[String], + deny_hosts: &[String], + ) -> Self { + Self { + allow_private_networks, + allow_hosts: allow_hosts.iter().filter_map(|h| HostPattern::parse(h)).collect(), + deny_hosts: deny_hosts.iter().filter_map(|h| HostPattern::parse(h)).collect(), + } + } + + /// Check a URL before a hop is attempted: scheme, host lists, and literal IPs. A hostname + /// that passes here still has its resolved addresses checked by [`Self::address_filter`]. + pub fn check_url(&self, raw: &str) -> Result<(), String> { + let url = Url::parse(raw).map_err(|e| format!("Invalid URL {raw:?}: {e}"))?; + match url.scheme() { + "http" | "https" => {} + other => return Err(format!("Refusing to send over {other:?}; only http and https")), + } + let host = url.host_str().ok_or_else(|| format!("URL {raw:?} has no host"))?; + let host = + host.trim_matches(|c| c == '[' || c == ']').trim_end_matches('.').to_ascii_lowercase(); + + if self.deny_hosts.iter().any(|p| p.matches(&host)) { + return Err(format!("Host {host:?} is on this proxy's deny list")); + } + if !self.allow_hosts.is_empty() && !self.allow_hosts.iter().any(|p| p.matches(&host)) { + return Err(format!("Host {host:?} is not on this proxy's allow list")); + } + + // A literal IP never reaches the resolver, so it is checked here. Hostnames are checked + // where their addresses become known. + if let Ok(ip) = host.parse::() { + self.check_ip(ip)?; + } + Ok(()) + } + + /// The veto the engine's resolver applies to every address a hostname resolves to. + pub fn address_filter(&self) -> AddressFilter { + let policy = self.clone(); + Arc::new(move |ip| policy.check_ip(ip)) + } + + pub fn check_ip(&self, ip: IpAddr) -> Result<(), String> { + if self.allow_private_networks { + return Ok(()); + } + match non_public_reason(ip) { + Some(reason) => Err(format!( + "Refusing to connect to {ip}: {reason}. This proxy only sends to public addresses" + )), + None => Ok(()), + } + } +} + +/// Why an address is not a public internet address, or `None` if it is one. +/// +/// Every range here is one a hosted relay must never be talked into reaching: the machine +/// itself, the network it sits on, and the link-local range where cloud metadata services +/// (169.254.169.254) live. IPv4 addresses tunnelled inside IPv6 forms are unwrapped and judged +/// as IPv4, since that is what the socket would connect to. +pub fn non_public_reason(ip: IpAddr) -> Option<&'static str> { + match ip { + IpAddr::V4(v4) => non_public_v4(v4), + IpAddr::V6(v6) => { + if let Some(v4) = v6.to_ipv4_mapped() { + return non_public_v4(v4); + } + if let Some(v4) = nat64_embedded_v4(&v6) { + return non_public_v4(v4); + } + if v6.is_loopback() { + Some("loopback") + } else if v6.is_unspecified() { + Some("unspecified") + } else if v6.is_unique_local() { + Some("unique local (fc00::/7)") + } else if v6.is_unicast_link_local() { + Some("link-local (fe80::/10)") + } else if v6.is_multicast() { + Some("multicast") + } else if (v6.segments()[0] & 0xffc0) == 0xfec0 { + Some("site-local (fec0::/10)") + } else if v6.segments()[0] == 0x2001 && v6.segments()[1] == 0x0db8 { + Some("documentation (2001:db8::/32)") + } else { + None + } + } + } +} + +fn non_public_v4(v4: Ipv4Addr) -> Option<&'static str> { + let o = v4.octets(); + if v4.is_loopback() { + Some("loopback (127.0.0.0/8)") + } else if v4.is_private() { + Some("private (10/8, 172.16/12, 192.168/16)") + } else if v4.is_link_local() { + Some("link-local (169.254.0.0/16, where cloud metadata lives)") + } else if v4.is_unspecified() || o[0] == 0 { + Some("this network (0.0.0.0/8)") + } else if o[0] == 100 && (o[1] & 0xc0) == 64 { + Some("carrier-grade NAT (100.64.0.0/10)") + } else if v4.is_broadcast() { + Some("broadcast") + } else if v4.is_multicast() { + Some("multicast (224.0.0.0/4)") + } else if o[0] >= 240 { + Some("reserved (240.0.0.0/4)") + } else if v4.is_documentation() { + Some("documentation") + } else if o[0] == 192 && o[1] == 0 && o[2] == 0 { + Some("IETF protocol assignments (192.0.0.0/24)") + } else if o[0] == 198 && (o[1] & 0xfe) == 18 { + Some("benchmarking (198.18.0.0/15)") + } else { + None + } +} + +/// The IPv4 address inside a NAT64 (64:ff9b::/96) address, if this is one. +fn nat64_embedded_v4(v6: &Ipv6Addr) -> Option { + let s = v6.segments(); + if s[0] == 0x64 && s[1] == 0xff9b && s[2..6].iter().all(|x| *x == 0) { + let o = v6.octets(); + Some(Ipv4Addr::new(o[12], o[13], o[14], o[15])) + } else { + None + } +} + +/// An [`HttpSender`] that checks each hop's URL against the policy before delegating. +/// +/// The engine's redirect loop calls the sender once per hop with the hop's URL, so wrapping +/// the sender is what makes `Location:` headers subject to the same rules as the first URL — +/// including a redirect to a literal internal IP, which the resolver would never see. +pub struct GuardedSender { + inner: S, + policy: DestinationPolicy, +} + +impl GuardedSender { + pub fn new(inner: S, policy: DestinationPolicy) -> Self { + Self { inner, policy } + } +} + +#[async_trait] +impl HttpSender for GuardedSender { + async fn send( + &self, + request: SendableHttpRequest, + event_tx: mpsc::Sender, + ) -> yaak_http::error::Result { + if let Err(reason) = self.policy.check_url(&request.url) { + warn!("Refused {} {}: {reason}", request.method, request.url); + return Err(yaak_http::error::Error::RequestError(reason)); + } + self.inner.send(request, event_tx).await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ip(s: &str) -> IpAddr { + s.parse().unwrap() + } + + #[test] + fn refuses_the_ranges_a_relay_must_never_reach() { + for addr in [ + "127.0.0.1", + "127.9.9.9", + "10.0.0.1", + "172.16.0.1", + "172.31.255.255", + "192.168.1.1", + "169.254.169.254", + "169.254.0.1", + "0.0.0.0", + "100.64.0.1", + "255.255.255.255", + "224.0.0.1", + "240.0.0.1", + "::1", + "::", + "fc00::1", + "fd12::1", + "fe80::1", + "::ffff:127.0.0.1", + "::ffff:169.254.169.254", + "64:ff9b::7f00:1", + "ff02::1", + ] { + assert!(non_public_reason(ip(addr)).is_some(), "{addr} should be refused"); + } + } + + #[test] + fn allows_public_addresses() { + for addr in [ + "1.1.1.1", + "8.8.8.8", + "93.184.216.34", + "172.32.0.1", + "2606:4700:4700::1111", + ] { + assert!(non_public_reason(ip(addr)).is_none(), "{addr} should be allowed"); + } + } + + #[test] + fn private_networks_can_be_opted_in() { + let policy = DestinationPolicy::new(true, &[], &[]); + assert!(policy.check_url("http://127.0.0.1/").is_ok()); + assert!(policy.check_url("http://169.254.169.254/").is_ok()); + let policy = DestinationPolicy::new(false, &[], &[]); + assert!(policy.check_url("http://127.0.0.1/").is_err()); + assert!(policy.check_url("http://[::1]/").is_err()); + assert!(policy.check_url("http://169.254.169.254/latest/meta-data").is_err()); + } + + #[test] + fn host_lists() { + let policy = DestinationPolicy::new( + false, + &["*.example.com".into(), "api.test".into()], + &["bad.example.com".into()], + ); + assert!(policy.check_url("https://example.com/").is_ok()); + assert!(policy.check_url("https://a.b.example.com/").is_ok()); + assert!(policy.check_url("https://api.test/").is_ok()); + assert!(policy.check_url("https://API.TEST./").is_ok()); + assert!(policy.check_url("https://bad.example.com/").is_err(), "deny wins over allow"); + assert!(policy.check_url("https://notexample.com/").is_err()); + assert!(policy.check_url("https://httpbin.org/").is_err(), "not on the allow list"); + } + + #[test] + fn only_http_schemes() { + let policy = DestinationPolicy::new(true, &[], &[]); + assert!(policy.check_url("ftp://example.com/").is_err()); + assert!(policy.check_url("file:///etc/passwd").is_err()); + assert!(policy.check_url("https://example.com/").is_ok()); + } +} diff --git a/crates-server/yaak-send-proxy/src/limits.rs b/crates-server/yaak-send-proxy/src/limits.rs new file mode 100644 index 00000000..4714001c --- /dev/null +++ b/crates-server/yaak-send-proxy/src/limits.rs @@ -0,0 +1,90 @@ +//! Per-client rate limiting, kept deliberately small. +//! +//! One token bucket per client IP, refilled continuously, in a mutex-guarded +//! map that is swept of idle entries as it goes. Good enough to keep one +//! caller from monopolising a hosted instance; not a substitute for whatever +//! sits in front of it in production. + +use std::collections::HashMap; +use std::net::IpAddr; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +pub struct RateLimiter { + per_minute: u32, + buckets: Mutex>, +} + +struct Bucket { + tokens: f64, + last: Instant, +} + +impl RateLimiter { + /// `per_minute == 0` disables limiting. + pub fn new(per_minute: u32) -> Self { + Self { per_minute, buckets: Mutex::new(HashMap::new()) } + } + + /// Take one token for `client`, or say how long until one is available. + pub fn check(&self, client: IpAddr) -> Result<(), Duration> { + if self.per_minute == 0 { + return Ok(()); + } + let capacity = self.per_minute as f64; + let per_second = capacity / 60.0; + let now = Instant::now(); + + let mut buckets = self.buckets.lock().unwrap_or_else(|e| e.into_inner()); + + // Sweep buckets that have been idle long enough to be full again; there is nothing + // to remember about them. + if buckets.len() > 1024 { + buckets.retain(|_, b| now.duration_since(b.last).as_secs_f64() * per_second < capacity); + } + + let bucket = buckets.entry(client).or_insert(Bucket { tokens: capacity, last: now }); + let elapsed = now.duration_since(bucket.last).as_secs_f64(); + bucket.tokens = (bucket.tokens + elapsed * per_second).min(capacity); + bucket.last = now; + + if bucket.tokens >= 1.0 { + bucket.tokens -= 1.0; + Ok(()) + } else { + let wait = (1.0 - bucket.tokens) / per_second; + Err(Duration::from_secs_f64(wait.max(0.001))) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_full_bucket_then_a_wait() { + let limiter = RateLimiter::new(3); + let ip: IpAddr = "203.0.113.5".parse().unwrap(); + assert!(limiter.check(ip).is_ok()); + assert!(limiter.check(ip).is_ok()); + assert!(limiter.check(ip).is_ok()); + let wait = limiter.check(ip).expect_err("fourth call in a burst should wait"); + assert!(wait > Duration::ZERO && wait <= Duration::from_secs(20)); + } + + #[test] + fn clients_are_independent_and_zero_disables() { + let limiter = RateLimiter::new(1); + let a: IpAddr = "203.0.113.5".parse().unwrap(); + let b: IpAddr = "203.0.113.6".parse().unwrap(); + assert!(limiter.check(a).is_ok()); + assert!(limiter.check(a).is_err()); + assert!(limiter.check(b).is_ok()); + + let unlimited = RateLimiter::new(0); + for _ in 0..1000 { + assert!(unlimited.check(a).is_ok()); + } + } +} diff --git a/crates-server/yaak-send-proxy/src/main.rs b/crates-server/yaak-send-proxy/src/main.rs new file mode 100644 index 00000000..9b9f7dcc --- /dev/null +++ b/crates-server/yaak-send-proxy/src/main.rs @@ -0,0 +1,226 @@ +//! yaak-send-proxy: the network half of Yaak in a browser. +//! +//! A tab can't see a response the way a desktop app can — CORS hides most +//! headers, redirects are followed silently, there is no timeline. So the tab +//! renders the request and hands it here; this process puts it on the network +//! with the desktop's own engine and streams back everything that happened, +//! for the tab to store. It keeps nothing: no database, no files, no session. +//! +//! One binary, configured by flags or `YAAK_PROXY_*` environment variables. +//! See README.md for running and self-hosting it, and `guard.rs` for what it +//! refuses to talk to. + +mod config; +mod guard; +mod limits; +mod send; +mod wire; + +use axum::Router; +use axum::body::Body; +use axum::extract::{ConnectInfo, DefaultBodyLimit, State}; +use axum::http::{HeaderMap, HeaderValue, Method, StatusCode, header}; +use axum::response::{IntoResponse, Json, Response}; +use axum::routing::{get, post}; +use clap::Parser; +use config::Config; +use guard::DestinationPolicy; +use limits::RateLimiter; +use log::{info, warn}; +use send::{Refusal, SendLimits}; +use serde_json::json; +use std::net::{IpAddr, SocketAddr}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::Semaphore; +use tower_http::cors::{AllowOrigin, CorsLayer}; +use wire::SendRequest; + +#[derive(Clone)] +struct AppState { + config: Arc, + limits: Arc, + rate_limiter: Arc, + in_flight: Arc, +} + +#[tokio::main] +async fn main() { + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).init(); + let config = Config::parse(); + + let policy = DestinationPolicy::new( + config.allow_private_networks, + &config.allow_hosts, + &config.deny_hosts, + ); + let state = AppState { + limits: Arc::new(SendLimits { + policy, + max_response_bytes: config.max_response_bytes, + max_timeout: Duration::from_secs(config.max_timeout_secs), + }), + rate_limiter: Arc::new(RateLimiter::new(config.rate_limit_per_minute)), + in_flight: Arc::new(Semaphore::new(config.max_concurrent)), + config: Arc::new(config), + }; + + let cors = CorsLayer::new() + .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) + .allow_headers([header::CONTENT_TYPE, header::AUTHORIZATION]) + .allow_origin(allowed_origins(&state.config.allowed_origins)); + + let app = Router::new() + .route("/", get(root)) + .route("/v1/health", get(health)) + // A WebSocket or gRPC relay would sit beside this as `/v1/ws/relay` and `/v1/grpc/relay` + // on the same router, behind the same policy, limits and auth. Not built; see README. + .route("/v1/http/send", post(send_http)) + .layer(DefaultBodyLimit::max(state.config.max_request_bytes)) + .layer(cors) + .with_state(state.clone()); + + let bind = state.config.bind; + let listener = tokio::net::TcpListener::bind(bind).await.unwrap_or_else(|e| { + eprintln!("Failed to bind {bind}: {e}"); + std::process::exit(1); + }); + info!( + "yaak-send-proxy listening on http://{bind} (private networks: {}, token: {}, rate limit: {}/min)", + if state.config.allow_private_networks { "allowed" } else { "refused" }, + if state.config.token.is_some() { "required" } else { "none" }, + state.config.rate_limit_per_minute, + ); + + axum::serve(listener, app.into_make_service_with_connect_info::()) + .with_graceful_shutdown(async { + let _ = tokio::signal::ctrl_c().await; + info!("Shutting down"); + }) + .await + .expect("server error"); +} + +fn allowed_origins(origins: &[String]) -> AllowOrigin { + if origins.iter().any(|o| o.trim() == "*") { + return AllowOrigin::any(); + } + let parsed: Vec = + origins.iter().filter_map(|o| HeaderValue::from_str(o.trim()).ok()).collect(); + AllowOrigin::list(parsed) +} + +async fn root() -> impl IntoResponse { + "yaak-send-proxy: POST /v1/http/send (see https://github.com/mountain-loop/yaak)\n" +} + +async fn health(State(state): State) -> impl IntoResponse { + Json(json!({ + "ok": true, + "version": env!("CARGO_PKG_VERSION"), + "privateNetworks": state.config.allow_private_networks, + "tokenRequired": state.config.token.is_some(), + "maxResponseBytes": state.config.max_response_bytes, + "maxTimeoutSecs": state.config.max_timeout_secs, + })) +} + +fn error_response(status: StatusCode, message: impl Into) -> Response { + let message = message.into(); + (status, Json(json!({ "error": message }))).into_response() +} + +/// The client's address for rate limiting: the socket peer, or the first `X-Forwarded-For` +/// hop when the operator has said the header can be trusted. +fn client_ip(config: &Config, headers: &HeaderMap, peer: SocketAddr) -> IpAddr { + if config.trust_forwarded_for + && let Some(forwarded) = headers.get("x-forwarded-for").and_then(|v| v.to_str().ok()) + && let Some(first) = forwarded.split(',').next() + && let Ok(ip) = first.trim().parse::() + { + return ip; + } + peer.ip() +} + +fn authorized(config: &Config, headers: &HeaderMap) -> bool { + let Some(expected) = config.token.as_deref() else { + return true; + }; + headers + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .is_some_and(|got| constant_time_eq(got.as_bytes(), expected.as_bytes())) +} + +fn constant_time_eq(a: &[u8], b: &[u8]) -> bool { + if a.len() != b.len() { + return false; + } + a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 +} + +async fn send_http( + State(state): State, + ConnectInfo(peer): ConnectInfo, + headers: HeaderMap, + Json(body): Json, +) -> Response { + if !authorized(&state.config, &headers) { + return error_response(StatusCode::UNAUTHORIZED, "This proxy requires a token"); + } + + let ip = client_ip(&state.config, &headers, peer); + if let Err(wait) = state.rate_limiter.check(ip) { + warn!("Rate limited {ip}"); + let mut res = error_response( + StatusCode::TOO_MANY_REQUESTS, + format!("Rate limit reached; try again in {}s", wait.as_secs().max(1)), + ); + res.headers_mut().insert(header::RETRY_AFTER, HeaderValue::from(wait.as_secs().max(1))); + return res; + } + + let Ok(permit) = state.in_flight.clone().try_acquire_owned() else { + warn!("At capacity; refusing {ip}"); + return error_response(StatusCode::SERVICE_UNAVAILABLE, "This proxy is at capacity"); + }; + + let prepared = match send::prepare(state.limits.clone(), body).await { + Ok(p) => p, + Err(Refusal::Unsupported(m)) => return error_response(StatusCode::BAD_REQUEST, m), + Err(Refusal::Invalid(m)) => return error_response(StatusCode::BAD_REQUEST, m), + Err(Refusal::Destination(m)) => { + warn!("Refused send from {ip}: {m}"); + return error_response(StatusCode::FORBIDDEN, m); + } + }; + + let description = prepared.describe(); + info!("{ip} -> {description}"); + let started = Instant::now(); + + let (tx, rx) = tokio::sync::mpsc::channel(send::FRAME_CHANNEL_CAPACITY); + tokio::spawn(async move { + prepared.run(tx).await; + send::log_outcome(&description, started, "finished"); + drop(permit); + }); + + let stream = tokio_stream_from(rx); + Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/x-ndjson") + .header(header::CACHE_CONTROL, "no-store") + // Some reverse proxies buffer streamed responses unless told not to + .header("x-accel-buffering", "no") + .body(Body::from_stream(stream)) + .expect("valid response") +} + +fn tokio_stream_from( + mut rx: tokio::sync::mpsc::Receiver, +) -> impl futures_util::Stream + Send + 'static { + futures_util::stream::poll_fn(move |cx| rx.poll_recv(cx)) +} diff --git a/crates-server/yaak-send-proxy/src/send.rs b/crates-server/yaak-send-proxy/src/send.rs new file mode 100644 index 00000000..c8c08ab0 --- /dev/null +++ b/crates-server/yaak-send-proxy/src/send.rs @@ -0,0 +1,356 @@ +//! The one thing this binary does: execute a rendered request and stream back what happened. +//! +//! This is the "execute" half of the desktop's `send_http_request` — the part after rendering +//! and before storage — driven through the same `HttpTransaction` the desktop drives, with the +//! same redirect loop, cookie jar, decompression and timeline events. Everything the desktop +//! would write to its database is written to the reply stream instead, and the tab stores it. + +use crate::guard::{DestinationPolicy, GuardedSender}; +use crate::wire::{Frame, SendRequest, WireHeader}; +use base64::Engine; +use bytes::Bytes; +use log::{info, warn}; +use std::convert::Infallible; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; +use tokio::io::AsyncReadExt; +use tokio::sync::{mpsc, watch}; +use yaak_http::client::{HttpConnectionOptions, HttpConnectionProxySetting}; +use yaak_http::cookies::CookieStore; +use yaak_http::sender::{HttpResponseEvent, ReqwestSender}; +use yaak_http::transaction::HttpTransaction; +use yaak_http::types::{SendableHttpRequest, SendableHttpRequestOptions}; + +/// How many frames may sit unread by the client before body reading pauses. Backpressure, so a +/// slow tab slows the upstream read rather than filling memory. +pub const FRAME_CHANNEL_CAPACITY: usize = 64; +const EVENT_CHANNEL_CAPACITY: usize = 256; +const BODY_READ_CHUNK: usize = 64 * 1024; + +/// What a send needs from the process, beyond the request itself. +pub struct SendLimits { + pub policy: DestinationPolicy, + pub max_response_bytes: usize, + pub max_timeout: Duration, +} + +/// Why a send was refused before anything was put on the network. Distinct from a failure +/// mid-stream: these become a plain HTTP error, not a stream with an error frame. +#[derive(Debug)] +pub enum Refusal { + /// The request asks for something a browser-originated send cannot mean. + Unsupported(String), + /// The destination is not one this proxy will talk to. + Destination(String), + /// The request could not be turned into something sendable. + Invalid(String), +} + +pub type FrameSender = mpsc::Sender>; + +/// Check and prepare a send, then hand back the task that runs it. Refusals happen here, before +/// the caller has committed to a streaming response. +pub async fn prepare(limits: Arc, send: SendRequest) -> Result { + let request = send.request; + + // The engine reads files for these body types. There are no files here that a browser tab + // could legitimately mean, and letting a request name a path on this machine would be a + // local file read for anyone who can reach the proxy. + if request.body_type.as_deref() == Some("binary") { + return Err(Refusal::Unsupported( + "Binary file bodies can't be sent from the browser: the proxy has no access to your files" + .to_string(), + )); + } + if request.body_type.as_deref() == Some("multipart/form-data") { + let names_a_file = + request.body.get("form").and_then(|f| f.as_array()).is_some_and(|entries| { + entries.iter().any(|e| { + e.get("enabled").and_then(|v| v.as_bool()).unwrap_or(true) + && e.get("file").and_then(|v| v.as_str()).is_some_and(|f| !f.is_empty()) + }) + }); + if names_a_file { + return Err(Refusal::Unsupported( + "Multipart file fields can't be sent from the browser: the proxy has no access to your files" + .to_string(), + )); + } + } + + // The tab's requested timeout, capped. Zero means "none", which here means the cap. + let requested = if send.settings.timeout_ms > 0 { + Some(Duration::from_millis(send.settings.timeout_ms as u64)) + } else { + None + }; + let timeout = requested.map_or(limits.max_timeout, |t| t.min(limits.max_timeout)); + let timeout_capped = requested.is_none_or(|t| t > limits.max_timeout); + + let sendable = SendableHttpRequest::from_http_request( + &request, + SendableHttpRequestOptions { + timeout: Some(timeout), + follow_redirects: send.settings.follow_redirects, + }, + ) + .await + .map_err(|e| Refusal::Invalid(e.to_string()))?; + + // The first hop, checked up front so a bad destination is a clean refusal rather than a + // stream that opens and immediately errors. Every later hop is checked by GuardedSender. + limits.policy.check_url(&sendable.url).map_err(Refusal::Destination)?; + + Ok(PreparedSend { + limits, + sendable, + settings: send.settings, + cookies: send.cookies, + timeout, + timeout_capped, + }) +} + +pub struct PreparedSend { + limits: Arc, + sendable: SendableHttpRequest, + settings: crate::wire::SendSettings, + cookies: Option>, + timeout: Duration, + timeout_capped: bool, +} + +impl PreparedSend { + pub fn describe(&self) -> String { + format!("{} {}", self.sendable.method, self.sendable.url) + } + + /// Run the send, writing frames to `frames` until the terminal frame. Returns when the + /// stream is complete or the client has gone away. + pub async fn run(mut self, frames: FrameSender) { + let cookie_store = self.cookies.take().map(CookieStore::from_cookies); + let store_for_result = cookie_store.clone(); + let outcome = self.execute(frames.clone(), cookie_store).await; + + let cookies = store_for_result.as_ref().map(|s| s.get_all_cookies()); + let terminal = match outcome { + Ok(done) => Frame::Done { + elapsed: done.elapsed, + content_length: done.content_length, + content_length_compressed: done.content_length_compressed, + cookies, + }, + Err(message) => Frame::Error { message, cookies }, + }; + let _ = write_frame(&frames, &terminal).await; + } + + async fn execute( + self, + frames: FrameSender, + cookie_store: Option, + ) -> Result { + let limits = self.limits; + + let (client, resolver) = HttpConnectionOptions { + id: uuid::Uuid::new_v4().to_string(), + validate_certificates: self.settings.validate_certificates, + // The proxy connects directly. Going through a system proxy would move DNS, and + // therefore the address check, somewhere this process can't see. + proxy: HttpConnectionProxySetting::Disabled, + client_certificate: None, + dns_overrides: Vec::new(), + address_filter: Some(limits.policy.address_filter()), + } + .build_client() + .map_err(|e| format!("Failed to build HTTP client: {e}"))?; + + // Timeline events go into the same frame stream as everything else, as they happen. + // The desktop persists them from a task like this one; here the task serialises them. + let (event_tx, mut event_rx) = mpsc::channel::(EVENT_CHANNEL_CAPACITY); + resolver.set_event_sender(Some(event_tx.clone())).await; + let dns_elapsed = Arc::new(AtomicU64::new(0)); + let event_frames = frames.clone(); + let event_dns = dns_elapsed.clone(); + let event_task = tokio::spawn(async move { + while let Some(event) = event_rx.recv().await { + if let HttpResponseEvent::DnsResolved { duration, .. } = &event { + event_dns.store(*duration, Ordering::Relaxed); + } + let frame = Frame::Event { event: event.into() }; + if write_frame(&event_frames, &frame).await.is_err() { + break; + } + } + }); + + // Cancellation: the client hanging up, or the overall deadline. The deadline exists + // because a per-hop timeout times each hop separately; ten slow redirects must not add + // up to ten timeouts. + let (cancel_tx, cancel_rx) = watch::channel(false); + let deadline = self.timeout * 2 + Duration::from_secs(5); + let deadline_cancel = cancel_tx.clone(); + let deadline_task = tokio::spawn(async move { + tokio::time::sleep(deadline).await; + let _ = deadline_cancel.send(true); + }); + let hangup_frames = frames.clone(); + let hangup_task = tokio::spawn(async move { + hangup_frames.closed().await; + let _ = cancel_tx.send(true); + }); + + if self.timeout_capped { + let _ = event_tx.try_send(HttpResponseEvent::Info(format!( + "Timeout set to {:?} (this proxy's ceiling)", + self.timeout + ))); + } + + let sender = GuardedSender::new(ReqwestSender::with_client(client), limits.policy.clone()); + let transaction = match cookie_store { + Some(store) => HttpTransaction::with_cookie_behavior( + sender, + store, + self.settings.send_cookies, + self.settings.store_cookies, + ), + None => HttpTransaction::new(sender), + }; + + let started_at = Instant::now(); + let result = transaction + .execute_with_cancellation(self.sendable, cancel_rx.clone(), event_tx.clone()) + .await; + resolver.set_event_sender(None).await; + + let mut response = match result { + Ok(response) => response, + Err(err) => { + drop(event_tx); + let _ = event_task.await; + deadline_task.abort(); + hangup_task.abort(); + return Err(describe_error(&err)); + } + }; + + let elapsed_headers = started_at.elapsed().as_millis() as u64; + let head = Frame::Response { + status: response.status, + status_reason: response.status_reason.clone(), + url: response.url.clone(), + remote_addr: response.remote_addr.clone(), + version: response.version.clone(), + headers: to_wire_headers(&response.headers), + request_headers: to_wire_headers(&response.request_headers), + content_length: response.content_length, + elapsed_headers, + elapsed_dns: dns_elapsed.load(Ordering::Relaxed), + }; + write_frame(&frames, &head).await.map_err(|_| "Client went away".to_string())?; + + let declared_length = response.content_length; + let mut body = response + .into_body_stream() + .map_err(|e| format!("Failed to read response body: {e}"))?; + let mut buf = vec![0u8; BODY_READ_CHUNK]; + let mut total: usize = 0; + let mut cancel_rx = cancel_rx; + let base64 = base64::engine::general_purpose::STANDARD; + + let read_result: Result<(), String> = loop { + if *cancel_rx.borrow() { + break Err("Request canceled".to_string()); + } + let read = tokio::select! { + biased; + _ = cancel_rx.changed() => break Err("Request canceled".to_string()), + r = body.read(&mut buf) => r, + }; + match read { + Ok(0) => break Ok(()), + Ok(n) => { + total += n; + if total > limits.max_response_bytes { + break Err(format!( + "Response body exceeds this proxy's limit of {} bytes", + limits.max_response_bytes + )); + } + let frame = Frame::Body { data: base64.encode(&buf[..n]) }; + if write_frame(&frames, &frame).await.is_err() { + break Err("Client went away".to_string()); + } + } + Err(e) => break Err(format!("Failed to read response body: {e}")), + } + }; + drop(body); + + // Let the timeline drain before the terminal frame, so nothing arrives after "done". + drop(event_tx); + let _ = event_task.await; + deadline_task.abort(); + hangup_task.abort(); + + read_result?; + Ok(DoneStats { + elapsed: started_at.elapsed().as_millis() as u64, + content_length: total as u64, + content_length_compressed: declared_length.unwrap_or(total as u64), + }) + } +} + +/// A send error as a sentence, not a debug dump. +/// +/// A connection error from reqwest arrives wrapped several layers deep, and the layer that +/// says something useful — "Refusing to connect to ::1: loopback" — is the innermost. The +/// desktop shows the outer `Debug`; a stranger reading a proxy's reply deserves the reason. +fn describe_error(err: &yaak_http::error::Error) -> String { + match err { + yaak_http::error::Error::Client(e) => { + let mut leaf: &dyn std::error::Error = e; + while let Some(next) = leaf.source() { + leaf = next; + } + let outer = e.to_string(); + let inner = leaf.to_string(); + if inner == outer { outer } else { format!("{outer}: {inner}") } + } + yaak_http::error::Error::RequestError(message) => format!("Request failed: {message}"), + other => other.to_string(), + } +} + +struct DoneStats { + elapsed: u64, + content_length: u64, + content_length_compressed: u64, +} + +fn to_wire_headers(headers: &[(String, String)]) -> Vec { + headers + .iter() + .map(|(name, value)| WireHeader { name: name.clone(), value: value.clone() }) + .collect() +} + +async fn write_frame(frames: &FrameSender, frame: &Frame) -> Result<(), ()> { + let mut line = match serde_json::to_vec(frame) { + Ok(v) => v, + Err(e) => { + warn!("Failed to serialize frame: {e}"); + return Err(()); + } + }; + line.push(b'\n'); + frames.send(Ok(Bytes::from(line))).await.map_err(|_| ()) +} + +/// Log a finished send at info: destination, outcome, and how long, never the content. +pub fn log_outcome(description: &str, started: Instant, outcome: &str) { + info!("{description} -> {outcome} in {:?}", started.elapsed()); +} diff --git a/crates-server/yaak-send-proxy/src/wire.rs b/crates-server/yaak-send-proxy/src/wire.rs new file mode 100644 index 00000000..6751349e --- /dev/null +++ b/crates-server/yaak-send-proxy/src/wire.rs @@ -0,0 +1,100 @@ +//! What crosses the wire between a tab and this proxy. +//! +//! One `POST /v1/http/send` carries a request the tab has already rendered — +//! templates resolved, inheritance applied — plus the send settings and the +//! cookies the send starts with. The reply is a stream of newline-delimited +//! JSON frames: timeline events as they happen, the response head as soon as +//! headers arrive, body chunks as they are read, and one terminal frame. +//! +//! Nothing here names a workspace, a request id, or a response id. The proxy +//! does not know what the tab will call this response; it only knows what came +//! back. + +use serde::{Deserialize, Serialize}; +use yaak_models::models::{Cookie, HttpRequest, HttpResponseEventData}; + +/// The body of `POST /v1/http/send`. +#[derive(Deserialize, Debug)] +#[serde(rename_all = "camelCase")] +pub struct SendRequest { + /// The request to send, in the desktop's own model shape but with every template already + /// rendered by the tab. The proxy builds the URL, headers and body from it exactly the way + /// the desktop does after rendering. + pub request: HttpRequest, + pub settings: SendSettings, + /// The cookies to start with. `None` means no jar at all: nothing sent, nothing kept. + #[serde(default)] + pub cookies: Option>, +} + +/// The resolved send settings, values only. Where they came from (request, folder, workspace) +/// is the tab's to record in its timeline; the proxy only needs to obey them. +#[derive(Deserialize, Debug, Clone)] +#[serde(rename_all = "camelCase")] +pub struct SendSettings { + pub validate_certificates: bool, + pub follow_redirects: bool, + /// Milliseconds. Zero or negative means "no timeout", which the proxy caps regardless. + pub timeout_ms: i64, + pub send_cookies: bool, + pub store_cookies: bool, +} + +#[derive(Serialize, Debug)] +#[serde(rename_all = "camelCase")] +pub struct WireHeader { + pub name: String, + pub value: String, +} + +/// One line of the reply stream. Tags are snake_case like the timeline event tags; fields are +/// camelCase like every model the tab stores. +#[derive(Serialize, Debug)] +#[serde( + tag = "type", + rename_all = "snake_case", + rename_all_fields = "camelCase" +)] +pub enum Frame { + /// A timeline event, in the same shape the desktop stores. Interleaved with everything + /// else in the order the engine produced it. + Event { event: HttpResponseEventData }, + /// The response head. Sent once, as soon as the final hop's headers are in — before any of + /// the body — so the tab can show status and headers while the body streams. + Response { + status: u16, + status_reason: Option, + /// The URL that answered, after redirects. + url: String, + remote_addr: Option, + version: Option, + headers: Vec, + /// The headers that were actually sent on the final hop, cookies and all. + request_headers: Vec, + /// `Content-Length` as declared by the server, if it declared one. + content_length: Option, + /// Milliseconds from the start of the send to the response head. + elapsed_headers: u64, + /// Milliseconds spent in DNS on the last lookup, or zero. + elapsed_dns: u64, + }, + /// A piece of the response body, decompressed, base64-encoded. + Body { data: String }, + /// The send finished. The last frame on a successful stream. + Done { + /// Milliseconds from the start of the send to the end of the body. + elapsed: u64, + /// Bytes of body relayed, after decompression. + content_length: u64, + /// Bytes on the wire as declared by the server, or the relayed size when unknown. + content_length_compressed: u64, + /// The jar as the send left it, for the tab to persist. `None` when the tab sent none. + cookies: Option>, + }, + /// The send failed. The last frame on a failed stream. Cookies collected before the failure + /// still come back — the transaction may have set some before the hop that failed. + Error { + message: String, + cookies: Option>, + }, +} diff --git a/crates/yaak-http/Cargo.toml b/crates/yaak-http/Cargo.toml index 0da64313..08d9f032 100644 --- a/crates/yaak-http/Cargo.toml +++ b/crates/yaak-http/Cargo.toml @@ -19,7 +19,6 @@ hyper-util = { version = "0.1.17", default-features = false, features = ["client log = { workspace = true } mime_guess = "2.0.5" native-tls = { version = "0.2", features = ["alpn"] } -regex = "1.11.1" reqwest = { workspace = true, features = [ "rustls-tls-manual-roots-no-provider", "native-tls", diff --git a/crates/yaak-http/src/client.rs b/crates/yaak-http/src/client.rs index 71c69ce8..139357f7 100644 --- a/crates/yaak-http/src/client.rs +++ b/crates/yaak-http/src/client.rs @@ -1,4 +1,4 @@ -use crate::dns::LocalhostResolver; +use crate::dns::{AddressFilter, LocalhostResolver}; use crate::error::Result; use log::{debug, info, warn}; use reqwest::{Client, ClientBuilder, Proxy, redirect}; @@ -103,13 +103,18 @@ pub struct HttpConnectionOptions { pub proxy: HttpConnectionProxySetting, pub client_certificate: Option, pub dns_overrides: Vec, + /// Refuse connections to addresses a hostname resolves to. `None` means + /// every resolved address is connectable, which is what the desktop wants: + /// a user sending to their own machine or their own network is the point. + /// A hosted sender is the caller that supplies one. + pub address_filter: Option, } impl HttpConnectionOptions { /// Build a reqwest Client and return it along with the DNS resolver. /// The resolver is returned separately so it can be configured per-request /// to emit DNS timing events to the appropriate channel. - pub(crate) fn build_client(&self) -> Result<(ConfiguredClient, Arc)> { + pub fn build_client(&self) -> Result<(ConfiguredClient, Arc)> { let mut client = client_builder() .connection_verbose(true) .redirect(redirect::Policy::none()) @@ -135,7 +140,10 @@ impl HttpConnectionOptions { } // Configure DNS resolver - keep a reference to configure per-request - let resolver = LocalhostResolver::new(self.dns_overrides.clone()); + let resolver = LocalhostResolver::with_address_filter( + self.dns_overrides.clone(), + self.address_filter.clone(), + ); client = client.dns_resolver(resolver.clone()); // Configure proxy diff --git a/crates/yaak-http/src/dns.rs b/crates/yaak-http/src/dns.rs index 4747f73a..913e690e 100644 --- a/crates/yaak-http/src/dns.rs +++ b/crates/yaak-http/src/dns.rs @@ -20,15 +20,32 @@ pub struct ResolvedOverride { pub ipv6: Vec, } +/// A veto on the addresses a hostname resolves to, consulted after resolution +/// and before any connection is made. Returning `Err` refuses the whole lookup +/// with that message; a hostname is never partially allowed. +/// +/// A hosted sender uses this to refuse private and metadata ranges no matter +/// what name they hide behind. Checking here rather than on the URL is what +/// catches a public hostname that resolves to an internal address. +pub type AddressFilter = Arc std::result::Result<(), String> + Send + Sync>; + #[derive(Clone)] pub struct LocalhostResolver { fallback: HyperGaiResolver, event_tx: Arc>>>, overrides: Arc>, + address_filter: Option, } impl LocalhostResolver { pub fn new(dns_overrides: Vec) -> Arc { + Self::with_address_filter(dns_overrides, None) + } + + pub fn with_address_filter( + dns_overrides: Vec, + address_filter: Option, + ) -> Arc { let resolver = HyperGaiResolver::new(); // Pre-parse DNS overrides into a lookup map @@ -55,9 +72,25 @@ impl LocalhostResolver { fallback: resolver, event_tx: Arc::new(RwLock::new(None)), overrides: Arc::new(overrides), + address_filter, }) } + /// Apply the address filter, if any, to a resolved address list. + fn filter_addrs( + filter: &Option, + addrs: &[SocketAddr], + ) -> std::result::Result<(), Box> { + if let Some(filter) = filter { + for addr in addrs { + if let Err(reason) = filter(addr.ip()) { + return Err(Box::new(std::io::Error::other(reason))); + } + } + } + Ok(()) + } + /// Set the event sender for the current request. /// This should be called before each request to direct DNS events /// to the appropriate channel. @@ -72,6 +105,7 @@ impl Resolve for LocalhostResolver { let host = name.as_str().to_lowercase(); let event_tx = self.event_tx.clone(); let overrides = self.overrides.clone(); + let address_filter = self.address_filter.clone(); info!("DNS resolve called for: {}", host); @@ -94,6 +128,8 @@ impl Resolve for LocalhostResolver { let addresses: Vec = addrs.iter().map(|a| a.ip().to_string()).collect(); return Box::pin(async move { + Self::filter_addrs(&address_filter, &addrs)?; + // Emit DNS event for override let guard = event_tx.read().await; if let Some(tx) = guard.as_ref() { @@ -125,6 +161,8 @@ impl Resolve for LocalhostResolver { let addresses: Vec = addrs.iter().map(|a| a.ip().to_string()).collect(); return Box::pin(async move { + Self::filter_addrs(&address_filter, &addrs)?; + // Emit DNS event for localhost resolution let guard = event_tx.read().await; if let Some(tx) = guard.as_ref() { @@ -161,6 +199,7 @@ impl Resolve for LocalhostResolver { Ok(addrs) => { // Collect addresses for event emission let addr_vec: Vec = addrs.collect(); + Self::filter_addrs(&address_filter, &addr_vec)?; let addresses: Vec = addr_vec.iter().map(|a| a.ip().to_string()).collect(); diff --git a/crates/yaak-http/src/lib.rs b/crates/yaak-http/src/lib.rs index cee46d80..2a4155c9 100644 --- a/crates/yaak-http/src/lib.rs +++ b/crates/yaak-http/src/lib.rs @@ -5,9 +5,12 @@ pub mod decompress; pub mod dns; pub mod error; pub mod manager; -pub mod path_placeholders; mod proto; pub mod sender; pub mod tee_reader; pub mod transaction; pub mod types; + +// Moved to yaak-models so the browser's wasm host can render requests with the +// same code; re-exported here so existing callers keep their path. +pub use yaak_models::path_placeholders; diff --git a/crates/yaak-models/Cargo.toml b/crates/yaak-models/Cargo.toml index fe781836..ae318887 100644 --- a/crates/yaak-models/Cargo.toml +++ b/crates/yaak-models/Cargo.toml @@ -11,6 +11,7 @@ hex = { workspace = true } include_dir = "0.7" log = { workspace = true } nanoid = "0.4.0" +regex-lite = "0.1" rusqlite = { version = "0.38", features = ["bundled", "chrono"] } sea-query = { version = "1.0", features = ["with-chrono", "attr"] } sea-query-rusqlite = { version = "0.8.0", features = ["with-chrono"] } @@ -19,8 +20,10 @@ serde_json = { workspace = true } schemars = { workspace = true } sha2 = { workspace = true } thiserror = { workspace = true } +urlencoding = "2.1.3" ts-rs = { workspace = true, features = ["chrono-impl", "serde-json-impl"] } yaak-core = { workspace = true } +yaak-templates = { path = "../yaak-templates", default-features = false } [target.'cfg(not(target_arch = "wasm32"))'.dependencies] r2d2 = "0.8.10" diff --git a/crates/yaak-models/src/lib.rs b/crates/yaak-models/src/lib.rs index 97dfff80..1c95b28d 100644 --- a/crates/yaak-models/src/lib.rs +++ b/crates/yaak-models/src/lib.rs @@ -15,6 +15,7 @@ pub mod error; pub mod migrate; pub mod models; pub mod models_ops; +pub mod path_placeholders; pub mod queries; pub mod query_manager; pub mod render; diff --git a/crates/yaak-http/src/path_placeholders.rs b/crates/yaak-models/src/path_placeholders.rs similarity index 95% rename from crates/yaak-http/src/path_placeholders.rs rename to crates/yaak-models/src/path_placeholders.rs index 25aee789..8f39ad19 100644 --- a/crates/yaak-http/src/path_placeholders.rs +++ b/crates/yaak-models/src/path_placeholders.rs @@ -1,4 +1,4 @@ -use yaak_models::models::HttpUrlParameter; +use crate::models::HttpUrlParameter; pub fn apply_path_placeholders( url: &str, @@ -37,9 +37,9 @@ fn replace_path_placeholder(p: &HttpUrlParameter, url: &str) -> String { // A path placeholder is terminated by `/`, `?`, `#`, end-of-string, or a literal `:`. // The `:` boundary is what lets `/:id:increment-importance` substitute the `:id` // placeholder while leaving `:increment-importance` as literal text. - let re = regex::Regex::new(format!("(/){}([/?#:]|$)", p.name).as_str()).unwrap(); + let re = regex_lite::Regex::new(format!("(/){}([/?#:]|$)", p.name).as_str()).unwrap(); let result = re - .replace_all(url, |cap: ®ex::Captures| { + .replace_all(url, |cap: ®ex_lite::Captures| { format!( "{}{}{}", cap[1].to_string(), @@ -54,7 +54,7 @@ fn replace_path_placeholder(p: &HttpUrlParameter, url: &str) -> String { #[cfg(test)] mod placeholder_tests { use crate::path_placeholders::{apply_path_placeholders, replace_path_placeholder}; - use yaak_models::models::{HttpRequest, HttpUrlParameter}; + use crate::models::{HttpRequest, HttpUrlParameter}; #[test] fn placeholder_middle() { diff --git a/crates/yaak-models/src/render.rs b/crates/yaak-models/src/render.rs index 997f0061..c509788d 100644 --- a/crates/yaak-models/src/render.rs +++ b/crates/yaak-models/src/render.rs @@ -1,5 +1,99 @@ -use crate::models::{Environment, EnvironmentVariable}; -use std::collections::HashMap; +use crate::models::{ + Environment, EnvironmentVariable, HttpRequest, HttpRequestHeader, HttpUrlParameter, +}; +use crate::path_placeholders::apply_path_placeholders; +use log::info; +use serde_json::Value; +use std::collections::{BTreeMap, HashMap}; +use yaak_templates::{RenderOptions, TemplateCallback, parse_and_render, render_json_value_raw}; + +/// Render every template in a request against an environment chain. +/// +/// Lives here rather than beside the send engine so that the browser's wasm +/// host, which has the model layer but no sockets, renders exactly what the +/// desktop renders. +pub async fn render_http_request( + request: &HttpRequest, + environment_chain: Vec, + callback: &T, + options: &RenderOptions, +) -> yaak_templates::error::Result { + let vars = &make_vars_hashmap(environment_chain); + + let mut url_parameters = Vec::new(); + for parameter in request.url_parameters.clone() { + if !parameter.enabled { + continue; + } + + url_parameters.push(HttpUrlParameter { + enabled: parameter.enabled, + name: parse_and_render(parameter.name.as_str(), vars, callback, options).await?, + value: parse_and_render(parameter.value.as_str(), vars, callback, options).await?, + id: parameter.id, + }) + } + + let mut headers = Vec::new(); + for header in request.headers.clone() { + if !header.enabled { + continue; + } + + headers.push(HttpRequestHeader { + enabled: header.enabled, + name: parse_and_render(header.name.as_str(), vars, callback, options).await?, + value: parse_and_render(header.value.as_str(), vars, callback, options).await?, + id: header.id, + }) + } + + let mut body = BTreeMap::new(); + for (key, value) in request.body.clone() { + let value = if key == "form" { strip_disabled_form_entries(value) } else { value }; + body.insert(key, render_json_value_raw(value, vars, callback, options).await?); + } + + let authentication = { + let mut disabled = false; + let mut auth = BTreeMap::new(); + + match request.authentication.get("disabled") { + Some(Value::Bool(true)) => { + disabled = true; + } + Some(Value::String(template)) => { + disabled = parse_and_render(template.as_str(), vars, callback, options) + .await + .unwrap_or_default() + .is_empty(); + info!( + "Rendering authentication.disabled as a template: {disabled} from \"{template}\"" + ); + } + _ => {} + } + + if disabled { + auth.insert("disabled".to_string(), Value::Bool(true)); + } else { + for (key, value) in request.authentication.clone() { + if key == "disabled" { + auth.insert(key, Value::Bool(false)); + } else { + auth.insert(key, render_json_value_raw(value, vars, callback, options).await?); + } + } + } + + auth + }; + + let url = parse_and_render(request.url.clone().as_str(), vars, callback, options).await?; + let (url, url_parameters) = apply_path_placeholders(&url, &url_parameters); + + Ok(HttpRequest { url, url_parameters, headers, body, authentication, ..request.to_owned() }) +} pub fn make_vars_hashmap(environment_chain: Vec) -> HashMap { let mut variables = HashMap::new(); @@ -27,3 +121,70 @@ fn add_variable_to_map( map } + +fn strip_disabled_form_entries(v: Value) -> Value { + match v { + Value::Array(items) => Value::Array( + items + .into_iter() + .filter(|item| item.get("enabled").and_then(|e| e.as_bool()).unwrap_or(true)) + .collect(), + ), + v => v, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn test_strip_disabled_form_entries() { + let input = json!([ + {"enabled": true, "name": "foo", "value": "bar"}, + {"enabled": false, "name": "disabled", "value": "gone"}, + {"enabled": true, "name": "baz", "value": "qux"}, + ]); + let result = strip_disabled_form_entries(input); + assert_eq!( + result, + json!([ + {"enabled": true, "name": "foo", "value": "bar"}, + {"enabled": true, "name": "baz", "value": "qux"}, + ]) + ); + } + + #[test] + fn test_strip_disabled_form_entries_all_disabled() { + let input = json!([ + {"enabled": false, "name": "a", "value": "b"}, + {"enabled": false, "name": "c", "value": "d"}, + ]); + let result = strip_disabled_form_entries(input); + assert_eq!(result, json!([])); + } + + #[test] + fn test_strip_disabled_form_entries_missing_enabled_defaults_to_kept() { + let input = json!([ + {"name": "no_enabled_field", "value": "kept"}, + {"enabled": false, "name": "disabled", "value": "gone"}, + ]); + let result = strip_disabled_form_entries(input); + assert_eq!( + result, + json!([ + {"name": "no_enabled_field", "value": "kept"}, + ]) + ); + } + + #[test] + fn test_strip_disabled_form_entries_non_array_passthrough() { + let input = json!("just a string"); + let result = strip_disabled_form_entries(input.clone()); + assert_eq!(result, input); + } +} diff --git a/crates/yaak-templates/Cargo.toml b/crates/yaak-templates/Cargo.toml index f3fc2510..7cc2e1ab 100644 --- a/crates/yaak-templates/Cargo.toml +++ b/crates/yaak-templates/Cargo.toml @@ -10,6 +10,13 @@ wasm-opt = false # Causes errors in CI (haven't figured out why yet) [lib] crate-type = ["cdylib", "rlib"] +[features] +default = ["wasm"] +# The `#[wasm_bindgen]` exports (parse_template etc.) that make up the +# @yaakapp-internal/templates package. Off for crates that link this one into +# their own wasm module and do not want these re-exported from theirs. +wasm = [] + [dependencies] base64 = "0.22.1" serde = { workspace = true, features = ["derive"] } diff --git a/crates/yaak-templates/src/lib.rs b/crates/yaak-templates/src/lib.rs index 1fc5a6ea..d97dfb29 100644 --- a/crates/yaak-templates/src/lib.rs +++ b/crates/yaak-templates/src/lib.rs @@ -4,6 +4,7 @@ pub mod format_json; pub mod parser; pub mod renderer; pub mod strip_json_comments; +#[cfg(feature = "wasm")] pub mod wasm; pub use parser::*; diff --git a/crates/yaak-web/Cargo.toml b/crates/yaak-web/Cargo.toml index 02e83860..1ace7893 100644 --- a/crates/yaak-web/Cargo.toml +++ b/crates/yaak-web/Cargo.toml @@ -26,6 +26,8 @@ log = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } yaak-models = { workspace = true } +# No default features: the template exports belong to @yaakapp-internal/templates, not this module +yaak-templates = { path = "../yaak-templates", default-features = false } [target.'cfg(target_arch = "wasm32")'.dependencies] console_error_panic_hook = "0.1" diff --git a/crates/yaak-web/index.ts b/crates/yaak-web/index.ts index 7311891c..78694a95 100644 --- a/crates/yaak-web/index.ts +++ b/crates/yaak-web/index.ts @@ -3,4 +3,4 @@ // This is loaded by the SharedWorker in packages/platform/src/web/worker.ts and // nowhere else: it owns a SQLite database, and there must be exactly one of it // per origin. -export { blob_delete, blob_get, blob_put, boot, rpc } from "./pkg"; +export { blob_delete, blob_get, blob_put, boot, prepare_http_send, rpc } from "./pkg"; diff --git a/crates/yaak-web/pkg/yaak_web.d.ts b/crates/yaak-web/pkg/yaak_web.d.ts index 35416d74..42d31de3 100644 --- a/crates/yaak-web/pkg/yaak_web.d.ts +++ b/crates/yaak-web/pkg/yaak_web.d.ts @@ -25,6 +25,17 @@ export function blob_put(id: string, bytes: Uint8Array): void; */ export function boot(): Promise; +/** + * Resolve and render a request for sending, exactly as the desktop does before it puts the + * request on the network: the environment chain, inherited headers and auth, request + * settings, the cookie jar. Nothing here touches a socket. What comes back is what the tab + * posts to the send proxy. + * + * Refuses, with a message the user can act on, when the request needs something this host + * doesn't have: an authentication plugin, or a template function. + */ +export function prepare_http_send(payload: any): Promise; + /** * Run one command as `label` (the calling tab's identity, which stands in for * the desktop's window label on every write it makes). diff --git a/crates/yaak-web/pkg/yaak_web_bg.js b/crates/yaak-web/pkg/yaak_web_bg.js index c87d81b1..3d04e046 100644 --- a/crates/yaak-web/pkg/yaak_web_bg.js +++ b/crates/yaak-web/pkg/yaak_web_bg.js @@ -62,6 +62,22 @@ export function boot() { return ret; } +/** + * Resolve and render a request for sending, exactly as the desktop does before it puts the + * request on the network: the environment chain, inherited headers and auth, request + * settings, the cookie jar. Nothing here touches a socket. What comes back is what the tab + * posts to the send proxy. + * + * Refuses, with a message the user can act on, when the request needs something this host + * doesn't have: an authentication plugin, or a template function. + * @param {any} payload + * @returns {Promise} + */ +export function prepare_http_send(payload) { + const ret = wasm.prepare_http_send(payload); + return ret; +} + /** * Run one command as `label` (the calling tab's identity, which stands in for * the desktop's window label on every write it makes). @@ -496,7 +512,7 @@ export function __wbg_new_typed_c072c4ce9a2a0cdf(arg0, arg1) { const a = state0.a; state0.a = 0; try { - return wasm_bindgen__convert__closures_____invoke__h2c72ca09e851b7f3(a, state0.b, arg0, arg1); + return wasm_bindgen__convert__closures_____invoke__h2cf3f4cce3b29948(a, state0.b, arg0, arg1); } finally { state0.a = a; } @@ -678,23 +694,23 @@ export function __wbg_versions_215a3ab1c9d5745a(arg0) { return ret; } export function __wbindgen_cast_0000000000000001(arg0, arg1) { - // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [Externref], shim_idx: 1104, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`. - const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4); + // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [Externref], shim_idx: 1124, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`. + const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__ha1c2fa93df0107f3); return ret; } export function __wbindgen_cast_0000000000000002(arg0, arg1) { - // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("Event")], shim_idx: 202, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`. - const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h7e06bb925b918fbb); + // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("Event")], shim_idx: 214, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`. + const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__ha7903b6e296dd8f4); return ret; } export function __wbindgen_cast_0000000000000003(arg0, arg1) { - // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("IDBVersionChangeEvent")], shim_idx: 180, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`. - const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9); + // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [NamedExternref("IDBVersionChangeEvent")], shim_idx: 198, ret: Result(Unit), inner_ret: Some(Result(Unit)) }, mutable: true }) -> Externref`. + const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h999df771f7987dc3); return ret; } export function __wbindgen_cast_0000000000000004(arg0, arg1) { - // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [], shim_idx: 200, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`. - const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__h87640adb2bbfa2fc); + // Cast intrinsic for `Closure(Closure { owned: true, function: Function { arguments: [], shim_idx: 212, ret: Unit, inner_ret: Some(Unit) }, mutable: true }) -> Externref`. + const ret = makeMutClosure(arg0, arg1, wasm_bindgen__convert__closures_____invoke__ha1b480b83daa641f); return ret; } export function __wbindgen_cast_0000000000000005(arg0) { @@ -731,30 +747,30 @@ export function __wbindgen_init_externref_table() { table.set(offset + 2, true); table.set(offset + 3, false); } -function wasm_bindgen__convert__closures_____invoke__h87640adb2bbfa2fc(arg0, arg1) { - wasm.wasm_bindgen__convert__closures_____invoke__h87640adb2bbfa2fc(arg0, arg1); +function wasm_bindgen__convert__closures_____invoke__ha1b480b83daa641f(arg0, arg1) { + wasm.wasm_bindgen__convert__closures_____invoke__ha1b480b83daa641f(arg0, arg1); } -function wasm_bindgen__convert__closures_____invoke__h7e06bb925b918fbb(arg0, arg1, arg2) { - wasm.wasm_bindgen__convert__closures_____invoke__h7e06bb925b918fbb(arg0, arg1, arg2); +function wasm_bindgen__convert__closures_____invoke__ha7903b6e296dd8f4(arg0, arg1, arg2) { + wasm.wasm_bindgen__convert__closures_____invoke__ha7903b6e296dd8f4(arg0, arg1, arg2); } -function wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4(arg0, arg1, arg2) { - const ret = wasm.wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4(arg0, arg1, arg2); +function wasm_bindgen__convert__closures_____invoke__ha1c2fa93df0107f3(arg0, arg1, arg2) { + const ret = wasm.wasm_bindgen__convert__closures_____invoke__ha1c2fa93df0107f3(arg0, arg1, arg2); if (ret[1]) { throw takeFromExternrefTable0(ret[0]); } } -function wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9(arg0, arg1, arg2) { - const ret = wasm.wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9(arg0, arg1, arg2); +function wasm_bindgen__convert__closures_____invoke__h999df771f7987dc3(arg0, arg1, arg2) { + const ret = wasm.wasm_bindgen__convert__closures_____invoke__h999df771f7987dc3(arg0, arg1, arg2); if (ret[1]) { throw takeFromExternrefTable0(ret[0]); } } -function wasm_bindgen__convert__closures_____invoke__h2c72ca09e851b7f3(arg0, arg1, arg2, arg3) { - wasm.wasm_bindgen__convert__closures_____invoke__h2c72ca09e851b7f3(arg0, arg1, arg2, arg3); +function wasm_bindgen__convert__closures_____invoke__h2cf3f4cce3b29948(arg0, arg1, arg2, arg3) { + wasm.wasm_bindgen__convert__closures_____invoke__h2cf3f4cce3b29948(arg0, arg1, arg2, arg3); } diff --git a/crates/yaak-web/pkg/yaak_web_bg.wasm b/crates/yaak-web/pkg/yaak_web_bg.wasm index c14a70bc..685aa2c8 100644 Binary files a/crates/yaak-web/pkg/yaak_web_bg.wasm and b/crates/yaak-web/pkg/yaak_web_bg.wasm differ diff --git a/crates/yaak-web/pkg/yaak_web_bg.wasm.d.ts b/crates/yaak-web/pkg/yaak_web_bg.wasm.d.ts index 6a4067a4..d7cfef7c 100644 --- a/crates/yaak-web/pkg/yaak_web_bg.wasm.d.ts +++ b/crates/yaak-web/pkg/yaak_web_bg.wasm.d.ts @@ -5,6 +5,7 @@ export const blob_delete: (a: number, b: number) => [number, number]; export const blob_get: (a: number, b: number) => [number, number, number, number]; export const blob_put: (a: number, b: number, c: number, d: number) => [number, number]; export const boot: () => any; +export const prepare_http_send: (a: any) => any; export const rpc: (a: number, b: number, c: any, d: number, e: number) => [number, number, number]; export const rust_sqlite_wasm_abort: () => void; export const rust_sqlite_wasm_assert_fail: (a: number, b: number, c: number, d: number) => void; @@ -16,11 +17,11 @@ export const rust_sqlite_wasm_malloc: (a: number) => number; export const rust_sqlite_wasm_realloc: (a: number, b: number) => number; export const sqlite3_os_end: () => number; export const sqlite3_os_init: () => number; -export const wasm_bindgen__convert__closures_____invoke__hf84d53817e0238b4: (a: number, b: number, c: any) => [number, number]; -export const wasm_bindgen__convert__closures_____invoke__h7e53e249a4dc4aa9: (a: number, b: number, c: any) => [number, number]; -export const wasm_bindgen__convert__closures_____invoke__h2c72ca09e851b7f3: (a: number, b: number, c: any, d: any) => void; -export const wasm_bindgen__convert__closures_____invoke__h7e06bb925b918fbb: (a: number, b: number, c: any) => void; -export const wasm_bindgen__convert__closures_____invoke__h87640adb2bbfa2fc: (a: number, b: number) => void; +export const wasm_bindgen__convert__closures_____invoke__ha1c2fa93df0107f3: (a: number, b: number, c: any) => [number, number]; +export const wasm_bindgen__convert__closures_____invoke__h999df771f7987dc3: (a: number, b: number, c: any) => [number, number]; +export const wasm_bindgen__convert__closures_____invoke__h2cf3f4cce3b29948: (a: number, b: number, c: any, d: any) => void; +export const wasm_bindgen__convert__closures_____invoke__ha7903b6e296dd8f4: (a: number, b: number, c: any) => void; +export const wasm_bindgen__convert__closures_____invoke__ha1b480b83daa641f: (a: number, b: number) => void; export const __wbindgen_malloc: (a: number, b: number) => number; export const __wbindgen_realloc: (a: number, b: number, c: number, d: number) => number; export const __wbindgen_exn_store: (a: number) => void; diff --git a/crates/yaak-web/src/lib.rs b/crates/yaak-web/src/lib.rs index 2be014a5..aa2d7999 100644 --- a/crates/yaak-web/src/lib.rs +++ b/crates/yaak-web/src/lib.rs @@ -12,8 +12,10 @@ //! JavaScript side owns that; this crate assumes it is the only writer. //! //! The command surface is deliberately narrow: what the frontend needs to keep -//! its model store coherent, and blob storage. Sending, plugins, git, sync and -//! everything else with a socket or a filesystem behind it lives elsewhere. +//! its model store coherent, blob storage, and the "prepare" half of a send +//! (resolve, inherit, render — see [`prepare_http_send`]). Putting bytes on the +//! network, plugins, git, sync and everything else with a socket or a +//! filesystem behind it lives elsewhere. // Nothing in here means anything off wasm32, and building it there would drag // SQLite's wasm C shim into a native compile. So on any other target the crate @@ -24,12 +26,18 @@ use std::cell::RefCell; use std::sync::mpsc; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use wasm_bindgen::prelude::*; use yaak_models::blob_manager::{BlobManager, BodyChunk}; -use yaak_models::models::AnyModel; +use yaak_models::models::{ + AnyModel, CookieJar, HttpRequest, HttpResponseEvent, HttpResponseEventData, + ResolvedHttpRequestSettings, ResolvedSetting, +}; use yaak_models::models_ops; use yaak_models::query_manager::QueryManager; +use yaak_models::render::render_http_request; use yaak_models::util::{ModelPayload, UpdateSource}; +use yaak_templates::{RenderOptions, TemplateCallback}; /// Names inside the VFS, not paths on any disk. Two files because the desktop /// keeps two: models in one, blobs in the other. @@ -201,6 +209,20 @@ struct UpsertIntrospectionReq { content: Option, } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct ResponseIdReq { + response_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct InsertResponseEventsReq { + response_id: String, + workspace_id: String, + events: Vec, +} + fn dispatch( host: &Host, cmd: &str, @@ -305,6 +327,34 @@ fn dispatch( // Nothing here can open a socket, so no connection ever produced any. "models_grpc_events" | "models_websocket_events" => to_json(Vec::<()>::new()), + "web_get_http_request" => { + let req: RequestIdReq = from_js(payload)?; + to_json(host.queries.connect().get_http_request(&req.request_id).map_err(js_error)?) + } + + "cmd_get_http_response_events" => { + let req: ResponseIdReq = from_js(payload)?; + to_json( + host.queries + .connect() + .list_http_response_events(&req.response_id) + .map_err(js_error)?, + ) + } + + // The tab's half of the send timeline: the events the proxy streamed back, recorded + // under the response they belong to. Same rows the desktop's send task writes, and the + // writes fan out to every tab as `model_writes` like any other. + "web_insert_http_response_events" => { + let req: InsertResponseEventsReq = from_js(payload)?; + let db = host.queries.connect(); + for event in req.events { + let model = HttpResponseEvent::new(&req.response_id, &req.workspace_id, event); + db.upsert_http_response_event(&model, source).map_err(js_error)?; + } + to_json(()) + } + "cmd_get_workspace_meta" => { let req: WorkspaceIdReq = from_js(payload)?; let db = host.queries.connect(); @@ -316,6 +366,182 @@ fn dispatch( } } +/* -------------------------------------------------------------------------- */ +/* Preparing a send */ +/* -------------------------------------------------------------------------- */ + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +struct PrepareHttpSendReq { + request_id: String, + environment_id: Option, + cookie_jar_id: Option, +} + +/// The values the proxy needs to obey. Where each came from is in `setting_events`. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct PreparedSendSettings { + validate_certificates: bool, + follow_redirects: bool, + timeout_ms: i64, + send_cookies: bool, + store_cookies: bool, +} + +/// Everything a send needs that lives in the database, resolved and rendered: the desktop's +/// `HttpSendInputs`, in the shape a tab hands to the proxy and keeps for itself. +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct PreparedHttpSend { + /// The request with inherited headers and authentication applied and every template + /// rendered. What the proxy sends, and what the response records as its request. + request: HttpRequest, + settings: PreparedSendSettings, + /// The `* Setting name=value` timeline lines the desktop writes at the top of a send, + /// sources and all. The tab records them before the proxy's own events. + setting_events: Vec, + /// The jar the send starts with, so the tab can write it back with the proxy's changes. + cookie_jar: Option, +} + +/// A template callback for a host with no plugins. Variables render; a function is a clear +/// refusal naming the function, so the user knows what the request needs rather than seeing +/// an empty string sent in its place. +struct NoPluginsCallback; + +impl TemplateCallback for NoPluginsCallback { + fn run( + &self, + fn_name: &str, + _args: HashMap, + ) -> impl std::future::Future> + Send { + let message = format!( + "This request uses the template function \"{fn_name}\", which needs plugins. \ + Plugins aren't available in the browser yet" + ); + async move { Err(yaak_templates::error::Error::RenderError(message)) } + } + + fn transform_arg( + &self, + _fn_name: &str, + _arg_name: &str, + arg_value: &str, + ) -> yaak_templates::error::Result { + Ok(arg_value.to_string()) + } +} + +/// Resolve and render a request for sending, exactly as the desktop does before it puts the +/// request on the network: the environment chain, inherited headers and auth, request +/// settings, the cookie jar. Nothing here touches a socket. What comes back is what the tab +/// posts to the send proxy. +/// +/// Refuses, with a message the user can act on, when the request needs something this host +/// doesn't have: an authentication plugin, or a template function. +#[wasm_bindgen] +pub async fn prepare_http_send(payload: JsValue) -> Result { + let req: PrepareHttpSendReq = from_js(payload)?; + + // Everything from the database first, then release the host borrow before rendering. + let (request, environment_chain, settings, cookie_jar) = with_host(|host| { + let db = host.queries.connect(); + let request = db.get_http_request(&req.request_id).map_err(js_error)?; + let environment_chain = db + .resolve_environments( + &request.workspace_id, + request.folder_id.as_deref(), + req.environment_id.as_deref(), + ) + .map_err(js_error)?; + let (authentication_type, authentication, _auth_context_id) = + db.resolve_auth_for_http_request(&request).map_err(js_error)?; + let headers = db.resolve_headers_for_http_request(&request).map_err(js_error)?; + let settings = db.resolve_settings_for_http_request(&request).map_err(js_error)?; + let cookie_jar = match req.cookie_jar_id.as_deref() { + Some(id) => Some(db.get_cookie_jar(id).map_err(js_error)?), + None => None, + }; + let request = HttpRequest { authentication_type, authentication, headers, ..request }; + Ok((request, environment_chain, settings, cookie_jar)) + })?; + + let rendered = render_http_request( + &request, + environment_chain, + &NoPluginsCallback, + &RenderOptions::throw(), + ) + .await + .map_err(js_error)?; + + // Authentication is applied by a plugin on the desktop. There is no plugin here, and a + // request sent without the auth it asked for is worse than one refused with the reason. + let auth_disabled = + rendered.authentication.get("disabled").and_then(|v| v.as_bool()) == Some(true); + if let Some(auth_type) = rendered.authentication_type.as_deref() + && auth_type != "none" + && !auth_disabled + { + return Err(js_error(format!( + "This request uses {auth_type} authentication, which needs plugins. \ + Plugins aren't available in the browser yet" + ))); + } + + let setting_events = setting_events(&settings); + let prepared = PreparedHttpSend { + request: rendered, + settings: PreparedSendSettings { + validate_certificates: settings.validate_certificates.value, + follow_redirects: settings.follow_redirects.value, + timeout_ms: settings.request_timeout.value as i64, + send_cookies: settings.send_cookies.value, + store_cookies: settings.store_cookies.value, + }, + setting_events, + cookie_jar, + }; + // JSON-compatible, as `rpc` does: the tab posts this to the proxy with `JSON.stringify`, + // and the default serializer's `Map` for the request body would stringify to `{}`. + use serde::Serialize as _; + prepared.serialize(&serde_wasm_bindgen::Serializer::json_compatible()).map_err(js_error) +} + +/// The same five `Setting` lines `crates/yaak/src/send.rs` writes at the top of every send. +fn setting_events(settings: &ResolvedHttpRequestSettings) -> Vec { + fn event( + name: &str, + value: String, + setting: &ResolvedSetting, + ) -> HttpResponseEventData { + HttpResponseEventData::Setting { + name: name.to_string(), + value, + source_model: Some(setting.source_model.clone()), + source_id: setting.source_id.clone(), + source_name: setting.source_name.clone(), + } + } + let timeout = if settings.request_timeout.value > 0 { + format!("{}ms", settings.request_timeout.value) + } else { + "Infinity".to_string() + }; + vec![ + event( + "validate_certificates", + settings.validate_certificates.value.to_string(), + &settings.validate_certificates, + ), + event("redirects", settings.follow_redirects.value.to_string(), &settings.follow_redirects), + event("timeout", timeout, &settings.request_timeout), + event("send_cookies", settings.send_cookies.value.to_string(), &settings.send_cookies), + event("store_cookies", settings.store_cookies.value.to_string(), &settings.store_cookies), + ] +} + /* -------------------------------------------------------------------------- */ /* Blobs */ /* -------------------------------------------------------------------------- */ diff --git a/crates/yaak/src/render.rs b/crates/yaak/src/render.rs index 5523c9ce..5cecf215 100644 --- a/crates/yaak/src/render.rs +++ b/crates/yaak/src/render.rs @@ -1,95 +1,11 @@ use log::info; use serde_json::Value; use std::collections::BTreeMap; -use yaak_http::path_placeholders::apply_path_placeholders; -use yaak_models::models::{ - Environment, GrpcRequest, HttpRequest, HttpRequestHeader, HttpUrlParameter, -}; +use yaak_models::models::{Environment, GrpcRequest, HttpRequestHeader}; use yaak_models::render::make_vars_hashmap; use yaak_templates::{RenderOptions, TemplateCallback, parse_and_render, render_json_value_raw}; -pub async fn render_http_request( - request: &HttpRequest, - environment_chain: Vec, - callback: &T, - options: &RenderOptions, -) -> yaak_templates::error::Result { - let vars = &make_vars_hashmap(environment_chain); - - let mut url_parameters = Vec::new(); - for parameter in request.url_parameters.clone() { - if !parameter.enabled { - continue; - } - - url_parameters.push(HttpUrlParameter { - enabled: parameter.enabled, - name: parse_and_render(parameter.name.as_str(), vars, callback, options).await?, - value: parse_and_render(parameter.value.as_str(), vars, callback, options).await?, - id: parameter.id, - }) - } - - let mut headers = Vec::new(); - for header in request.headers.clone() { - if !header.enabled { - continue; - } - - headers.push(HttpRequestHeader { - enabled: header.enabled, - name: parse_and_render(header.name.as_str(), vars, callback, options).await?, - value: parse_and_render(header.value.as_str(), vars, callback, options).await?, - id: header.id, - }) - } - - let mut body = BTreeMap::new(); - for (key, value) in request.body.clone() { - let value = if key == "form" { strip_disabled_form_entries(value) } else { value }; - body.insert(key, render_json_value_raw(value, vars, callback, options).await?); - } - - let authentication = { - let mut disabled = false; - let mut auth = BTreeMap::new(); - - match request.authentication.get("disabled") { - Some(Value::Bool(true)) => { - disabled = true; - } - Some(Value::String(template)) => { - disabled = parse_and_render(template.as_str(), vars, callback, options) - .await - .unwrap_or_default() - .is_empty(); - info!( - "Rendering authentication.disabled as a template: {disabled} from \"{template}\"" - ); - } - _ => {} - } - - if disabled { - auth.insert("disabled".to_string(), Value::Bool(true)); - } else { - for (key, value) in request.authentication.clone() { - if key == "disabled" { - auth.insert(key, Value::Bool(false)); - } else { - auth.insert(key, render_json_value_raw(value, vars, callback, options).await?); - } - } - } - - auth - }; - - let url = parse_and_render(request.url.clone().as_str(), vars, callback, options).await?; - let (url, url_parameters) = apply_path_placeholders(&url, &url_parameters); - - Ok(HttpRequest { url, url_parameters, headers, body, authentication, ..request.to_owned() }) -} +pub use yaak_models::render::render_http_request; pub async fn render_grpc_request( r: &GrpcRequest, @@ -148,70 +64,3 @@ pub async fn render_grpc_request( Ok(GrpcRequest { url, metadata, authentication, ..r.to_owned() }) } - -fn strip_disabled_form_entries(v: Value) -> Value { - match v { - Value::Array(items) => Value::Array( - items - .into_iter() - .filter(|item| item.get("enabled").and_then(|e| e.as_bool()).unwrap_or(true)) - .collect(), - ), - v => v, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - #[test] - fn test_strip_disabled_form_entries() { - let input = json!([ - {"enabled": true, "name": "foo", "value": "bar"}, - {"enabled": false, "name": "disabled", "value": "gone"}, - {"enabled": true, "name": "baz", "value": "qux"}, - ]); - let result = strip_disabled_form_entries(input); - assert_eq!( - result, - json!([ - {"enabled": true, "name": "foo", "value": "bar"}, - {"enabled": true, "name": "baz", "value": "qux"}, - ]) - ); - } - - #[test] - fn test_strip_disabled_form_entries_all_disabled() { - let input = json!([ - {"enabled": false, "name": "a", "value": "b"}, - {"enabled": false, "name": "c", "value": "d"}, - ]); - let result = strip_disabled_form_entries(input); - assert_eq!(result, json!([])); - } - - #[test] - fn test_strip_disabled_form_entries_missing_enabled_defaults_to_kept() { - let input = json!([ - {"name": "no_enabled_field", "value": "kept"}, - {"enabled": false, "name": "disabled", "value": "gone"}, - ]); - let result = strip_disabled_form_entries(input); - assert_eq!( - result, - json!([ - {"name": "no_enabled_field", "value": "kept"}, - ]) - ); - } - - #[test] - fn test_strip_disabled_form_entries_non_array_passthrough() { - let input = json!("just a string"); - let result = strip_disabled_form_entries(input.clone()); - assert_eq!(result, input); - } -} diff --git a/crates/yaak/src/send.rs b/crates/yaak/src/send.rs index bcd61386..e48e0606 100644 --- a/crates/yaak/src/send.rs +++ b/crates/yaak/src/send.rs @@ -193,6 +193,7 @@ impl SendRequestExecutor for ConnectionManagerSendRequestExecutor<'_> { proxy: runtime_config.proxy.clone(), client_certificate, dns_overrides: runtime_config.dns_overrides.clone(), + address_filter: None, }) .await?; diff --git a/packages/platform/src/tauri/index.ts b/packages/platform/src/tauri/index.ts index 9d8c1005..03652f2c 100644 --- a/packages/platform/src/tauri/index.ts +++ b/packages/platform/src/tauri/index.ts @@ -49,6 +49,7 @@ function toSyncUnsubscribe(pending: Promise): Unsubscribe { } const ALL_CAPABILITIES: PlatformCapabilities = { + httpSending: true, grpc: true, websocket: true, git: true, diff --git a/packages/platform/src/types.ts b/packages/platform/src/types.ts index b34fd7a8..1e2deb53 100644 --- a/packages/platform/src/types.ts +++ b/packages/platform/src/types.ts @@ -237,6 +237,8 @@ export interface Platform { * from the cargo features they were built with. */ export interface PlatformCapabilities { + /** Send HTTP requests and see the whole response: every header, the redirect chain, timing. */ + httpSending: boolean; /** Send gRPC requests. Needs HTTP/2 trailers, so it needs a real backend. */ grpc: boolean; /** Send WebSocket requests with custom headers and auth. */ diff --git a/packages/platform/src/web/README.md b/packages/platform/src/web/README.md index b8c2db81..5a672556 100644 --- a/packages/platform/src/web/README.md +++ b/packages/platform/src/web/README.md @@ -24,8 +24,10 @@ installs the Tauri host exactly as before. ``` tab (index.ts, commands.ts) ──MessagePort──▶ worker.ts ──▶ @yaakapp-internal/web (wasm) - ◀── model_writes ── crates/yaak-web → yaak-models → SQLite - └─ pages in IndexedDB + │ ◀── model_writes ── crates/yaak-web → yaak-models → SQLite + │ └─ pages in IndexedDB + └── send.ts ──POST rendered request──▶ yaak-send-proxy (crates-server) ──▶ the internet + ◀── NDJSON: events, response, body, cookies ── ``` | File | What it is | @@ -33,13 +35,19 @@ tab (index.ts, commands.ts) ──MessagePort──▶ worker.ts ──▶ @yaak | `index.ts` | The `Platform` implementation. | | `commands.ts` | The command table: model commands forward to the worker; the rest is fixed answers and refusals-with-a-reason. | | `connection.ts` | A tab's end of the wire: request/response over a `MessagePort`, event delivery, and the tab's identity (`label`). | +| `send.ts` | Sending: the worker renders (`prepare_http_send`), the proxy executes, this file stores what comes back where the desktop stores it. | +| `proxy.ts` | The proxy's location and wire shapes, mirrored by hand from `crates-server/yaak-send-proxy/src/wire.rs`. | | `worker.ts` | The process that owns the database. Loads the wasm, opens the DB once, answers each port, fans `model_writes` out to every port. | | `protocol.ts` | The message shapes both sides import. | | `errors.ts` | `UnsupportedCommandError`, the structured refusal. | | `storage.ts` | `navigator.storage.persist()`. | The Rust side is `crates/yaak-web` (`@yaakapp-internal/web`): `boot()`, -`rpc(cmd, payload, label)` returning `{ result, events }`, and blob get/put. +`rpc(cmd, payload, label)` returning `{ result, events }`, blob get/put, and +`prepare_http_send(payload)` — the database half of a send (environment chain, +inherited headers and auth, request settings, cookie jar, rendering), which is +`yaak_models::render::render_http_request`, the same function the desktop +renders with. Its `pkg/` is committed; rebuilding needs a clang with a WebAssembly backend (`brew install llvm`), and `build-wasm.cjs` skips with a notice when there isn't one, so a desktop `npm run bootstrap` never depends on it. @@ -70,13 +78,14 @@ Behaviours worth knowing before changing anything: ## Commands 109 commands are declared in `@yaakapp-internal/rpc-schema`. This host answers -31, declines 44 by name with a reason, and refuses the remaining 34 generically. +32, declines 43 by name with a reason, and refuses the remaining 34 generically. -### Implemented (31) +### Implemented (32) | Group | Commands | | --- | --- | | Models | `models_workspace_models`, `models_upsert`, `models_delete`, `models_duplicate`, `models_get_settings`, `models_get_graphql_introspection`, `models_upsert_graphql_introspection`, `models_grpc_events`, `models_websocket_events` | +| Sending | `cmd_send_http_request` (through the send proxy; see below) | | App | `cmd_metadata`, `cmd_get_workspace_meta`, `cmd_default_headers`, `cmd_get_themes`, `cmd_check_for_updates`, `cmd_dismiss_notification`, `cmd_plugin_init_errors` | | Bodies | `cmd_http_response_body`, `cmd_http_response_body_path`, `cmd_http_request_body`, `cmd_get_http_response_events`, `cmd_get_sse_events` | | Plugin surfaces (empty results) | `cmd_http_request_actions`, `cmd_websocket_request_actions`, `cmd_grpc_request_actions`, `cmd_workspace_actions`, `cmd_folder_actions`, `cmd_template_function_summaries`, `cmd_get_http_authentication_summaries`, `cmd_get_http_authentication_config` | @@ -97,7 +106,7 @@ Some of these answer honestly rather than fully, and the difference matters: - `cmd_metadata` reports empty strings for the data, log, plugin and project directories. There is no filesystem behind this host. -### Declined by name (44) +### Declined by name (43) Each returns an `UnsupportedCommandError` carrying `cmd`, a user-facing `message`, and the `capability` a caller should have checked. The UI turns it @@ -105,7 +114,7 @@ into a toast. | Reason | Commands | | --- | --- | -| Sending isn't available yet (slice 2) | `cmd_send_http_request`, `cmd_send_ephemeral_request`, `cmd_delete_send_history`, `cmd_delete_all_http_responses`, `cmd_import_url` | +| Sending, the parts not wired yet | `cmd_send_ephemeral_request`, `cmd_delete_send_history`, `cmd_delete_all_http_responses`, `cmd_import_url` | | No plugin runtime | `cmd_reload_plugins`, `cmd_plugin_info`, `cmd_plugins_search`, `cmd_plugins_install`, `cmd_plugins_install_from_directory`, `cmd_plugins_uninstall`, `cmd_plugins_updates`, `cmd_plugins_update_all`, `cmd_template_function_config`, `cmd_template_tokens_to_string`, `cmd_call_http_request_action`, `cmd_call_websocket_request_action`, `cmd_call_grpc_request_action`, `cmd_call_workspace_action`, `cmd_call_folder_action`, `cmd_call_http_authentication_action`, `cmd_curl_to_request`, `cmd_format_graphql` | | No filesystem | `cmd_import_data`, `cmd_export_data`, `cmd_save_response`, `cmd_save_base64_to_binary` | | Needs a real socket | `cmd_grpc_reflect`, `cmd_grpc_go`, `cmd_delete_all_grpc_connections`, `cmd_ws_connect`, `cmd_ws_send`, `cmd_ws_close`, `cmd_ws_delete_connections` | @@ -129,7 +138,7 @@ Reported honestly, so callers gate on the question rather than on the host: | True | False | | --- | --- | -| `cookieJar` (the jar stores and edits here; only filling it needs the sender) | `grpc`, `websocket`, `git`, `sync`, `tlsOptions`, `localFiles`, `timeline`, `multiWindow`, `plugins`, `encryption`, `updater`, `clipboardRead`, `systemFonts`, `license` | +| `httpSending`, `timeline`, `cookieJar` | `grpc`, `websocket`, `git`, `sync`, `tlsOptions`, `localFiles`, `multiWindow`, `plugins`, `encryption`, `updater`, `clipboardRead`, `systemFonts`, `license` | `multiWindow: false` means the host cannot open a *second window* on demand — what `cmd_new_child_window` does for Settings and workspace switching. It is not @@ -168,26 +177,35 @@ other's writes for an echo of their own and drop them. crates for their types), so the crate declares the handful of request shapes it needs locally, and `commands.ts` stays typed against `RpcSchema`. -## What slice 2 (the send proxy) will need from this layer +## Sending -Sending becomes a stateless hosted service; this layer stays the only place data -lives. Concretely: +A page cannot see a response the way a desktop app can — CORS exposes a handful +of headers, redirects are followed silently, there is no timeline — so the +network half of a send runs on a small stateless proxy, +`crates-server/yaak-send-proxy`. This layer stays the only place data lives: -1. **A rendered request to send.** The client assembles `HttpSendInputs` and - posts it. Nothing about the workspace is uploaded except what this request - needs. -2. **Cookies out, cookies in.** The active `cookie_jar` model's `cookies` array - goes up with the request; the proxy returns the jar as the exchange left it, - and the client upserts it back through `models_upsert` like any other write. - The proxy keeps nothing. -3. **A response body sink.** `blob_put(responseId, bytes)` in the worker - writes through the desktop's `blob_manager`, chunked the way it chunks. - Streaming will want an append path rather than one whole-body write. -4. **A request body sink** under `${responseId}.request`, which - `cmd_http_request_body` already reads. -5. **Response and timeline models.** `cmd_send_http_request` currently declines; - it will instead upsert an `http_response` as the exchange progresses, plus - `http_response_event` rows once `timeline` becomes true. Both flow through - the same `write()` helper, so other tabs see a send land live. -6. **Blob cleanup is the desktop's.** `delete_http_response` and - `delete_workspace` in `yaak-models` already remove blob chunks. +1. `send.ts` creates the `http_response` row (state `initialized`), as the + desktop does, so anything that goes wrong lands in the response pane. +2. The worker resolves and renders the request (`prepare_http_send`): the + environment chain, inherited headers and auth, request settings, the cookie + jar. This is the desktop's `HttpSendInputs`, in Rust, on the same model layer, + with `yaak_models::render::render_http_request`. Variables (`${[ name ]}`) + render here with no plugins involved. +3. The rendered request, the settings and the jar's cookies are POSTed to the + proxy. It streams back timeline events, the response head, body chunks and a + terminal frame carrying the jar as the send left it. +4. Each frame is written where the desktop writes it: the response row as it + progresses, `http_response_event` rows for the timeline (which is why + `timeline` is true), the body under the response id via `blob_put`, and the + cookie jar through `models_upsert`. Every write fans out to every tab. + +**What sends today:** any saved request whose templates are variables and whose +authentication is none, or an inline header. Sending a request that needs a +template *function* (`${[ timestamp() ]}`) or an authentication plugin (bearer, +basic, OAuth, …) is refused before anything leaves the tab, with a message naming +what it needs; those light up when plugins run in the browser. Requests with a +file body or multipart file fields are refused by the proxy (it has no access to +your files, and must not read its own). + +The proxy URL is `VITE_YAAK_SEND_PROXY_URL` at build time, defaulting to +`http://127.0.0.1:9227` (see `proxy.ts`). Run one with `cargo run -p yaak-send-proxy`. diff --git a/packages/platform/src/web/commands.ts b/packages/platform/src/web/commands.ts index f3eb5b29..324c044b 100644 --- a/packages/platform/src/web/commands.ts +++ b/packages/platform/src/web/commands.ts @@ -21,6 +21,7 @@ import type { RpcSchema } from "@yaakapp-internal/rpc-schema"; import type { CapabilityName, RpcPayload } from "../types"; import type { WorkerConnection } from "./connection"; import { unsupported } from "./errors"; +import { sendHttpRequest } from "./send"; export type AppCmd = keyof RpcSchema; @@ -66,6 +67,18 @@ const HANDLERS: Partial> = { models_websocket_events: (payload, db) => db.rpc("models_websocket_events", payload), cmd_get_workspace_meta: (payload, db) => db.rpc("cmd_get_workspace_meta", payload), + /* ------------------------------- sending ------------------------------- */ + + // The tab renders and stores; a stateless proxy puts the bytes on the wire. + // See send.ts for the whole shape of it. + cmd_send_http_request: (payload, db) => + sendHttpRequest( + db, + text(payload, "requestId"), + str(payload, "environmentId"), + str(payload, "cookieJarId"), + ), + /* -------------------------------- app ---------------------------------- */ async cmd_metadata() { @@ -203,9 +216,8 @@ const HANDLERS: Partial> = { return bytes == null ? null : Array.from(bytes); }, - async cmd_get_http_response_events() { - return []; - }, + // The rows the sender wrote for that response, same table as the desktop. + cmd_get_http_response_events: (payload, db) => db.rpc("cmd_get_http_response_events", payload), async cmd_get_sse_events() { return []; @@ -237,16 +249,10 @@ const HTTP_AUTHENTICATION_SUMMARIES = [ * while the first is a slice away. */ const DECLINED: Partial> = { - // Sending — the next slice. Everything else about a request works today; - // only the part that puts bytes on the network is missing. - cmd_send_http_request: [ - "Sending isn't available in the browser yet — everything else about this request is saved", - null, - ], - cmd_send_ephemeral_request: [ - "Sending isn't available in the browser yet — everything else about this request is saved", - null, - ], + // Saved requests send through the proxy (see send.ts). Ephemeral sends — the + // ones nothing stores, used for GraphQL introspection — take the same road but + // return the body inline; not wired yet. + cmd_send_ephemeral_request: ["Sending unsaved requests isn't available in the browser yet", null], cmd_curl_to_request: ["Importing from cURL needs a plugin, which this host doesn't run", null], // Protocols that need a real socket. diff --git a/packages/platform/src/web/connection.ts b/packages/platform/src/web/connection.ts index 86a02c71..88940fb0 100644 --- a/packages/platform/src/web/connection.ts +++ b/packages/platform/src/web/connection.ts @@ -163,6 +163,11 @@ export class WorkerConnection { return this.request((id) => ({ type: "rpc", id, cmd, payload, label: this.label })); } + /** See `prepare_http_send` in crates/yaak-web: the database half of a send. */ + prepareHttpSend(payload: unknown): Promise { + return this.request((id) => ({ type: "prepare_http_send", id, payload })); + } + async blobGet(blobId: string): Promise | null> { const buf = await this.request((id) => ({ type: "blob_get", id, blobId })); return buf == null ? null : new Uint8Array(buf); diff --git a/packages/platform/src/web/index.ts b/packages/platform/src/web/index.ts index 006d2364..8863b7a5 100644 --- a/packages/platform/src/web/index.ts +++ b/packages/platform/src/web/index.ts @@ -7,11 +7,12 @@ * the origin, so two tabs stay coherent for the same reason two desktop windows * do: one process holds the data and pushes every write to all of them. * - * What a page genuinely cannot do is not faked. There is no file dialog, no - * second window, no clipboard read without a prompt, and — in this slice — no - * sending. Those report false through `capabilities` and refuse with a reason - * if called anyway, so a missing feature shows up as a disabled control or a - * toast that explains itself, never as a silent no-op. + * Sending goes through a small stateless proxy, because a page cannot see a + * response the way a desktop app can (see send.ts). What a page genuinely + * cannot do is not faked: there is no file dialog, no second window, no + * clipboard read without a prompt. Those report false through `capabilities` + * and refuse with a reason if called anyway, so a missing feature shows up as a + * disabled control or a toast that explains itself, never as a silent no-op. */ import type { @@ -32,17 +33,21 @@ import { requestPersistence } from "./storage"; /** What this host can do, reported honestly. */ function capabilitiesFor(): PlatformCapabilities { return { + // Through the send proxy: the tab renders, the proxy executes, the tab + // stores. Requests needing plugin auth or template functions are refused + // with the reason until plugins run here. + httpSending: true, grpc: false, websocket: false, git: false, sync: false, // Certificates and proxies are decided by whoever puts the bytes on the - // wire. Nothing in the browser does yet. + // wire, and the send proxy uses its own. tlsOptions: false, - // The jar can be edited and stored here; only filling it needs the sender. cookieJar: true, localFiles: false, - timeline: false, + // The proxy streams the engine's events back and the sender stores them. + timeline: true, // Whether the host can put a *second window* on this data on demand — what // `cmd_new_child_window` does for Settings and workspace switching. A tab // can't, so those open in place instead. This is not a claim that nothing diff --git a/packages/platform/src/web/protocol.ts b/packages/platform/src/web/protocol.ts index 10a07847..5104463e 100644 --- a/packages/platform/src/web/protocol.ts +++ b/packages/platform/src/web/protocol.ts @@ -10,6 +10,12 @@ /** Tab → worker */ export type ToWorker = | { type: "rpc"; id: number; cmd: string; payload: unknown; label: string } + /** + * The prepare half of a send: resolve, inherit and render a request against + * the database. Its own message rather than an `rpc` command because it is + * async in the engine (rendering is), where every `rpc` command is not. + */ + | { type: "prepare_http_send"; id: number; payload: unknown } | { type: "blob_get"; id: number; blobId: string } | { type: "blob_put"; id: number; blobId: string; bytes: ArrayBuffer } | { type: "blob_delete"; id: number; blobId: string } diff --git a/packages/platform/src/web/proxy.ts b/packages/platform/src/web/proxy.ts new file mode 100644 index 00000000..5246c8ff --- /dev/null +++ b/packages/platform/src/web/proxy.ts @@ -0,0 +1,106 @@ +/** + * The wire to the send proxy: where it is, what goes up, and what comes back. + * + * These shapes mirror `crates-server/yaak-send-proxy/src/wire.rs` by hand. The + * proxy is a separate binary with its own release cadence, so the contract is + * written down on both sides rather than generated across them; a change to one + * is a change to the other, and the frame `type` tags are the versioning. + */ + +/* ------------------------------- location -------------------------------- */ + +/** + * Where the tab sends. Build-time configuration for now: `VITE_YAAK_SEND_PROXY_URL` + * (Vite exposes `VITE_*` to the bundle), defaulting to a proxy on this machine at + * its default port. A per-user setting can replace this later without touching + * the callers, which only ever ask for the URL. + */ +export function proxyBaseUrl(): string { + const env = (import.meta as unknown as { env?: Record }).env; + const configured = env?.VITE_YAAK_SEND_PROXY_URL?.trim(); + return (configured || "http://127.0.0.1:9227").replace(/\/+$/, ""); +} + +export function proxySendUrl(): string { + return `${proxyBaseUrl()}/v1/http/send`; +} + +/* --------------------------------- up ------------------------------------ */ + +/** The body of `POST /v1/http/send`. */ +export interface ProxyRequestBody { + /** The rendered request, in the model shape (see `wire.rs` `SendRequest.request`). */ + request: Record; + settings: { + validateCertificates: boolean; + followRedirects: boolean; + timeoutMs: number; + sendCookies: boolean; + storeCookies: boolean; + }; + /** The jar's cookies to start from, or `null` for no jar at all. */ + cookies: unknown[] | null; +} + +/* -------------------------------- down ----------------------------------- */ + +interface WireHeader { + name: string; + value: string; +} + +export interface ProxySendResponse { + type: "response"; + status: number; + statusReason: string | null; + url: string; + remoteAddr: string | null; + version: string | null; + headers: WireHeader[]; + requestHeaders: WireHeader[]; + contentLength: number | null; + elapsedHeaders: number; + elapsedDns: number; +} + +export type ProxyFrame = + /** A timeline event in the `http_response_event.event` shape. */ + | { type: "event"; event: unknown } + | ProxySendResponse + /** A body chunk, decompressed, base64. */ + | { type: "body"; data: string } + | { + type: "done"; + elapsed: number; + contentLength: number; + contentLengthCompressed: number; + cookies: unknown[] | null; + } + | { type: "error"; message: string; cookies: unknown[] | null }; + +/** + * Yield frames from an NDJSON stream as they arrive. A partial trailing line is + * held until its newline comes; anything left when the stream ends is dropped, + * because a frame without its newline is a frame the proxy didn't finish writing. + */ +export async function* readFrames(stream: ReadableStream): AsyncGenerator { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + let newline = buffer.indexOf("\n"); + while (newline !== -1) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + if (line.trim() !== "") yield JSON.parse(line) as ProxyFrame; + newline = buffer.indexOf("\n"); + } + } + } finally { + reader.releaseLock(); + } +} diff --git a/packages/platform/src/web/send.ts b/packages/platform/src/web/send.ts new file mode 100644 index 00000000..b10ce6bd --- /dev/null +++ b/packages/platform/src/web/send.ts @@ -0,0 +1,386 @@ +/** + * Sending an HTTP request from a tab. + * + * A tab can't see a response the way the desktop can — CORS hides most headers, + * redirects are followed silently, there is no timeline — so the network half of + * a send happens on a small stateless proxy (`crates-server/yaak-send-proxy`). + * Everything else happens here, against this tab's own database, in the same + * order the desktop does it: + * + * 1. create the `http_response` row (state: initialized); + * 2. resolve and render the request in the worker (`prepare_http_send`: the + * environment chain, inherited headers and auth, request settings, cookie + * jar — the desktop's `HttpSendInputs`, in Rust, on the same model layer); + * 3. POST the rendered request to the proxy and consume its stream: timeline + * events, the response head, body chunks, and a terminal frame; + * 4. write what comes back where the desktop writes it — the response row as + * it progresses, `http_response_event` rows for the timeline, the body + * blob under the response id, the cookie jar with the proxy's changes. + * + * The proxy keeps nothing. Every byte it sees comes from this tab and every + * byte it returns is stored by this tab. + */ + +import type { WorkerConnection } from "./connection"; +import type { ProxyFrame, ProxyRequestBody, ProxySendResponse } from "./proxy"; +import { proxySendUrl, readFrames } from "./proxy"; + +/* -------------------------------- shapes --------------------------------- */ + +// The model types this file writes, spelled out rather than imported from +// `@yaakapp-internal/models`: the platform package sits underneath the model +// package in the dependency graph and must not import it. + +interface HttpResponseHeader { + name: string; + value: string; +} + +/** The subset of the `http_response` model this sender writes. */ +interface ResponsePatch { + model: "http_response"; + id: string; + requestId: string; + workspaceId: string; + state: "initialized" | "connected" | "closed"; + url: string; + status: number; + statusReason: string | null; + version: string | null; + remoteAddr: string | null; + headers: HttpResponseHeader[]; + requestHeaders: HttpResponseHeader[]; + contentLength: number | null; + contentLengthCompressed: number | null; + elapsed: number; + elapsedHeaders: number; + elapsedDns: number; + error: string | null; +} + +interface CookieJarModel { + model: "cookie_jar"; + id: string; + cookies: unknown[]; + [key: string]: unknown; +} + +/** What `prepare_http_send` (crates/yaak-web) hands back. */ +interface PreparedHttpSend { + request: { url: string; [key: string]: unknown }; + settings: ProxyRequestBody["settings"]; + settingEvents: unknown[]; + cookieJar: CookieJarModel | null; +} + +/** The desktop writes progress at most this often while a body streams in. */ +const PROGRESS_INTERVAL_MS = 100; + +/* --------------------------------- send ---------------------------------- */ + +export async function sendHttpRequest( + db: WorkerConnection, + requestId: string, + environmentId: string | null, + cookieJarId: string | null, +): Promise { + // The response row exists before anything can go wrong, as on the desktop, so + // a failure to render or to reach the proxy lands in the response pane as + // that response's error rather than as a toast that names no request. + const workspaceId = await workspaceIdOfRequest(db, requestId); + const response = new ResponseWriter(db, { + model: "http_response", + id: "", + requestId, + workspaceId, + state: "initialized", + url: "", + status: 0, + statusReason: null, + version: null, + remoteAddr: null, + headers: [], + requestHeaders: [], + contentLength: null, + contentLengthCompressed: null, + elapsed: 0, + elapsedHeaders: 0, + elapsedDns: 0, + error: null, + }); + await response.create(); + + const cancel = new AbortController(); + const unlistenCancel = db.listen(`cancel_http_response_${response.id}`, () => cancel.abort()); + + try { + await runSend(db, response, requestId, environmentId, cookieJarId, cancel.signal); + } catch (err) { + const message = cancel.signal.aborted ? "Request canceled" : errorMessage(err); + await response.finish({ error: message }); + } finally { + unlistenCancel(); + } + return response.current(); +} + +async function runSend( + db: WorkerConnection, + response: ResponseWriter, + requestId: string, + environmentId: string | null, + cookieJarId: string | null, + signal: AbortSignal, +): Promise { + const prepared = await db.prepareHttpSend({ + requestId, + environmentId, + cookieJarId, + }); + await response.patch({ url: prepared.request.url }); + + const timeline = new TimelineWriter(db, response.id, response.workspaceId); + timeline.push(prepared.settingEvents); + + const body: ProxyRequestBody = { + request: prepared.request, + settings: prepared.settings, + cookies: prepared.cookieJar?.cookies ?? null, + }; + const startedAt = performance.now(); + const res = await fetch(proxySendUrl(), { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + signal, + }).catch((err: unknown) => { + if (signal.aborted) throw err; + throw new Error(`Couldn't reach the send proxy at ${proxySendUrl()}: ${errorMessage(err)}`); + }); + + if (!res.ok) { + // A refusal, not a failed send: bad destination, rate limit, a body the + // proxy can't build. It comes as JSON with the reason. + const text = await res.text(); + let reason = text; + try { + reason = (JSON.parse(text) as { error?: string }).error ?? text; + } catch { + /* not JSON; the text is the reason */ + } + throw new Error(reason || `The send proxy answered ${res.status}`); + } + if (res.body == null) throw new Error("The send proxy sent no body"); + + const chunks: Uint8Array[] = []; + let received = 0; + let lastProgress = startedAt; + let terminal: ProxyFrame | null = null; + + for await (const frame of readFrames(res.body)) { + switch (frame.type) { + case "event": + timeline.push([frame.event]); + break; + case "response": + await response.patch(headOf(frame)); + break; + case "body": { + const bytes = base64ToBytes(frame.data); + chunks.push(bytes); + received += bytes.byteLength; + const now = performance.now(); + if (now - lastProgress >= PROGRESS_INTERVAL_MS) { + lastProgress = now; + await response.patch({ + contentLength: received, + elapsed: Math.round(now - startedAt), + }); + } + break; + } + case "done": + case "error": + terminal = frame; + break; + } + if (terminal != null) break; + } + + // Everything the proxy said about the timeline is in the database before the + // response is marked closed, so a reader that wakes on "closed" sees all of it. + await timeline.flush(); + + if (terminal == null) { + throw new Error("The send proxy closed the stream without finishing"); + } + + // Cookies come back on both outcomes: a hop before the failing one may have + // set some, and the desktop keeps those too. + if (prepared.cookieJar != null && terminal.cookies != null) { + await persistCookies(db, prepared.cookieJar, terminal.cookies); + } + + if (terminal.type === "error") { + throw new Error(terminal.message); + } + + // The body is written under the response id, which is how every reader — + // `cmd_http_response_body`, the image viewer, the download button — asks for + // it. One write, once the whole body is here: the worker's blob store has no + // append, and a body larger than memory is over the proxy's cap anyway. + await db.blobPut(response.id, concat(chunks, received)); + await response.finish({ + contentLength: terminal.contentLength, + contentLengthCompressed: terminal.contentLengthCompressed, + elapsed: terminal.elapsed, + }); +} + +function headOf(frame: ProxySendResponse): Partial { + return { + state: "connected", + status: frame.status, + statusReason: frame.statusReason, + url: frame.url, + remoteAddr: frame.remoteAddr, + version: frame.version, + headers: frame.headers, + requestHeaders: frame.requestHeaders, + contentLength: frame.contentLength, + elapsedHeaders: frame.elapsedHeaders, + elapsedDns: frame.elapsedDns, + }; +} + +/* ------------------------------- helpers --------------------------------- */ + +/** + * The response row, written the way the desktop writes it: created empty, + * patched as the send progresses, closed at the end. Each write goes through + * `models_upsert`, so every tab on this database sees the response land. + */ +class ResponseWriter { + private state: ResponsePatch; + + constructor( + private readonly db: WorkerConnection, + initial: ResponsePatch, + ) { + this.state = initial; + } + + get id(): string { + return this.state.id; + } + + get workspaceId(): string { + return this.state.workspaceId; + } + + current(): ResponsePatch { + return this.state; + } + + async create(): Promise { + const { id: _, ...withoutId } = this.state; + const id = await this.db.rpc("models_upsert", { model: withoutId }); + this.state = { ...this.state, id }; + } + + async patch(patch: Partial): Promise { + // Structured clone carries `undefined` across to the worker as a present + // key, and the model layer reads that as "wrong type" and refuses the whole + // model. Nothing here should produce one, but a missing wire field must + // not take the response row down with it. + const defined = Object.fromEntries(Object.entries(patch).filter(([, v]) => v !== undefined)); + this.state = { ...this.state, ...defined }; + await this.db.rpc("models_upsert", { model: this.state }); + } + + async finish(patch: Partial): Promise { + await this.patch({ ...patch, state: "closed" }); + } +} + +/** + * Timeline events, written in the order they arrived. Writes are chained rather + * than awaited inline so a burst of `header_down` events doesn't serialise the + * body read behind a database round trip each, and `flush()` is the point at + * which the whole timeline is known to be in the database. + */ +class TimelineWriter { + private queue: unknown[] = []; + private inFlight: Promise = Promise.resolve(); + + constructor( + private readonly db: WorkerConnection, + private readonly responseId: string, + private readonly workspaceId: string, + ) {} + + push(events: unknown[]): void { + if (events.length === 0) return; + this.queue.push(...events); + this.inFlight = this.inFlight.then(() => this.drain()); + } + + private async drain(): Promise { + if (this.queue.length === 0) return; + const events = this.queue; + this.queue = []; + await this.db.rpc("web_insert_http_response_events", { + responseId: this.responseId, + workspaceId: this.workspaceId, + events, + }); + } + + flush(): Promise { + return this.inFlight; + } +} + +async function persistCookies( + db: WorkerConnection, + jar: CookieJarModel, + cookies: unknown[], +): Promise { + // The desktop compares before writing so a jar edited mid-send isn't clobbered + // by an unchanged copy. Structural equality is enough here: cookies are plain + // data and the proxy hands back the whole jar. + if (JSON.stringify(cookies) === JSON.stringify(jar.cookies)) return; + await db.rpc("models_upsert", { model: { ...jar, cookies } }); +} + +/** + * The request's workspace, needed to create the response row before the worker + * has resolved the request (which is where a render refusal would land). + */ +async function workspaceIdOfRequest(db: WorkerConnection, requestId: string): Promise { + const req = await db.rpc<{ workspaceId: string }>("web_get_http_request", { requestId }); + return req.workspaceId; +} + +function errorMessage(err: unknown): string { + if (err instanceof Error) return err.message; + return String(err); +} + +function base64ToBytes(data: string): Uint8Array { + const bin = atob(data); + const out = new Uint8Array(bin.length); + for (let i = 0; i < bin.length; i++) out[i] = bin.charCodeAt(i); + return out; +} + +function concat(chunks: Uint8Array[], total: number): Uint8Array { + if (chunks.length === 1) return chunks[0]!; + const out = new Uint8Array(total); + let offset = 0; + for (const c of chunks) { + out.set(c, offset); + offset += c.byteLength; + } + return out; +} diff --git a/packages/platform/src/web/worker.ts b/packages/platform/src/web/worker.ts index ae638adc..bd6d75e7 100644 --- a/packages/platform/src/web/worker.ts +++ b/packages/platform/src/web/worker.ts @@ -122,7 +122,7 @@ async function handle(port: MessagePort, message: ToWorker): Promise { send(port, { type: "error", id: message.id, message: bootError ?? "Database failed to open" }); return; } - const { rpc, blob_get, blob_put, blob_delete } = engine!; + const { rpc, blob_get, blob_put, blob_delete, prepare_http_send } = engine!; try { switch (message.type) { @@ -142,6 +142,11 @@ async function handle(port: MessagePort, message: ToWorker): Promise { } return; } + case "prepare_http_send": { + const prepared = await prepare_http_send(message.payload); + send(port, { type: "result", id: message.id, result: prepared }); + return; + } case "blob_get": { const bytes = blob_get(message.blobId); if (bytes == null) {