diff --git a/CHANGELOG.md b/CHANGELOG.md index 78e63159..8e0e76c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## [Unreleased] +- Security (TC-1132, TC-1080 Parts C–D): webhook registration and every delivery attempt share an egress policy covering special-use IPv4/IPv6, embedded IPv4 (mapped, compatible, NAT64 and 6to4), Teredo and site-local IPv6. Ambiguous local-use NAT64 is refused if any valid IPv4 interpretation is forbidden. Corrected public controls such as `192.2.1.1` and `198.51.0.1` are no longer over-blocked. Mixed DNS answers retain only permitted addresses. +- Webhooks deny the resolved addresses of the configured network database, quota URL, space allowlist and explicit S3 endpoint on every port, including aliases and embedded IPv4 forms. The service deny-set refreshes every ten minutes; validation waits for an in-progress refresh, and failed/stale resolution closes egress until recovery. Temporary DNS/policy failures retry on the existing bounded schedule; invalid or forbidden callbacks dead-letter. Existing subscriptions whose destinations become forbidden dead-letter on the next attempt. +- Webhook DNS is bounded to one second. Each delivery resolves once and gives only vetted addresses to reqwest through a sealed, immutable-target sender. A fresh policy lease rechecks literal and hostname addresses and remains held through connection/send, so refresh cannot invalidate a vetted target before it connects. Proxy environment variables are ignored, redirects/non-HTTPS destinations refused, connect time capped at two seconds, and the configured total timeout includes lease/freshness work and sending. No production private-destination/development flag was added; body bytes, headers and HMAC signing are unchanged. +- Registration caps both raw and canonical callback URLs at 2,048 bytes and secrets at 256 bytes. Retry attempts 6 and later keep the one-hour backoff; `hooks.webhook_max_attempts` must not exceed 10. +- Quota requests encode the original SpaceId as one path segment and do not follow redirects; cache/override keys and fallback remain unchanged. Tunnel upstream requests do not follow redirects and have a 30-second total timeout, including body reads; 3xx responses and existing per-request errors retain their framing. +- Network DID methods remain disabled by default. Full TC-1080 Part B (node-owned did:web fetcher, destination policy, timeouts, body limits, and refusal tests) is **required before any future re-enable**; this change does not implement or enable that fetcher. No schema migration. + ## [1.20.6] - 2026-10-14 A security fix (TC-1080, Part A). UCAN issuer resolution no longer dereferences network DID methods by default, so a request can't make the node fetch arbitrary or internal hosts through a crafted issuer DID (#284). diff --git a/README.md b/README.md index d5bd8bba..e65dee08 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,12 @@ containing only `storage_limit_bytes` remain supported. Local testing can set an in-memory limit with authenticated `PUT /admin/quota/` and `{"limit_bytes":1}`; overrides reset on restart. +The billing request percent-encodes the original space ID as one path segment; +the quota/cache key itself is unchanged. The client never follows redirects, +so the service bearer is sent only on the configured billing request. A `3xx` +answer uses the existing quota fallback policy, with the existing 3-second total +and 2-second connect deadlines. + ### Signed KV URLs `POST /signed/kv` accepts a normal TinyCloud invocation whose authorized capabilities include a `tinycloud.kv/get` capability that can be attenuated to the requested `{space, path}`. The invocation may include other capabilities. The JSON body is: @@ -311,12 +317,18 @@ setting. DID parsing and string comparisons, including share-email `did:web` identifiers, are unchanged. Existing route status codes and policy machine codes are preserved. -Configure `[global.did_resolution] network_methods = false` in `tinycloud.toml` -or use the canonical environment variable -`TINYCLOUD_DID_RESOLUTION__NETWORK_METHODS`. For rollback without rebuilding, -set `TINYCLOUD_DID_RESOLUTION__NETWORK_METHODS=true` and restart the node. -This restores the previous methods with the unhardened SSI client, while issuer -queries remain refused. The previous image is the fallback rollback. +Keep `[global.did_resolution] network_methods = false` in `tinycloud.toml` +or the canonical environment variable +`TINYCLOUD_DID_RESOLUTION__NETWORK_METHODS=false`. Network DID methods, including +`did:web`, remain disabled under the current policy. + +**Full TC-1080 Part B hardening is REQUIRED before any future re-enable**, +including setting the existing `network_methods` option to `true`. The +unhardened SSI client is not an acceptable rollback path. Part B is not +implemented here, and there is no Part B development bypass. The required +node-side fetcher, destination enforcement, resource limits and local-listener +acceptance checks are described in the +[developer gate](docs/delegation-flow.md#network-did-method-future-enable-gate-tc-1132). The startup log records the effective policy. Refusals increment `tinycloud_did_resolution_refused_total{method,reason}` (even when latency diff --git a/docs/delegation-flow.md b/docs/delegation-flow.md index a1eaf0db..e8ff79ad 100644 --- a/docs/delegation-flow.md +++ b/docs/delegation-flow.md @@ -23,6 +23,44 @@ The delegation hierarchy enables: | Session Key | `did:key:z6Mk...#z6Mk...` | `did:key:z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK#z6MkhaXgBZDvotDkL5257faiztiGiC2QtKLGpbnnEGta2doK` | | PKH (Wallet) | `did:pkh:eip155:{chainId}:{address}` | `did:pkh:eip155:1:0x742d35Cc6634C0532925a3b844Bc9e7595f1Ac8b` | +### Network DID method future-enable gate (TC-1132) + +The node's UCAN issuer resolver currently permits only offline `did:key`, +`did:pkh` and `did:jwk` resolution by default. Network methods, including +`did:web`, remain disabled with `network_methods = false`. DID identifiers used +for parsing or identity comparisons (for example share-email `did:web` values) +are unaffected; this gate concerns network resolution during verification. + +**Full TC-1080 plan Part B is REQUIRED before any future re-enable**, including +the existing `network_methods = true` setting. Do not restore the unhardened SSI +fetcher as a rollback. Part B is not implemented by TC-1132 Parts C/D, and no +Part B development flag or bypass is provided. + +Before a future change enables `did:web`, it must implement and verify all of +Part B: + +- A node-owned DID-web fetcher behind the node resolver, not an SSI-fork change; + absent a hardened fetcher, network resolution must stay refused. +- Strict DID-to-HTTPS URL construction and document identity checks, with + IP-literal hosts, unsafe path segments, unexpected ports and issuer queries + refused. +- Destination policy enforced on the exact DNS addresses used to connect: + forbidden IPv4/IPv6 classes (including embedded IPv4), metadata/link-local + destinations and the node's database/internal endpoint deny-set. No second + unchecked DNS lookup, environment-proxy bypass or redirect following. +- Bounded DNS, connect and total request time; status/content-type checks; a + 64 KiB response cap; concurrency and per-host rate limits; bounded positive + and negative caches with in-flight deduplication; count-only refusal evidence. +- Local-listener and injected-resolver regressions for every refused address + class, database/internal destinations, URL construction, ignored proxies, + DNS rebinding, redirects, oversized bodies, timeouts and rate/cache behavior. + Retain Part A refusal checks and prove a permitted DID document verifies + end-to-end only after the hardened fetcher is installed. + +Any development exception described by the original Part B plan is itself +subject to this gate; it must never exempt metadata/link-local addresses or +the internal endpoint deny-set. This documentation does not enable one. + ### Space ID Format Spaces are identified using the format: diff --git a/docs/specs/node-control-plane-v1.md b/docs/specs/node-control-plane-v1.md index 542a3dd2..3816cf0e 100644 --- a/docs/specs/node-control-plane-v1.md +++ b/docs/specs/node-control-plane-v1.md @@ -1162,6 +1162,11 @@ Connection lifecycle (owned by `serve`, not the CLI): response streamed back as `response` + `responseBody...` frames. A per-request failure sends an `error` frame for that request's `id` only — it never closes the socket. +- The loopback HTTP client never follows redirects. A `3xx` response, its + `Location` header and body are returned through the ordinary response frames; + the node makes no request to the redirect target. Each upstream request has + a 30-second total deadline, including response-body reads. A timeout uses the + existing per-request `error` frame and leaves the tunnel socket open. - The relay pings every 30s; the underlying WebSocket library answers pings automatically. If no frame (data, ping, or close) arrives for an extended idle window, the node treats the connection as dead and reconnects. diff --git a/tinycloud-node-server/src/config.rs b/tinycloud-node-server/src/config.rs index f8d5f911..ef2b39d0 100644 --- a/tinycloud-node-server/src/config.rs +++ b/tinycloud-node-server/src/config.rs @@ -863,6 +863,16 @@ impl Default for HooksConfig { } } +impl HooksConfig { + /// Reject retry budgets outside the supported bound rather than silently truncating. + pub fn validate(&self) -> Result<(), &'static str> { + if self.webhook_max_attempts > 10 { + return Err("hooks.webhook_max_attempts must not exceed 10"); + } + Ok(()) + } +} + /// Retention sweep policy for terminal `hook_delivery` rows and expired /// `signed_kv_ticket` rows (TC-287). /// @@ -1425,6 +1435,20 @@ mod tests { use std::io::Write; use tempfile::NamedTempFile; + #[test] + fn tc1132_startup_validates_retry_budget_bounds() { + for attempts in [0, 1, 10] { + let mut config = Config::default(); + config.hooks.webhook_max_attempts = attempts; + assert!(crate::resolve_runtime_config(&config).is_ok()); + } + for attempts in [11, usize::MAX] { + let mut config = Config::default(); + config.hooks.webhook_max_attempts = attempts; + assert!(crate::resolve_runtime_config(&config).is_err()); + } + } + fn enabled_config() -> ShareEmailConfig { ShareEmailConfig { enabled: true, diff --git a/tinycloud-node-server/src/egress.rs b/tinycloud-node-server/src/egress.rs new file mode 100644 index 00000000..f27e03de --- /dev/null +++ b/tinycloud-node-server/src/egress.rs @@ -0,0 +1,613 @@ +//! Outbound webhook destination policy. This does not enable network DID resolution. +use crate::{config::Config, BlockConfig}; +use reqwest::{ + dns::{Addrs, Resolve, Resolving}, + header::{HeaderMap, HeaderName, HeaderValue, InvalidHeaderName, InvalidHeaderValue}, + redirect, Client, Url, +}; +use std::{ + collections::HashSet, + net::{IpAddr, Ipv4Addr, SocketAddr}, + sync::Arc, + time::Duration, +}; +use tokio::{ + sync::{Mutex, RwLock, RwLockReadGuard}, + time::Instant, +}; + +const DNS_TIMEOUT: Duration = Duration::from_secs(1); +const REFRESH_INTERVAL: Duration = Duration::from_secs(600); + +/// Fail-closed destination and internal-endpoint resolution errors. +#[derive(Debug, thiserror::Error)] +pub enum EgressError { + #[error("destination DNS lookup failed")] + Dns, + #[error("destination DNS lookup timed out")] + DnsTimeout, + #[error("destination has no permitted address")] + Forbidden, + #[error("internal endpoint deny-set is unavailable")] + Unavailable, + #[error("invalid configured internal endpoint")] + InvalidEndpoint, +} + +impl EgressError { + /// Temporary failures may be retried without permitting a network connection. + pub fn is_retryable(&self) -> bool { + matches!(self, Self::Dns | Self::DnsTimeout | Self::Unavailable) + } +} + +#[async_trait::async_trait] +pub(crate) trait LookupHost: Send + Sync { + async fn lookup(&self, host: &str) -> Result, EgressError>; +} + +struct SystemLookup; +#[async_trait::async_trait] +impl LookupHost for SystemLookup { + async fn lookup(&self, host: &str) -> Result, EgressError> { + tokio::net::lookup_host((host, 0)) + .await + .map(|addresses| addresses.map(|address| address.ip()).collect()) + .map_err(|_| EgressError::Dns) + } +} + +struct DenySet { + addresses: Arc>, + refreshed: Option, + healthy: bool, +} + +/// Shared policy for callback registration and every delivery attempt. +/// Internal endpoint lookup failure or stale state denies all destinations. +pub struct EgressPolicy { + lookup: Arc, + internal_hosts: Vec, + deny: RwLock, + refresh_lock: Mutex<()>, + #[cfg(test)] + allow_loopback: bool, + #[cfg(test)] + roots: Vec, +} + +impl EgressPolicy { + /// Gather operator-configured service hosts; SQLite has no network host. + /// Credentials and paths are neither retained nor included in errors. + pub fn from_config(config: &Config, quota_url: Option<&str>) -> Result { + let mut hosts = Vec::new(); + if let Some(database) = config.storage.database.as_deref() { + if !database.starts_with("sqlite:") { + hosts.push(endpoint_host(database)?); + } + } + if let Some(url) = quota_url { + hosts.push(endpoint_host(url)?); + } + if let Some(allowlist) = &config.spaces.allowlist { + hosts.push(endpoint_host(&allowlist.0)?); + } + if let BlockConfig::A(s3) = &config.storage.blocks { + if let Some(endpoint) = &s3.endpoint { + hosts.push(endpoint_host(&endpoint.to_string())?); + } + } + hosts.sort_unstable(); + hosts.dedup(); + Ok(Self::new(hosts, Arc::new(SystemLookup))) + } + + fn new(internal_hosts: Vec, lookup: Arc) -> Self { + let healthy = internal_hosts.is_empty(); + Self { + lookup, + internal_hosts, + deny: RwLock::new(DenySet { + addresses: Arc::new(HashSet::new()), + refreshed: healthy.then(Instant::now), + healthy, + }), + refresh_lock: Mutex::new(()), + #[cfg(test)] + allow_loopback: false, + #[cfg(test)] + roots: Vec::new(), + } + } + + async fn lookup_bounded(&self, host: &str) -> Result, EgressError> { + if let Ok(ip) = host.trim_matches(['[', ']']).parse() { + return Ok(vec![ip]); + } + tokio::time::timeout(DNS_TIMEOUT, self.lookup.lookup(host)) + .await + .map_err(|_| EgressError::DnsTimeout)? + } + + /// Atomically replace the deny-set only after every host resolves. + /// On failure retain the last set but close all webhook egress until recovery. + pub async fn refresh(&self) -> Result<(), EgressError> { + let _guard = self.refresh_lock.lock().await; + self.refresh_locked().await + } + + async fn refresh_locked(&self) -> Result<(), EgressError> { + // Exclusive state access closes egress while DNS is in progress. Keep + // the prior state intact until completion so cancellation is harmless. + let mut deny = self.deny.write().await; + let results = futures::future::join_all( + self.internal_hosts + .iter() + .map(|host| self.lookup_bounded(host)), + ) + .await; + let mut addresses = HashSet::new(); + let mut success = true; + for result in results { + match result { + Ok(ips) if !ips.is_empty() => { + for ip in ips { + addresses.insert(ip); + for v4 in embedded_v4(ip) { + addresses.insert(IpAddr::V4(v4)); + } + } + } + _ => success = false, + } + } + deny.refreshed = Some(Instant::now()); + if success { + deny.addresses = Arc::new(addresses); + deny.healthy = true; + Ok(()) + } else { + deny.healthy = false; + Err(EgressError::Unavailable) + } + } + + fn stale(&self, deny: &DenySet) -> bool { + !self.internal_hosts.is_empty() + && deny + .refreshed + .is_none_or(|at| at.elapsed() >= REFRESH_INTERVAL) + } + + async fn read_state(&self) -> Result, EgressError> { + tokio::time::timeout(DNS_TIMEOUT, self.deny.read()) + .await + .map_err(|_| EgressError::Unavailable) + } + + async fn ensure_fresh(&self) -> Result<(), EgressError> { + if self.internal_hosts.is_empty() { + return Ok(()); + } + let stale = { + let state = self.read_state().await?; + self.stale(&state) + }; + if stale { + let _guard = tokio::time::timeout(DNS_TIMEOUT, self.refresh_lock.lock()) + .await + .map_err(|_| EgressError::Unavailable)?; + let stale = { + let state = self.read_state().await?; + self.stale(&state) + }; + if stale { + self.refresh_locked().await?; + } + } + if !self.read_state().await?.healthy { + return Err(EgressError::Unavailable); + } + Ok(()) + } + + async fn send_lease(&self) -> Result, EgressError> { + loop { + self.ensure_fresh().await?; + let lease = self.read_state().await?; + if !lease.healthy { + return Err(EgressError::Unavailable); + } + if !self.stale(&lease) { + return Ok(lease); + } + // A refresh may have completed, failed or expired between checks. + drop(lease); + } + } + + fn permits_ip(&self, ip: IpAddr, denied: &HashSet) -> bool { + if denied.contains(&ip) || embedded_v4(ip).any(|v4| denied.contains(&IpAddr::V4(v4))) { + return false; + } + let forbidden = Self::forbidden(ip).is_some(); + #[cfg(test)] + let forbidden = forbidden && !(self.allow_loopback && ip.is_loopback()); + !forbidden + } + + async fn permitted( + &self, + ips: impl Iterator, + port: u16, + ) -> Result, EgressError> { + let deny = self.send_lease().await?; + let mut addresses = Vec::new(); + for ip in ips { + let address = SocketAddr::new(ip, port); + if self.permits_ip(ip, &deny.addresses) && !addresses.contains(&address) { + addresses.push(address); + } + } + if addresses.is_empty() { + Err(EgressError::Forbidden) + } else { + Ok(addresses) + } + } + + /// Resolve once, filter each address, and bind the result to this URL's host. + /// Literal hosts are checked here because reqwest bypasses DNS for them. + pub async fn resolve_url(&self, url: &Url) -> Result { + self.ensure_fresh().await?; + let host = url.host_str().ok_or(EgressError::Forbidden)?; + let port = url.port_or_known_default().ok_or(EgressError::Forbidden)?; + let addresses = if let Ok(ip) = host.trim_matches(['[', ']']).parse() { + self.permitted(std::iter::once(ip), port).await? + } else { + let ips = self.lookup_bounded(host).await?; + self.permitted(ips.into_iter(), port).await? + }; + Ok(VettedDestination { + url: url.clone(), + addresses: addresses.into(), + }) + } + + /// Return the special-use class, recursively checking embedded IPv4. + pub fn forbidden(ip: IpAddr) -> Option<&'static str> { + match ip { + IpAddr::V4(ip) => { + let [a, b, c, _] = ip.octets(); + match (a, b, c) { + (0, _, _) => Some("this_network"), + (10, _, _) | (172, 16..=31, _) | (192, 168, _) => Some("private"), + (100, 64..=127, _) => Some("cgnat"), + (127, _, _) => Some("loopback"), + (169, 254, _) => Some("link_local"), + (192, 0, 0) => Some("ietf"), + (192, 0, 2) | (198, 51, 100) | (203, 0, 113) => Some("documentation"), + (198, 18..=19, _) => Some("benchmark"), + (192, 88, 99) => Some("6to4_relay"), + (224..=239, _, _) => Some("multicast"), + (240..=255, _, _) => Some("reserved"), + _ => None, + } + } + IpAddr::V6(ip) => { + let s = ip.segments(); + if ip.is_unspecified() { + return Some("unspecified"); + } + if ip.is_loopback() { + return Some("loopback"); + } + if s[0] & 0xffc0 == 0xfe80 { + return Some("link_local"); + } + if s[0] & 0xffc0 == 0xfec0 { + return Some("site_local"); + } + if s[0] & 0xfe00 == 0xfc00 { + return Some("unique_local"); + } + if ip.is_multicast() { + return Some("multicast"); + } + if s[0] == 0x2001 && s[1] == 0xdb8 { + return Some("documentation"); + } + if s[..4] == [0x100, 0, 0, 0] { + return Some("discard"); + } + if s[0] == 0x2001 && s[1] == 0 { + return Some("teredo"); + } + embedded_v4(IpAddr::V6(ip)).find_map(|v4| Self::forbidden(IpAddr::V4(v4))) + } + } + } + + #[cfg(test)] + pub(crate) fn for_test( + hosts: Vec, + lookup: Arc, + allow_loopback: bool, + ) -> Self { + let mut policy = Self::new(hosts, lookup); + policy.allow_loopback = allow_loopback; + policy + } + + #[cfg(test)] + pub(crate) fn test_loopback() -> Arc { + Arc::new(Self::for_test(Vec::new(), Arc::new(SystemLookup), true)) + } + + #[cfg(test)] + pub(crate) fn permits_test_http(&self) -> bool { + self.allow_loopback + } + + #[cfg(test)] + pub(crate) fn with_test_root(mut self, root: reqwest::Certificate) -> Self { + self.roots.push(root); + self + } +} + +fn endpoint_host(value: &str) -> Result { + let url = Url::parse(value).map_err(|_| EgressError::InvalidEndpoint)?; + url.host_str() + .filter(|host| !host.is_empty()) + .map(|host| host.trim_matches(['[', ']']).to_owned()) + .ok_or(EgressError::InvalidEndpoint) +} + +// RFC 6052 layouts. Local-use NAT64 can use /48, /56, /64 or /96 within +// 64:ff9b:1::/48; reject if any canonical interpretation carries a denied v4. +// Check the u octet and zero suffix for the shorter layouts, not just the tail. +fn embedded_v4(ip: IpAddr) -> impl Iterator { + let mut values = [None; 4]; + if let IpAddr::V6(ip) = ip { + let s = ip.segments(); + let b = ip.octets(); + if s[..6] == [0, 0, 0, 0, 0, 0] + || s[..6] == [0, 0, 0, 0, 0, 0xffff] + || s[..6] == [0x64, 0xff9b, 0, 0, 0, 0] + { + values[0] = Some(Ipv4Addr::new(b[12], b[13], b[14], b[15])); + } else if s[0] == 0x2002 { + values[0] = Some(Ipv4Addr::new(b[2], b[3], b[4], b[5])); + } else if s[..3] == [0x64, 0xff9b, 1] { + if b[8] == 0 && b[11..].iter().all(|byte| *byte == 0) { + values[0] = Some(Ipv4Addr::new(b[6], b[7], b[9], b[10])); + } + if b[8] == 0 && b[12..].iter().all(|byte| *byte == 0) { + values[1] = Some(Ipv4Addr::new(b[7], b[9], b[10], b[11])); + } + if b[8] == 0 && b[13..].iter().all(|byte| *byte == 0) { + values[2] = Some(Ipv4Addr::new(b[9], b[10], b[11], b[12])); + } + values[3] = Some(Ipv4Addr::new(b[12], b[13], b[14], b[15])); + } + } + values.into_iter().flatten() +} + +/// A single lookup's vetted addresses. No caller can insert unvetted addresses. +pub struct VettedDestination { + url: Url, + addresses: Arc<[SocketAddr]>, +} + +impl VettedDestination { + /// The immutable URL whose addresses were vetted by this lookup. + pub fn url(&self) -> &Url { + &self.url + } + + #[cfg(test)] + pub(crate) fn is_loopback(&self) -> bool { + self.addresses + .iter() + .all(|address| address.ip().is_loopback()) + } +} + +/// Resolver for the sealed sender's lease-bound address snapshot. It never +/// performs a second DNS lookup or reacquires the policy lock. +struct PolicyResolver { + host: String, + addresses: Arc<[SocketAddr]>, + policy: Arc, + denied: Arc>, +} + +impl Resolve for PolicyResolver { + fn resolve(&self, name: hyper::client::connect::dns::Name) -> Resolving { + let policy = self.policy.clone(); + let addresses = self.addresses.clone(); + let denied = self.denied.clone(); + let matches = name.as_str() == self.host; + Box::pin(async move { + if !matches { + return Err( + Box::new(EgressError::Forbidden) as Box + ); + } + let mut vetted = (0..addresses.len()) + .map(move |index| addresses[index]) + .filter(move |address| policy.permits_ip(address.ip(), &denied)); + let first = vetted.next().ok_or(EgressError::Forbidden)?; + Ok(Box::new(std::iter::once(first).chain(vetted)) as Addrs) + }) + } +} + +/// A failed sealed send retains the destination/transport distinction. +#[derive(Debug, thiserror::Error)] +pub enum SendError { + #[error(transparent)] + Policy(#[from] EgressError), + #[error(transparent)] + Transport(#[from] reqwest::Error), + #[error("invalid request header name: {0}")] + HeaderName(#[from] InvalidHeaderName), + #[error("invalid request header value: {0}")] + HeaderValue(#[from] InvalidHeaderValue), + #[error("outbound request timed out")] + Timeout, +} + +impl SendError { + /// Only a proven forbidden destination is a terminal policy rejection. + pub fn is_retryable(&self) -> bool { + match self { + Self::Policy(error) => error.is_retryable(), + _ => true, + } + } +} + +/// A target-bound POST request. No raw client, URL setter, redirect policy or +/// proxy override is exposed; even literal hosts are rechecked at `send`. +pub struct EgressRequest { + policy: Arc, + destination: VettedDestination, + timeout: Duration, + headers: HeaderMap, + body: Option, + error: Option, +} + +/// Prepare a POST to exactly the URL carried by the vetted destination. +/// The actual connection is permitted only under a fresh, healthy read lease. +pub fn post( + policy: Arc, + destination: VettedDestination, + timeout: Duration, +) -> EgressRequest { + EgressRequest { + policy, + destination, + timeout, + headers: HeaderMap::new(), + body: None, + error: None, + } +} + +impl EgressRequest { + /// Set a header without exposing or changing the target URL. + pub fn header(mut self, name: &'static str, value: impl AsRef) -> Self { + if self.error.is_none() { + match HeaderName::from_bytes(name.as_bytes()) { + Ok(name) => match HeaderValue::from_str(value.as_ref()) { + Ok(value) => { + self.headers.insert(name, value); + } + Err(error) => self.error = Some(error.into()), + }, + Err(error) => self.error = Some(error.into()), + } + } + self + } + + /// Move the original body bytes into the request. + pub fn body(mut self, body: impl Into) -> Self { + if self.error.is_none() { + self.body = Some(body.into()); + } + self + } + + /// Recheck the bound destination and keep refresh from invalidating it + /// until the connection/request completes. Cancellation releases the lease. + pub async fn send(self) -> Result { + let timeout = self.timeout; + let started = Instant::now(); + tokio::time::timeout(timeout, self.send_under_lease(started)) + .await + .map_err(|_| SendError::Timeout)? + } + + async fn send_under_lease(self, started: Instant) -> Result { + if let Some(error) = self.error { + return Err(error); + } + let lease = self.policy.send_lease().await?; + if !self + .destination + .addresses + .iter() + .any(|address| self.policy.permits_ip(address.ip(), &lease.addresses)) + { + return Err(EgressError::Forbidden.into()); + } + let host = self + .destination + .url + .host_str() + .ok_or(EgressError::Forbidden)? + .to_owned(); + #[cfg(not(test))] + let https_only = true; + #[cfg(test)] + let https_only = !(self.policy.allow_loopback && self.destination.is_loopback()); + let builder = Client::builder() + .no_proxy() + .https_only(https_only) + .redirect(redirect::Policy::none()) + .connect_timeout(Duration::from_secs(2)) + .timeout(self.timeout); + #[cfg(test)] + let builder = self + .policy + .roots + .iter() + .cloned() + .fold(builder, |builder, root| builder.add_root_certificate(root)); + let client = builder + .dns_resolver(Arc::new(PolicyResolver { + host, + addresses: self.destination.addresses, + policy: self.policy.clone(), + denied: lease.addresses.clone(), + })) + .build()?; + let mut request = client.post(self.destination.url).headers(self.headers); + if let Some(body) = self.body { + request = request.body(body); + } + let result = request + .timeout(self.timeout.saturating_sub(started.elapsed())) + .send() + .await; + drop(lease); + result.map_err(SendError::Transport) + } +} + +/// Refresh the service deny-set every ten minutes and stop with Rocket. +pub async fn run_refresh(policy: Arc, shutdown: rocket::Shutdown) { + let mut interval = + tokio::time::interval_at(Instant::now() + REFRESH_INTERVAL, REFRESH_INTERVAL); + tokio::pin!(shutdown); + loop { + tokio::select! { + _ = interval.tick() => { + if policy.refresh().await.is_err() { + tracing::warn!("webhook egress closed: internal endpoint refresh failed"); + } + } + _ = &mut shutdown => break, + } + } +} + +#[cfg(test)] +pub(crate) mod test_support; +#[cfg(test)] +mod tests; diff --git a/tinycloud-node-server/src/egress/test_support.rs b/tinycloud-node-server/src/egress/test_support.rs new file mode 100644 index 00000000..58ca68b7 --- /dev/null +++ b/tinycloud-node-server/src/egress/test_support.rs @@ -0,0 +1,153 @@ +use super::{EgressError, LookupHost}; +use std::{ + collections::{HashMap, VecDeque}, + net::IpAddr, + sync::atomic::{AtomicUsize, Ordering}, +}; +use tokio::sync::Mutex; + +#[derive(Default)] +pub(crate) struct TestLookup { + answers: Mutex>>>, + pub calls: AtomicUsize, +} + +impl TestLookup { + pub async fn set(&self, host: &str, addresses: &[&str]) { + self.sequence(host, &[addresses]).await; + } + + pub async fn sequence(&self, host: &str, answers: &[&[&str]]) { + self.answers.lock().await.insert( + host.to_owned(), + answers + .iter() + .map(|addresses| addresses.iter().map(|ip| ip.parse().unwrap()).collect()) + .collect(), + ); + } + + pub fn call_count(&self) -> usize { + self.calls.load(Ordering::SeqCst) + } +} + +#[async_trait::async_trait] +impl LookupHost for TestLookup { + async fn lookup(&self, host: &str) -> Result, EgressError> { + self.calls.fetch_add(1, Ordering::SeqCst); + let mut answers = self.answers.lock().await; + let sequence = answers.get_mut(host).ok_or(EgressError::Dns)?; + if sequence.len() > 1 { + Ok(sequence.pop_front().unwrap()) + } else { + sequence.front().cloned().ok_or(EgressError::Dns) + } + } +} + +pub(crate) struct GatedLookup { + pub answers: TestLookup, + pub started: tokio::sync::Notify, + pub release: tokio::sync::Semaphore, + pub paused: std::sync::atomic::AtomicBool, +} + +impl Default for GatedLookup { + fn default() -> Self { + Self { + answers: TestLookup::default(), + started: tokio::sync::Notify::new(), + release: tokio::sync::Semaphore::new(0), + paused: std::sync::atomic::AtomicBool::new(false), + } + } +} + +#[async_trait::async_trait] +impl LookupHost for GatedLookup { + async fn lookup(&self, host: &str) -> Result, EgressError> { + if host == "db.test" && self.paused.load(Ordering::SeqCst) { + self.started.notify_one(); + self.release + .acquire() + .await + .map_err(|_| EgressError::Dns)? + .forget(); + } + self.answers.lookup(host).await + } +} + +pub(crate) struct CallbackServer { + pub url: String, + pub certificate: reqwest::Certificate, + pub requests: tokio::sync::mpsc::UnboundedReceiver<(Vec, hyper::HeaderMap)>, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for CallbackServer { + fn drop(&mut self) { + self.task.abort(); + } +} + +pub(crate) async fn callback_server( + status: u16, + location: Option, +) -> anyhow::Result { + callback_server_for_host("callback.test", status, location).await +} + +pub(crate) async fn callback_server_for_host( + host: &str, + status: u16, + location: Option, +) -> anyhow::Result { + use hyper::{service::service_fn, Body, Request, Response}; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let generated = rcgen::generate_simple_self_signed(vec![host.to_owned()])?; + let certificate = reqwest::Certificate::from_pem(generated.cert.pem().as_bytes())?; + let key = rustls::pki_types::PrivatePkcs8KeyDer::from(generated.key_pair.serialize_der()); + let tls = rustls::ServerConfig::builder() + .with_no_client_auth() + .with_single_cert(vec![generated.cert.der().clone()], key.into())?; + let acceptor = tokio_rustls::TlsAcceptor::from(std::sync::Arc::new(tls)); + let (sender, requests) = tokio::sync::mpsc::unbounded_channel(); + let task = tokio::spawn(async move { + while let Ok((stream, _)) = listener.accept().await { + let acceptor = acceptor.clone(); + let sender = sender.clone(); + let location = location.clone(); + tokio::spawn(async move { + let Ok(stream) = acceptor.accept(stream).await else { + return; + }; + let service = service_fn(move |request: Request| { + let sender = sender.clone(); + let location = location.clone(); + async move { + let (parts, body) = request.into_parts(); + let bytes = hyper::body::to_bytes(body).await.unwrap(); + let _ = sender.send((bytes.to_vec(), parts.headers)); + let mut response = Response::builder().status(status); + if let Some(location) = location { + response = response.header("Location", location); + } + Ok::<_, std::convert::Infallible>(response.body(Body::empty()).unwrap()) + } + }); + let _ = hyper::server::conn::Http::new() + .serve_connection(stream, service) + .await; + }); + } + }); + Ok(CallbackServer { + url: format!("https://{host}:{}/hooks", address.port()), + certificate, + requests, + task, + }) +} diff --git a/tinycloud-node-server/src/egress/tests.rs b/tinycloud-node-server/src/egress/tests.rs new file mode 100644 index 00000000..68e7d984 --- /dev/null +++ b/tinycloud-node-server/src/egress/tests.rs @@ -0,0 +1,571 @@ +use super::test_support::TestLookup; +use super::*; + +#[test] +fn tc1132_special_use_classes_and_public_boundaries() { + for (class, addresses) in [ + ("this_network", vec!["0.0.0.0", "0.255.255.255"]), + ( + "private", + vec![ + "10.0.0.1", + "172.16.0.1", + "172.31.255.255", + "192.168.1.1", + "64:ff9b:1:808:8:808:a00:0", + ], + ), + ("cgnat", vec!["100.64.0.1", "100.127.255.255"]), + ( + "loopback", + vec![ + "127.0.0.1", + "::1", + "::ffff:127.0.0.1", + "::127.0.0.1", + "2002:7f00:1::", + "64:ff9b:1:808:8:808:7f00:0", + ], + ), + ( + "link_local", + vec![ + "169.254.169.254", + "169.254.170.2", + "fe80::1", + "febf::1", + "64:ff9b::a9fe:a9fe", + ], + ), + ("ietf", vec!["192.0.0.1"]), + ( + "documentation", + vec!["192.0.2.1", "198.51.100.1", "203.0.113.1", "2001:db8::1"], + ), + ("benchmark", vec!["198.18.0.1", "198.19.255.255"]), + ("6to4_relay", vec!["192.88.99.1"]), + ("multicast", vec!["224.0.0.1", "239.255.255.255", "ff02::1"]), + ("reserved", vec!["240.0.0.1", "255.255.255.255"]), + ("unspecified", vec!["::"]), + ("site_local", vec!["fec0::1", "feff::1"]), + ("unique_local", vec!["fc00::1", "fd00:ec2::254"]), + ("discard", vec!["100::1"]), + ("teredo", vec!["2001::1"]), + ] { + for address in addresses { + assert_eq!( + EgressPolicy::forbidden(address.parse().unwrap()), + Some(class), + "{address}" + ); + } + } + for address in [ + "1.1.1.1", + "192.2.1.1", + "198.51.0.1", + "192.0.3.1", + "198.51.99.255", + "198.51.101.0", + "100.63.255.255", + "100.128.0.0", + "172.15.255.255", + "172.32.0.0", + "198.17.255.255", + "198.20.0.0", + "2606:4700::1111", + "::ffff:8.8.8.8", + "::8.8.8.8", + "64:ff9b::808:808", + "64:ff9b:1::808:808", + "64:ff9b:1:808:8:808:800:0", + "2002:808:808::", + ] { + assert_eq!( + EgressPolicy::forbidden(address.parse().unwrap()), + None, + "{address}" + ); + } + for address in [ + "64:ff9b:1::a00:1", + "64:ff9b:1:a00:0:100::", + "64:ff9b:1:10a:0:100::", + "64:ff9b:1:1:a:1::", + ] { + assert!( + EgressPolicy::forbidden(address.parse().unwrap()).is_some(), + "{address}" + ); + } +} + +#[tokio::test] +async fn tc1132_mixed_dns_exposes_only_vetted_addresses_to_reqwest() { + let lookup = Arc::new(TestLookup::default()); + lookup + .set( + "mixed.test", + &["10.0.0.1", "169.254.169.254", "1.1.1.1", "2606:4700::1111"], + ) + .await; + let policy = Arc::new(EgressPolicy::for_test(vec![], lookup.clone(), false)); + let destination = policy + .resolve_url(&Url::parse("https://mixed.test/").unwrap()) + .await + .unwrap(); + let denied = policy.deny.read().await.addresses.clone(); + let resolver = PolicyResolver { + host: destination.url().host_str().unwrap().to_owned(), + denied, + addresses: destination.addresses, + policy, + }; + let addresses: Vec<_> = resolver + .resolve("mixed.test".parse().unwrap()) + .await + .unwrap() + .collect(); + assert_eq!( + addresses, + vec![ + "1.1.1.1:443".parse::().unwrap(), + "[2606:4700::1111]:443".parse().unwrap() + ] + ); + assert_eq!(lookup.call_count(), 1, "no second DNS lookup at connect"); +} + +#[tokio::test] +async fn tc1132_database_and_other_endpoint_aliases_are_denied_on_every_port() { + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["1.1.1.1"]).await; + lookup.set("quota.test", &["8.8.8.8"]).await; + lookup.set("callback.test", &["1.1.1.1", "8.8.8.8"]).await; + let mut config = Config::default(); + config.storage.database = Some("postgres://user:password@db.test:5432/node".into()); + let mut policy = + EgressPolicy::from_config(&config, Some("https://quota.test/internal")).unwrap(); + policy.lookup = lookup; + policy.refresh().await.unwrap(); + for url in [ + "https://callback.test/", + "https://1.1.1.1:8443/", + "https://[64:ff9b::101:101]/", + "https://8.8.8.8/", + ] { + assert!( + matches!( + policy.resolve_url(&Url::parse(url).unwrap()).await, + Err(EgressError::Forbidden) + ), + "{url}" + ); + } +} + +#[tokio::test(start_paused = true)] +async fn tc1132_endpoint_refresh_is_periodic_fail_closed_and_recovers() { + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["1.1.1.1"]).await; + let policy = EgressPolicy::for_test(vec!["db.test".into()], lookup.clone(), false); + policy.refresh().await.unwrap(); + let old = Url::parse("https://1.1.1.1/").unwrap(); + let new = Url::parse("https://8.8.8.8/").unwrap(); + assert!(policy.resolve_url(&new).await.is_ok()); + lookup.set("db.test", &[]).await; + tokio::time::advance(REFRESH_INTERVAL).await; + assert!(matches!( + policy.resolve_url(&new).await, + Err(EgressError::Unavailable) + )); + assert!(matches!( + policy.resolve_url(&old).await, + Err(EgressError::Unavailable) + )); + lookup.set("db.test", &["8.8.8.8"]).await; + tokio::time::advance(REFRESH_INTERVAL).await; + assert!(matches!( + policy.resolve_url(&new).await, + Err(EgressError::Forbidden) + )); + assert!(policy.resolve_url(&old).await.is_ok()); +} + +struct StalledLookup; +#[async_trait::async_trait] +impl LookupHost for StalledLookup { + async fn lookup(&self, _: &str) -> Result, EgressError> { + std::future::pending().await + } +} + +#[tokio::test(start_paused = true)] +async fn tc1132_registration_dns_and_internal_refresh_have_one_second_deadlines() { + let policy = EgressPolicy::for_test(vec![], Arc::new(StalledLookup), false); + let start = Instant::now(); + let result = + crate::routes::hooks::validate_webhook_callback_url("https://slow.test/", &policy).await; + assert_eq!( + result.err().unwrap().into_http_error().0, + rocket::http::Status::BadRequest + ); + assert_eq!(start.elapsed(), DNS_TIMEOUT); + let policy = EgressPolicy::for_test(vec!["slow.test".into()], Arc::new(StalledLookup), false); + let start = Instant::now(); + assert!(matches!( + policy.refresh().await, + Err(EgressError::Unavailable) + )); + assert_eq!(start.elapsed(), DNS_TIMEOUT); + assert!(matches!( + policy + .resolve_url(&Url::parse("https://1.1.1.1/").unwrap()) + .await, + Err(EgressError::Unavailable) + )); +} + +#[tokio::test] +async fn tc1132_redirects_never_reach_another_listener() -> anyhow::Result<()> { + let target = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let mut callback = test_support::callback_server( + 302, + Some(format!("http://{}/redirected", target.local_addr()?)), + ) + .await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("callback.test", &["127.0.0.1"]).await; + let policy = Arc::new( + EgressPolicy::for_test(vec![], lookup, true).with_test_root(callback.certificate.clone()), + ); + let url = Url::parse(&callback.url)?; + let destination = policy.resolve_url(&url).await?; + let response = post(policy, destination, Duration::from_secs(3)) + .body("signed event") + .send() + .await?; + assert_eq!(response.status(), reqwest::StatusCode::FOUND); + assert_eq!(callback.requests.recv().await.unwrap().0, b"signed event"); + assert!( + tokio::time::timeout(Duration::from_millis(100), target.accept()) + .await + .is_err() + ); + Ok(()) +} + +#[tokio::test] +async fn tc1132_connect_time_policy_refuses_a_previously_vetted_now_internal_address( +) -> anyhow::Result<()> { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("callback.test", &["127.0.0.1"]).await; + lookup.set("db.test", &["1.1.1.1"]).await; + let policy = Arc::new(EgressPolicy::for_test( + vec!["db.test".into()], + lookup.clone(), + true, + )); + policy.refresh().await?; + let url = Url::parse(&format!( + "https://callback.test:{}/", + listener.local_addr()?.port() + ))?; + let destination = policy.resolve_url(&url).await?; + lookup.set("db.test", &["127.0.0.1"]).await; + policy.refresh().await?; + let error = post(policy, destination, Duration::from_secs(2)) + .send() + .await + .unwrap_err(); + assert!(error.to_string().contains("no permitted address")); + assert!( + tokio::time::timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err() + ); + Ok(()) +} + +#[tokio::test] +async fn tc1132_tls_connect_stall_is_bounded_before_total_timeout() -> anyhow::Result<()> { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let stall = tokio::spawn(async move { + let (_stream, _) = listener.accept().await.unwrap(); + tokio::time::sleep(Duration::from_secs(5)).await; + }); + let lookup = Arc::new(TestLookup::default()); + lookup.set("callback.test", &["127.0.0.1"]).await; + let policy = Arc::new(EgressPolicy::for_test(vec![], lookup, true)); + let url = Url::parse(&format!("https://callback.test:{}/", address.port()))?; + let destination = policy.resolve_url(&url).await?; + let start = Instant::now(); + let error = post(policy, destination, Duration::from_secs(10)) + .send() + .await + .unwrap_err(); + assert!( + matches!(&error, SendError::Transport(error) if error.is_timeout()), + "{error}" + ); + assert!(start.elapsed() >= Duration::from_millis(1800)); + assert!(start.elapsed() < Duration::from_secs(3)); + stall.abort(); + Ok(()) +} + +#[tokio::test] +async fn tc1132_literal_vetted_before_refresh_never_connects_after_becoming_internal( +) -> anyhow::Result<()> { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["192.2.1.1"]).await; + let policy = Arc::new(EgressPolicy::for_test( + vec!["db.test".into()], + lookup.clone(), + true, + )); + policy.refresh().await?; + let url = Url::parse(&format!( + "https://127.0.0.1:{}/hooks", + listener.local_addr()?.port() + ))?; + let destination = policy.resolve_url(&url).await?; + let request = post(policy.clone(), destination, Duration::from_secs(2)); + lookup.set("db.test", &["127.0.0.1"]).await; + policy.refresh().await?; + assert!(matches!( + policy.resolve_url(&url).await, + Err(EgressError::Forbidden) + )); + let error = request.send().await.unwrap_err(); + assert!( + error.to_string().contains("no permitted address"), + "{error}" + ); + assert!( + tokio::time::timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err() + ); + Ok(()) +} + +#[tokio::test] +async fn tc1132_send_rechecks_mixed_snapshot_without_contacting_newly_denied_address( +) -> anyhow::Result<()> { + let mut callback = test_support::callback_server(200, None).await?; + let url = Url::parse(&callback.url)?; + let denied = + tokio::net::TcpListener::bind(format!("127.0.0.2:{}", url.port().unwrap())).await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["192.2.1.1"]).await; + lookup + .set("callback.test", &["127.0.0.2", "127.0.0.1"]) + .await; + let policy = Arc::new( + EgressPolicy::for_test(vec!["db.test".into()], lookup.clone(), true) + .with_test_root(callback.certificate.clone()), + ); + policy.refresh().await?; + let destination = policy.resolve_url(&url).await?; + let request = post(policy.clone(), destination, Duration::from_secs(3)).body("same body"); + lookup.set("db.test", &["127.0.0.2"]).await; + policy.refresh().await?; + assert_eq!(request.send().await?.status(), reqwest::StatusCode::OK); + assert_eq!(callback.requests.recv().await.unwrap().0, b"same body"); + assert!( + tokio::time::timeout(Duration::from_millis(100), denied.accept()) + .await + .is_err() + ); + assert_eq!(lookup.call_count(), 3); + Ok(()) +} + +#[tokio::test] +async fn tc1132_cancelled_send_releases_lease_before_refresh_commits() -> anyhow::Result<()> { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["192.2.1.1"]).await; + let policy = Arc::new(EgressPolicy::for_test( + vec!["db.test".into()], + lookup.clone(), + true, + )); + policy.refresh().await?; + let url = Url::parse(&format!( + "https://127.0.0.1:{}/", + listener.local_addr()?.port() + ))?; + let destination = policy.resolve_url(&url).await?; + let send = tokio::spawn(post(policy.clone(), destination, Duration::from_secs(10)).send()); + let (_stream, _) = tokio::time::timeout(Duration::from_secs(1), listener.accept()).await??; + lookup.set("db.test", &["127.0.0.1"]).await; + let mut refresh = { + let policy = policy.clone(); + tokio::spawn(async move { policy.refresh().await }) + }; + let early = tokio::time::timeout(Duration::from_millis(100), &mut refresh).await; + send.abort(); + assert!(send.await.unwrap_err().is_cancelled()); + assert!( + early.is_err(), + "a refresh may not invalidate an active send lease" + ); + tokio::time::timeout(Duration::from_secs(1), refresh).await???; + assert!(matches!( + policy.resolve_url(&url).await, + Err(EgressError::Forbidden) + )); + Ok(()) +} + +#[tokio::test] +async fn tc1132_cancellation_while_waiting_for_refresh_opens_no_connection() -> anyhow::Result<()> { + use std::sync::atomic::Ordering; + use test_support::GatedLookup; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let lookup = Arc::new(GatedLookup::default()); + lookup.answers.set("db.test", &["192.2.1.1"]).await; + let policy = Arc::new(EgressPolicy::for_test( + vec!["db.test".into()], + lookup.clone(), + true, + )); + policy.refresh().await?; + let url = Url::parse(&format!( + "https://127.0.0.1:{}/", + listener.local_addr()?.port() + ))?; + let destination = policy.resolve_url(&url).await?; + lookup.paused.store(true, Ordering::SeqCst); + let refresh = { + let policy = policy.clone(); + tokio::spawn(async move { policy.refresh().await }) + }; + tokio::time::timeout(Duration::from_secs(1), lookup.started.notified()).await?; + let mut send = tokio::spawn(post(policy.clone(), destination, Duration::from_secs(10)).send()); + let early = tokio::time::timeout(Duration::from_millis(100), &mut send).await; + send.abort(); + assert!(send.await.unwrap_err().is_cancelled()); + lookup.paused.store(false, Ordering::SeqCst); + lookup.release.add_permits(1); + refresh.await??; + assert!(early.is_err()); + assert!(policy.resolve_url(&url).await.is_ok()); + assert!( + tokio::time::timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err() + ); + Ok(()) +} + +#[tokio::test(start_paused = true)] +async fn tc1132_total_send_deadline_includes_refresh_wait() -> anyhow::Result<()> { + use std::sync::atomic::Ordering; + use test_support::GatedLookup; + let lookup = Arc::new(GatedLookup::default()); + lookup.answers.set("db.test", &["192.2.1.1"]).await; + let policy = Arc::new(EgressPolicy::for_test( + vec!["db.test".into()], + lookup.clone(), + false, + )); + policy.refresh().await?; + let destination = policy + .resolve_url(&Url::parse("https://198.51.0.1/")?) + .await?; + lookup.paused.store(true, Ordering::SeqCst); + let refresh = { + let policy = policy.clone(); + tokio::spawn(async move { policy.refresh().await }) + }; + lookup.started.notified().await; + let start = Instant::now(); + let error = post(policy, destination, Duration::from_millis(250)) + .send() + .await + .unwrap_err(); + assert!(matches!(error, SendError::Timeout)); + assert_eq!(start.elapsed(), Duration::from_millis(250)); + lookup.paused.store(false, Ordering::SeqCst); + lookup.release.add_permits(1); + refresh.await??; + Ok(()) +} + +#[tokio::test(start_paused = true)] +async fn tc1132_cancelled_refresh_preserves_fresh_state_but_never_lifts_stale_failure( +) -> anyhow::Result<()> { + use std::sync::atomic::Ordering; + use test_support::GatedLookup; + let lookup = Arc::new(GatedLookup::default()); + lookup.answers.set("db.test", &["192.2.1.1"]).await; + let policy = Arc::new(EgressPolicy::for_test( + vec!["db.test".into()], + lookup.clone(), + false, + )); + policy.refresh().await?; + lookup.paused.store(true, Ordering::SeqCst); + let refresh = { + let policy = policy.clone(); + tokio::spawn(async move { policy.refresh().await }) + }; + lookup.started.notified().await; + refresh.abort(); + assert!(refresh.await.unwrap_err().is_cancelled()); + let callback = Url::parse("https://198.51.0.1/")?; + assert!(policy.resolve_url(&callback).await.is_ok()); + tokio::time::advance(REFRESH_INTERVAL).await; + lookup.paused.store(false, Ordering::SeqCst); + lookup.answers.set("db.test", &[]).await; + assert!(matches!( + policy.resolve_url(&callback).await, + Err(EgressError::Unavailable) + )); + Ok(()) +} + +#[tokio::test] +async fn tc1132_literal_tls_identity_and_healthy_delivery_survive_sealed_api() -> anyhow::Result<()> +{ + let mut callback = test_support::callback_server_for_host("127.0.0.1", 200, None).await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["192.2.1.1"]).await; + let policy = Arc::new( + EgressPolicy::for_test(vec!["db.test".into()], lookup.clone(), true) + .with_test_root(callback.certificate.clone()), + ); + policy.refresh().await?; + let url = Url::parse(&callback.url)?; + let destination = policy.resolve_url(&url).await?; + assert_eq!( + post(policy.clone(), destination, Duration::from_secs(2)) + .body("before refresh") + .send() + .await? + .status(), + reqwest::StatusCode::OK + ); + assert_eq!(callback.requests.recv().await.unwrap().0, b"before refresh"); + let destination = policy.resolve_url(&url).await?; + let request = + post(policy.clone(), destination, Duration::from_secs(2)).body("must not be sent"); + lookup.set("db.test", &["127.0.0.1"]).await; + policy.refresh().await?; + assert!(matches!( + request.send().await, + Err(SendError::Policy(EgressError::Forbidden)) + )); + assert!( + tokio::time::timeout(Duration::from_millis(100), callback.requests.recv()) + .await + .is_err() + ); + Ok(()) +} diff --git a/tinycloud-node-server/src/lib.rs b/tinycloud-node-server/src/lib.rs index 27d86000..7e7b8def 100644 --- a/tinycloud-node-server/src/lib.rs +++ b/tinycloud-node-server/src/lib.rs @@ -30,6 +30,7 @@ pub mod authorization; pub mod config; #[cfg(feature = "dstack")] pub mod dstack; +pub mod egress; pub mod hooks; pub mod invocation_replay; pub mod link; @@ -423,6 +424,7 @@ pub async fn app(config: &Figment) -> Result> { /// The deploy job invokes this on the candidate image before stopping the node. pub fn resolve_runtime_config(tinycloud_config: &Config) -> Result { let mut resolved = tinycloud_config.clone(); + resolved.hooks.validate().map_err(anyhow::Error::msg)?; resolved.storage.resolve(); resolved.share_email = resolved .share_email @@ -490,6 +492,14 @@ pub async fn app_with_control( control: Option, ) -> Result> { let tinycloud_config = resolve_runtime_config(tinycloud_config)?; + let quota_url = std::env::var("TINYCLOUD_QUOTA_URL").ok(); + let egress_policy = Arc::new(egress::EgressPolicy::from_config( + &tinycloud_config, + quota_url.as_deref(), + )?); + if egress_policy.refresh().await.is_err() { + ::tracing::warn!("webhook egress closed: internal endpoint lookup failed"); + } tinycloud_core::did_resolution::configure( tinycloud_core::did_resolution::DidResolutionPolicy { network_methods: tinycloud_config.did_resolution.network_methods, @@ -694,11 +704,8 @@ pub async fn app_with_control( // Billing shares TINYCLOUD_ADMIN_SECRET with the node (it calls the // node's admin API with it) and accepts it on quota fetches (TC-627). - let quota_cache = QuotaCache::new( - tinycloud_config.storage.limit, - std::env::var("TINYCLOUD_QUOTA_URL").ok(), - ) - .with_service_secret(std::env::var("TINYCLOUD_ADMIN_SECRET").ok()); + let quota_cache = QuotaCache::new(tinycloud_config.storage.limit, quota_url) + .with_service_secret(std::env::var("TINYCLOUD_ADMIN_SECRET").ok()); let invocation_replay_cache = node_replay_cache(&tinycloud); let replay_cleanup = invocation_replay_cache.clone(); // TC-341: the periodic sweep also reclaims rows beyond the lifetime cap. @@ -709,6 +716,7 @@ pub async fn app_with_control( tinycloud.clone(), tinycloud_config.hooks.clone(), webhook_encryption.clone(), + egress_policy.clone(), )?; spawn_webhook_dispatcher(webhook_dispatcher); @@ -718,6 +726,16 @@ pub async fn app_with_control( let telemetry_tinycloud = tinycloud.clone(); let rocket = rocket::custom(config) + .attach(AdHoc::on_liftoff("webhook-egress-refresh", { + let policy = egress_policy.clone(); + move |rocket| { + let policy = policy.clone(); + let shutdown = rocket.shutdown(); + Box::pin(async move { + tokio::spawn(egress::run_refresh(policy, shutdown)); + }) + } + })) .mount("/", routes) .attach(AdHoc::config::()) .attach(tracing::TracingFairing::new(&tinycloud_config.log.tracing)) @@ -765,6 +783,7 @@ pub async fn app_with_control( .manage(hook_runtime) .manage(signed_url_runtime) .manage(webhook_encryption) + .manage(egress_policy) .manage(rate_limiter) .manage(tee_context) .manage(encryption_service) diff --git a/tinycloud-node-server/src/quota.rs b/tinycloud-node-server/src/quota.rs index ed334f16..1150756f 100644 --- a/tinycloud-node-server/src/quota.rs +++ b/tinycloud-node-server/src/quota.rs @@ -234,6 +234,7 @@ impl QuotaCache { reqwest::Client::builder() .timeout(FETCH_TIMEOUT) .connect_timeout(CONNECT_TIMEOUT) + .redirect(reqwest::redirect::Policy::none()) .build() .expect("failed to build quota reqwest client") }); @@ -447,7 +448,13 @@ async fn fetch_remote( secret: Option<&str>, key: &str, ) -> Option { - let mut request = client.get(format!("{}/api/quota/{}", url, key)); + // Encode only the space key, not the operator-controlled base URL. + // Cache/override keys remain the original SpaceId string. + let mut request = client.get(format!( + "{}/api/quota/{}", + url, + percent_encoding::utf8_percent_encode(key, percent_encoding::NON_ALPHANUMERIC), + )); if let Some(secret) = secret { request = request.bearer_auth(secret); } @@ -516,7 +523,7 @@ async fn record_fetch( #[cfg(test)] mod test { use super::*; - use tokio::io::AsyncWriteExt; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn space(n: u8) -> SpaceId { format!("tinycloud:pkh:eip155:1:0x7BD63AA37326a64d458559F44432103e3d6eEDE9:s{n}") @@ -569,6 +576,143 @@ mod test { ); } + async fn read_request(stream: &mut tokio::net::TcpStream) -> String { + tokio::time::timeout(Duration::from_secs(5), async { + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + let len = stream.read(&mut buffer).await.expect("read request"); + assert_ne!(len, 0, "request ended before its headers"); + request.extend_from_slice(&buffer[..len]); + assert!(request.len() <= 65536, "request headers too large"); + } + String::from_utf8(request).expect("HTTP request") + }) + .await + .expect("request headers timed out") + } + + #[tokio::test] + async fn billing_path_keeps_special_character_space_id_in_one_segment() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind quota listener"); + let quota_url = format!("http://{}/billing", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept quota request"); + let request = read_request(&mut stream).await; + let body = r#"{"storage_limit_bytes":500}"#; + stream + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await + .expect("write quota response"); + request + }); + // The public constructor permits characters that the URI parser + // refuses. Encoding must protect every SpaceId without changing its key. + let sid = SpaceId::new( + space(1).did().to_owned(), + "a/b?c#d%2F../é".parse().expect("space name"), + ); + let key = sid.to_string(); + let cache = QuotaCache::new(Some(ByteUnit::Byte(100)), Some(quota_url)); + assert_eq!(cache.get_limit(&sid).await, Some(ByteUnit::Byte(500))); + let request = server.await.expect("quota server"); + assert_eq!( + request.lines().next(), + Some("GET /billing/api/quota/tinycloud%3Apkh%3Aeip155%3A1%3A0x7BD63AA37326a64d458559F44432103e3d6eEDE9%3Aa%2Fb%3Fc%23d%252F%2E%2E%2F%C3%A9 HTTP/1.1"), + "the complete space key must be one encoded path segment", + ); + assert_eq!(cache.remote_entry(&key).await, Some((Some(500), false))); + cache.set_limit(&sid, 700).await; + assert_eq!(cache.get_limit(&sid).await, Some(ByteUnit::Byte(700))); + assert_eq!(cache.get_override(&sid).await, Some(700)); + } + + #[tokio::test] + async fn billing_redirect_never_requests_target_or_forwards_bearer() { + for same_origin in [true, false] { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind quota listener"); + let quota_addr = listener.local_addr().unwrap(); + let target = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind redirect target"); + let target_addr = if same_origin { + quota_addr + } else { + target.local_addr().unwrap() + }; + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept quota request"); + let first_request = read_request(&mut stream).await; + stream + .write_all( + format!( + "HTTP/1.1 302 Found\r\nLocation: http://{target_addr}/redirected\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + ) + .as_bytes(), + ) + .await + .expect("write redirect"); + drop(stream); + let target = if same_origin { listener } else { target }; + let redirected_request = match tokio::time::timeout( + Duration::from_millis(500), + target.accept(), + ) + .await + { + Ok(Ok((mut stream, _))) => { + let request = read_request(&mut stream).await; + let body = r#"{"storage_limit_bytes":900}"#; + stream + .write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) + .as_bytes(), + ) + .await + .expect("write redirected response"); + Some(request) + } + Err(_) => None, + Ok(Err(err)) => panic!("redirect target accept failed: {err}"), + }; + (first_request, redirected_request) + }); + let cache = QuotaCache::new( + Some(ByteUnit::Byte(100)), + Some(format!("http://{quota_addr}")), + ) + .with_service_secret(Some("quota-test-service-secret".into())); + let sid = space(1); + let limit = cache.get_limit(&sid).await; + let (first_request, redirected_request) = server.await.expect("quota server"); + assert!(first_request + .to_ascii_lowercase() + .contains("\r\nauthorization: bearer quota-test-service-secret\r\n")); + assert!( + redirected_request.is_none(), + "no redirect target may receive a request or bearer (same_origin={same_origin}): {redirected_request:?}", + ); + assert_eq!(limit, Some(ByteUnit::Byte(100)), "redirect uses fallback"); + assert_eq!( + cache.remote_entry(&sid.to_string()).await, + Some((None, false)) + ); + } + } + #[test] fn account_totals_need_both_numbers_and_a_complete_billing_answer() { let account = |body: &str| { diff --git a/tinycloud-node-server/src/routes/hooks.rs b/tinycloud-node-server/src/routes/hooks.rs index 9ecb8b55..3175984b 100644 --- a/tinycloud-node-server/src/routes/hooks.rs +++ b/tinycloud-node-server/src/routes/hooks.rs @@ -1,6 +1,7 @@ use crate::{ authorization::AuthHeaderGetter, config::Config, + egress::{EgressPolicy, VettedDestination}, hooks::{ hook_scope_path, matches_scope, normalize_path_prefix_for_service, HookRuntime, HookSubscription, HookTicketClaims, HookTicketRequest, HookTicketResponse, @@ -431,6 +432,7 @@ pub async fn create_webhook( policy_v3: &State, replay: &State, config: &State, + egress: &State>, ) -> Result, (Status, String)> { let invocation = authorize_hook_request( "server.hooks.webhook_register.auth", @@ -440,7 +442,7 @@ pub async fn create_webhook( config, ) .await?; - let normalized = normalize_webhook_request(&request)?; + let normalized = normalize_webhook_request(&request, egress)?; if !is_hook_action_authorized(&invocation, &normalized, "tinycloud.hooks/register") { return Err(( Status::Forbidden, @@ -448,7 +450,9 @@ pub async fn create_webhook( )); } // Resolve the callback host only for an authorized caller. - let callback_url = validate_webhook_callback_url(&request.callback_url).await?; + let (callback_url, _) = validate_webhook_callback_url(&request.callback_url, egress) + .await + .map_err(WebhookCallbackError::into_http_error)?; let active_count = tinycloud .count_active_hook_subscriptions(&normalized.space) @@ -704,6 +708,7 @@ fn invocation_expiry(invocation: &InvocationInfo) -> Result Result { let path_prefix = normalize_path_prefix_for_service(&request.service, request.path_prefix.clone()); @@ -722,10 +727,15 @@ pub fn normalize_webhook_request( if request.secret.is_empty() { return Err((Status::BadRequest, "secret is required".to_string())); } + if request.secret.len() > 256 { + return Err((Status::BadRequest, "secret exceeds 256 bytes".to_string())); + } + validate_callback_length(&request.callback_url)?; let url = reqwest::Url::parse(&request.callback_url) .map_err(|e| (Status::BadRequest, format!("invalid callbackUrl: {e}")))?; - validate_webhook_url_shape(&url)?; + validate_callback_length(url.as_str())?; + validate_webhook_shape(&url, policy)?; Ok(subscription) } @@ -767,82 +777,82 @@ fn validate_webhook_url_shape(url: &reqwest::Url) -> Result<(), (Status, String) } pub(crate) fn blocked_webhook_ip(ip: IpAddr) -> bool { - match ip { - IpAddr::V4(ip) => { - let octets = ip.octets(); - let first = octets[0]; - let second = octets[1]; - (first == 0) - || (first == 10) - || (first == 100 && (64..=127).contains(&second)) - || (first == 127) - || (first == 169 && second == 254) - || (first == 172 && (16..=31).contains(&second)) - || (first == 192 && (second == 0 || second == 2 || second == 168)) - || (first == 192 && second == 88 && octets[2] == 99) - || (first == 198 && (second == 18 || second == 19 || second == 51)) - || (first == 203 && second == 0 && octets[2] == 113) - || first >= 224 - } - IpAddr::V6(ip) => { - let segments = ip.segments(); - ip.is_unspecified() - || ip.is_loopback() - || ip.is_multicast() - || (segments[0] & 0xfe00) == 0xfc00 - || (segments[0] & 0xffc0) == 0xfe80 - || (segments[0] == 0x2001 && segments[1] == 0x0db8) - || ip - .to_ipv4_mapped() - .is_some_and(|mapped| blocked_webhook_ip(mapped.into())) - } + EgressPolicy::forbidden(ip).is_some() +} + +fn validate_webhook_shape( + url: &reqwest::Url, + policy: &EgressPolicy, +) -> Result<(), (Status, String)> { + #[cfg(test)] + if policy.permits_test_http() + && matches!(url.scheme(), "http" | "https") + && url.host_str().is_some() + && url.username().is_empty() + && url.password().is_none() + && url.fragment().is_none() + { + return Ok(()); } + #[cfg(not(test))] + let _ = policy; + validate_webhook_url_shape(url) } +/// Callback shape errors are terminal; temporary DNS/policy errors remain typed +/// so delivery can retry without permitting a forbidden connection. +#[derive(Debug, thiserror::Error)] +pub enum WebhookCallbackError { + #[error("{0}")] + Invalid(String), + #[error("callbackUrl rejected: {0}")] + Destination(#[from] crate::egress::EgressError), +} + +impl WebhookCallbackError { + /// Whether the callback may succeed once a temporary failure clears. + pub fn is_retryable(&self) -> bool { + matches!(self, Self::Destination(error) if error.is_retryable()) + } + + /// Preserve the registration endpoint's existing bad-callback status. + pub fn into_http_error(self) -> (Status, String) { + (Status::BadRequest, self.to_string()) + } +} + +/// Validate a callback and resolve it once through the shared destination policy. +/// Callers must authorize registration before invoking this bounded DNS lookup. pub async fn validate_webhook_callback_url( callback_url: &str, -) -> Result { + policy: &EgressPolicy, +) -> Result<(reqwest::Url, VettedDestination), WebhookCallbackError> { + validate_callback_length(callback_url) + .map_err(|(_, message)| WebhookCallbackError::Invalid(message))?; let url = reqwest::Url::parse(callback_url) - .map_err(|e| (Status::BadRequest, format!("invalid callbackUrl: {e}")))?; + .map_err(|e| WebhookCallbackError::Invalid(format!("invalid callbackUrl: {e}")))?; + validate_callback_length(url.as_str()) + .map_err(|(_, message)| WebhookCallbackError::Invalid(message))?; + validate_webhook_shape(&url, policy) + .map_err(|(_, message)| WebhookCallbackError::Invalid(message))?; + let destination = policy.resolve_url(&url).await?; #[cfg(test)] - if url.scheme() == "http" - && url - .host_str() - .and_then(|host| host.trim_matches(['[', ']']).parse::().ok()) - .is_some_and(|ip| ip.is_loopback()) - { - return Ok(url); + if url.scheme() == "http" && !destination.is_loopback() { + return Err(WebhookCallbackError::Invalid( + "test HTTP callback must resolve only to loopback".into(), + )); } - validate_webhook_url_shape(&url)?; - let host = url.host_str().ok_or_else(|| { - ( - Status::BadRequest, - "callbackUrl host is required".to_string(), - ) - })?; - let port = url - .port_or_known_default() - .expect("HTTPS has a default port"); - let addresses = tokio::net::lookup_host((host, port)) - .await - .map_err(|_| { - ( - Status::BadRequest, - "callbackUrl host cannot be resolved".to_string(), - ) - })? - .collect::>(); - if addresses.is_empty() - || addresses - .iter() - .any(|address| blocked_webhook_ip(address.ip())) - { + Ok((url, destination)) +} + +fn validate_callback_length(callback_url: &str) -> Result<(), (Status, String)> { + if callback_url.len() > 2048 { return Err(( Status::BadRequest, - "callbackUrl resolves to a reserved address".to_string(), + "callbackUrl exceeds 2048 bytes".to_string(), )); } - Ok(url) + Ok(()) } pub fn webhook_response_from_model( @@ -1150,7 +1160,9 @@ mod tests { secret: "dev-secret".to_string(), }; - let normalized = normalize_webhook_request(&request).expect("valid webhook request"); + let policy = EgressPolicy::from_config(&Config::default(), None).unwrap(); + let normalized = + normalize_webhook_request(&request, &policy).expect("valid webhook request"); assert_eq!(normalized.path_prefix.as_deref(), Some("documents")); } @@ -1173,7 +1185,11 @@ mod tests { secret: "dev-secret".to_string(), }; assert!( - normalize_webhook_request(&request).is_err(), + normalize_webhook_request( + &request, + &EgressPolicy::from_config(&Config::default(), None).unwrap() + ) + .is_err(), "{callback_url}" ); } @@ -1196,6 +1212,506 @@ mod tests { } } + #[test] + fn tc1132_callback_shape_rejects_embedded_private_and_allows_public_controls() { + for address in [ + "::127.0.0.1", + "64:ff9b::a9fe:a9fe", + "64:ff9b:1::a00:1", + "2002:7f00:1::", + "2001::1", + "fec0::1", + ] { + let url = reqwest::Url::parse(&format!("https://[{address}]/hooks")).unwrap(); + assert_eq!( + validate_webhook_url_shape(&url).unwrap_err().0, + Status::BadRequest + ); + } + for address in ["192.2.1.1", "198.51.0.1", "1.1.1.1", "2606:4700::1111"] { + assert!(!blocked_webhook_ip(address.parse().unwrap()), "{address}"); + } + } + + #[tokio::test] + async fn tc1132_registration_enforces_callback_and_secret_byte_limits() -> Result<()> { + let fixture = + hook_route_fixture(&[("hooks", "kv/documents", "tinycloud.hooks/register")]).await?; + for (callback, secret) in [ + ( + format!("https://1.1.1.1/{}", "x".repeat(2049)), + "secret".to_string(), + ), + ("https://1.1.1.1/hooks".to_string(), "é".repeat(129)), + ( + format!("https://1.1.1.1/{}", "é".repeat(400)), + "secret".to_string(), + ), + ] { + let header = fixture.session_header(&[( + fixture.hooks("kv/documents")?, + &["tinycloud.hooks/register"], + )])?; + let response = fixture.client.post("/hooks/webhooks") + .header(rocket::http::Header::new("Authorization", header)) + .header(rocket::http::ContentType::JSON) + .body(serde_json::json!({ + "space": fixture.space.to_string(), "service": "kv", "pathPrefix": "documents", + "callbackUrl": callback, "secret": secret, + }).to_string()).dispatch().await; + assert_eq!(response.status(), Status::BadRequest); + } + assert_eq!(fixture.persisted_rows().await?[4], 0); + Ok(()) + } + + #[derive(Deserialize)] + struct RegisteredWebhook { + id: String, + } + + async fn tc1132_register_url( + fixture: &HookRouteFixture, + callback: &str, + secret: &str, + ) -> Result<(Status, String)> { + let header = fixture.session_header(&[( + fixture.hooks("kv/documents")?, + &["tinycloud.hooks/register"], + )])?; + let response = fixture + .client + .post("/hooks/webhooks") + .header(rocket::http::Header::new("Authorization", header)) + .header(rocket::http::ContentType::JSON) + .body( + serde_json::json!({ + "space": fixture.space.to_string(), "service": "kv", "pathPrefix": "documents", + "callbackUrl": callback, "secret": secret, + }) + .to_string(), + ) + .dispatch() + .await; + Ok(( + response.status(), + response.into_string().await.unwrap_or_default(), + )) + } + + async fn tc1132_queue(fixture: &HookRouteFixture, subscription: &str, id: &str) -> Result<()> { + use tinycloud_core::{models::hook_delivery, sea_orm::ActiveModelTrait}; + hook_delivery::ActiveModel::from(hook_delivery::Model { + id: id.into(), + subscription_id: subscription.into(), + event_id: "event_egress".into(), + payload_json: r#"{"id":"event_egress","key":"documents/note"}"#.into(), + status: tinycloud_core::db::HOOK_DELIVERY_STATUS_PENDING.into(), + attempts: 0, + next_attempt_at: None, + last_error: None, + delivered_at: None, + created_at: "2026-10-14T00:00:00Z".into(), + }) + .insert(&fixture.db) + .await?; + Ok(()) + } + + #[tokio::test] + async fn tc1132_registration_and_delivery_refuse_private_metadata_and_embedded_forms( + ) -> Result<()> { + use crate::egress::test_support::TestLookup; + use tinycloud_core::models::hook_delivery; + let lookup = std::sync::Arc::new(TestLookup::default()); + let policy = std::sync::Arc::new(EgressPolicy::for_test(vec![], lookup.clone(), false)); + let fixture = hook_route_fixture_with_policy( + &[("hooks", "kv/documents", "tinycloud.hooks/register")], + policy.clone(), + ) + .await?; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let callbacks = [ + ("https://callback.test/hooks", Some("127.0.0.1")), + ("https://callback.test/hooks", Some("10.0.0.1")), + ("https://callback.test/hooks", Some("169.254.169.254")), + ("https://callback.test/hooks", Some("::127.0.0.1")), + ("https://callback.test/hooks", Some("64:ff9b::a9fe:a9fe")), + ("https://callback.test/hooks", Some("2002:7f00:1::")), + ( + "https://callback.test/hooks", + Some("64:ff9b:1:808:8:808:7f00:0"), + ), + ("https://[::ffff:7f00:1]/hooks", None), + ("https://[64:ff9b:1::a00:1]/hooks", None), + ]; + for (index, (callback, answer)) in callbacks.iter().enumerate() { + if let Some(answer) = answer { + lookup.set("callback.test", &[answer]).await; + } + let (status, body) = tc1132_register_url(&fixture, callback, "secret").await?; + assert_eq!(status, Status::BadRequest, "{callback}: {body}"); + let id = format!("sub_egress_{index}"); + fixture + .tinycloud + .create_hook_subscription(hook_subscription::Model { + id: id.clone(), + subscriber_did: did_of(&fixture.session.1).into(), + space_id: fixture.space.to_string(), + target_service: "kv".into(), + path_prefix: Some("documents".into()), + abilities_json: None, + callback_url: (*callback).into(), + encrypted_secret: ColumnEncryption::new([3; 32]).encrypt(b"secret"), + secret_key_id: "primary".into(), + active: true, + created_at: "2026-10-14T00:00:00Z".into(), + }) + .await?; + let delivery_id = format!("delivery_egress_{index}"); + tc1132_queue(&fixture, &id, &delivery_id).await?; + crate::webhook_dispatcher::WebhookDispatcher::new( + fixture.tinycloud.clone(), + crate::config::HooksConfig::default(), + ColumnEncryption::new([3; 32]), + policy.clone(), + )? + .dispatch_due_once() + .await?; + let delivery = hook_delivery::Entity::find_by_id(delivery_id) + .one(&fixture.db) + .await? + .unwrap(); + assert_eq!( + delivery.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_DEAD_LETTER + ); + assert_eq!(delivery.attempts, 1); + assert!(delivery + .last_error + .unwrap() + .starts_with("callback URL rejected:")); + } + assert!( + tokio::time::timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn tc1132_rebinding_is_refused_after_successful_registration() -> Result<()> { + use crate::egress::test_support::TestLookup; + use tinycloud_core::models::hook_delivery; + let lookup = std::sync::Arc::new(TestLookup::default()); + lookup + .sequence("callback.test", &[&["1.1.1.1"], &["127.0.0.1"]]) + .await; + let policy = std::sync::Arc::new(EgressPolicy::for_test(vec![], lookup.clone(), false)); + let fixture = hook_route_fixture_with_policy( + &[("hooks", "kv/documents", "tinycloud.hooks/register")], + policy.clone(), + ) + .await?; + let (status, body) = + tc1132_register_url(&fixture, "https://callback.test/hooks", "secret").await?; + assert_eq!(status, Status::Ok, "{body}"); + let subscription: RegisteredWebhook = serde_json::from_str(&body)?; + tc1132_queue(&fixture, &subscription.id, "delivery_rebind").await?; + crate::webhook_dispatcher::WebhookDispatcher::new( + fixture.tinycloud.clone(), + crate::config::HooksConfig::default(), + ColumnEncryption::new([3; 32]), + policy, + )? + .dispatch_due_once() + .await?; + let delivery = hook_delivery::Entity::find_by_id("delivery_rebind") + .one(&fixture.db) + .await? + .unwrap(); + assert_eq!( + delivery.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_DEAD_LETTER + ); + assert!(delivery + .last_error + .unwrap() + .starts_with("callback URL rejected:")); + assert_eq!(lookup.call_count(), 2); + Ok(()) + } + + #[tokio::test] + async fn tc1132_registration_accepts_corrected_public_ranges_and_exact_byte_limits( + ) -> Result<()> { + for callback in [ + "https://192.2.1.1/hooks".to_string(), + "https://198.51.0.1/hooks".to_string(), + format!( + "https://1.1.1.1/{}", + "x".repeat(2048 - "https://1.1.1.1/".len()) + ), + ] { + let fixture = + hook_route_fixture(&[("hooks", "kv/documents", "tinycloud.hooks/register")]) + .await?; + let secret = "é".repeat(128); + let (status, body) = tc1132_register_url(&fixture, &callback, &secret).await?; + assert_eq!(status, Status::Ok, "{body}"); + let subscription: RegisteredWebhook = serde_json::from_str(&body)?; + let stored = fixture + .tinycloud + .find_hook_subscription(&subscription.id) + .await? + .unwrap(); + assert_eq!(stored.callback_url, callback); + assert_eq!( + ColumnEncryption::new([3; 32]).decrypt(&stored.encrypted_secret)?, + secret.as_bytes() + ); + } + Ok(()) + } + + #[tokio::test] + async fn tc1132_unauthorized_registration_never_resolves_callback() -> Result<()> { + use crate::egress::test_support::TestLookup; + let lookup = std::sync::Arc::new(TestLookup::default()); + let policy = std::sync::Arc::new(EgressPolicy::for_test(vec![], lookup.clone(), false)); + let fixture = hook_route_fixture_with_policy( + &[("hooks", "kv/documents", "tinycloud.hooks/subscribe")], + policy, + ) + .await?; + let (status, _) = + tc1132_register_url(&fixture, "https://callback.test/hooks", "secret").await?; + assert_eq!(status, Status::Unauthorized); + assert_eq!(lookup.call_count(), 0); + Ok(()) + } + + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn tc1132_registered_delivery_uses_vetted_dns_without_proxy_and_preserves_hmac( + ) -> Result<()> { + use crate::{ + egress::test_support::{callback_server, TestLookup}, + test_support::{env_lock, EnvGuard}, + }; + use tinycloud_core::models::hook_delivery; + #[allow(clippy::await_holding_lock)] + let _lock = env_lock(); + let proxy = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let _proxy = EnvGuard::set("HTTPS_PROXY", format!("http://{}", proxy.local_addr()?)); + let _no_proxy = EnvGuard::set("NO_PROXY", ""); + let _no_proxy_lower = EnvGuard::set("no_proxy", ""); + let mut callback = callback_server(200, None).await?; + let lookup = std::sync::Arc::new(TestLookup::default()); + lookup + .set("callback.test", &["10.0.0.1", "127.0.0.1"]) + .await; + let policy = std::sync::Arc::new( + EgressPolicy::for_test(vec![], lookup.clone(), true) + .with_test_root(callback.certificate.clone()), + ); + let fixture = hook_route_fixture_with_policy( + &[("hooks", "kv/documents", "tinycloud.hooks/register")], + policy.clone(), + ) + .await?; + let (status, body) = tc1132_register_url(&fixture, &callback.url, "secret").await?; + assert_eq!(status, Status::Ok, "{body}"); + let subscription: RegisteredWebhook = serde_json::from_str(&body)?; + tc1132_queue(&fixture, &subscription.id, "delivery_tls").await?; + crate::webhook_dispatcher::WebhookDispatcher::new( + fixture.tinycloud.clone(), + crate::config::HooksConfig::default(), + ColumnEncryption::new([3; 32]), + policy, + )? + .dispatch_due_once() + .await?; + let (body, headers) = + tokio::time::timeout(Duration::from_secs(2), callback.requests.recv()) + .await? + .unwrap(); + assert_eq!(body, br#"{"id":"event_egress","key":"documents/note"}"#); + let mut mac = as hmac::Mac>::new_from_slice(b"secret")?; + hmac::Mac::update(&mut mac, &body); + let signature = format!( + "sha256={}", + hex::encode(hmac::Mac::finalize(mac).into_bytes()) + ); + assert_eq!(headers["X-TinyCloud-Signature"], signature); + assert_eq!(headers["X-TinyCloud-Delivery-Id"], "delivery_tls"); + assert_eq!(headers["X-TinyCloud-Event-Id"], "event_egress"); + assert_eq!(headers["X-TinyCloud-Subscription-Id"], subscription.id); + let delivery = hook_delivery::Entity::find_by_id("delivery_tls") + .one(&fixture.db) + .await? + .unwrap(); + assert_eq!( + delivery.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_DELIVERED + ); + assert_eq!(lookup.call_count(), 2, "registration once, delivery once"); + assert!( + tokio::time::timeout(Duration::from_millis(100), proxy.accept()) + .await + .is_err() + ); + Ok(()) + } + + #[tokio::test] + async fn tc1132_healthy_refresh_preserves_authorized_delivery() -> Result<()> { + use crate::egress::test_support::{callback_server, GatedLookup}; + use std::sync::{atomic::Ordering, Arc}; + use tinycloud_core::models::hook_delivery; + let mut callback = callback_server(200, None).await?; + let lookup = Arc::new(GatedLookup::default()); + lookup.answers.set("db.test", &["192.2.1.1"]).await; + lookup.answers.set("callback.test", &["127.0.0.1"]).await; + let policy = Arc::new( + EgressPolicy::for_test(vec!["db.test".into()], lookup.clone(), true) + .with_test_root(callback.certificate.clone()), + ); + policy.refresh().await?; + let fixture = hook_route_fixture_with_policy( + &[("hooks", "kv/documents", "tinycloud.hooks/register")], + policy.clone(), + ) + .await?; + let (status, body) = tc1132_register_url(&fixture, &callback.url, "refresh-secret").await?; + assert_eq!(status, Status::Ok, "{body}"); + let subscription: RegisteredWebhook = serde_json::from_str(&body)?; + tc1132_queue(&fixture, &subscription.id, "delivery_healthy_refresh").await?; + lookup.paused.store(true, Ordering::SeqCst); + let refresh = { + let policy = policy.clone(); + tokio::spawn(async move { policy.refresh().await }) + }; + tokio::time::timeout(Duration::from_secs(1), lookup.started.notified()).await?; + let dispatcher = crate::webhook_dispatcher::WebhookDispatcher::new( + fixture.tinycloud.clone(), + HooksConfig::default(), + ColumnEncryption::new([3; 32]), + policy.clone(), + )?; + let mut dispatch = tokio::spawn(async move { dispatcher.dispatch_due_once().await }); + let early = tokio::time::timeout(Duration::from_millis(100), &mut dispatch).await; + let pending = hook_delivery::Entity::find_by_id("delivery_healthy_refresh") + .one(&fixture.db) + .await? + .unwrap(); + lookup.paused.store(false, Ordering::SeqCst); + lookup.release.add_permits(1); + refresh.await??; + if let Ok(result) = early { + result??; + } else { + dispatch.await??; + } + assert_eq!( + pending.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_PENDING + ); + assert_eq!(pending.attempts, 0); + let (body, _) = tokio::time::timeout(Duration::from_secs(2), callback.requests.recv()) + .await? + .unwrap(); + assert_eq!(body, br#"{"id":"event_egress","key":"documents/note"}"#); + let delivered = hook_delivery::Entity::find_by_id("delivery_healthy_refresh") + .one(&fixture.db) + .await? + .unwrap(); + assert_eq!( + delivered.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_DELIVERED + ); + assert_eq!(delivered.attempts, 1); + Ok(()) + } + + #[tokio::test] + async fn tc1132_temporary_policy_failure_retries_and_recovers_with_original_payload( + ) -> Result<()> { + use crate::egress::test_support::{callback_server, TestLookup}; + use std::sync::Arc; + use tinycloud_core::{ + models::hook_delivery, + sea_orm::{ActiveModelTrait, ActiveValue::Set, IntoActiveModel}, + }; + let mut callback = callback_server(200, None).await?; + let lookup = Arc::new(TestLookup::default()); + lookup.set("db.test", &["192.2.1.1"]).await; + lookup.set("callback.test", &["127.0.0.1"]).await; + let policy = Arc::new( + EgressPolicy::for_test(vec!["db.test".into()], lookup.clone(), true) + .with_test_root(callback.certificate.clone()), + ); + policy.refresh().await?; + let fixture = hook_route_fixture_with_policy( + &[("hooks", "kv/documents", "tinycloud.hooks/register")], + policy.clone(), + ) + .await?; + let (status, body) = + tc1132_register_url(&fixture, &callback.url, "retry-policy-secret").await?; + assert_eq!(status, Status::Ok, "{body}"); + let subscription: RegisteredWebhook = serde_json::from_str(&body)?; + tc1132_queue(&fixture, &subscription.id, "delivery_policy_retry").await?; + lookup.set("db.test", &[]).await; + assert!(policy.refresh().await.is_err()); + let dispatcher = crate::webhook_dispatcher::WebhookDispatcher::new( + fixture.tinycloud.clone(), + HooksConfig::default(), + ColumnEncryption::new([3; 32]), + policy.clone(), + )?; + let before = OffsetDateTime::now_utc(); + dispatcher.dispatch_due_once().await?; + let row = hook_delivery::Entity::find_by_id("delivery_policy_retry") + .one(&fixture.db) + .await? + .unwrap(); + assert_eq!( + row.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_RETRYING + ); + assert_eq!(row.attempts, 1); + let next = OffsetDateTime::parse(row.next_attempt_at.as_deref().unwrap(), &Rfc3339)?; + assert!((30..=32).contains(&(next - before).whole_seconds())); + assert!( + callback.requests.try_recv().is_err(), + "failed policy must emit no request" + ); + lookup.set("db.test", &["192.2.1.1"]).await; + policy.refresh().await?; + let mut due = row.into_active_model(); + due.next_attempt_at = Set(Some("2000-01-01T00:00:00Z".into())); + due.update(&fixture.db).await?; + dispatcher.dispatch_due_once().await?; + let (body, headers) = + tokio::time::timeout(Duration::from_secs(2), callback.requests.recv()) + .await? + .unwrap(); + assert_eq!(body, br#"{"id":"event_egress","key":"documents/note"}"#); + assert_eq!(headers["X-TinyCloud-Delivery-Id"], "delivery_policy_retry"); + let row = hook_delivery::Entity::find_by_id("delivery_policy_retry") + .one(&fixture.db) + .await? + .unwrap(); + assert_eq!( + row.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_DELIVERED + ); + assert_eq!(row.attempts, 2); + Ok(()) + } + #[tokio::test] async fn converts_subscription_model_without_secret() { let model = hook_subscription::Model { @@ -1498,6 +2014,17 @@ mod tests { /// ability)` on the owner's space, plus a second owner's space for /// cross-space attempts. async fn hook_route_fixture(grants: &[(&str, &str, &str)]) -> Result { + hook_route_fixture_with_policy( + grants, + std::sync::Arc::new(EgressPolicy::from_config(&Config::default(), None)?), + ) + .await + } + + async fn hook_route_fixture_with_policy( + grants: &[(&str, &str, &str)], + policy: std::sync::Arc, + ) -> Result { use tinycloud_core::{ models::{abilities, actor, space}, sea_orm::{ActiveModelTrait, ActiveValue::Set}, @@ -1582,6 +2109,7 @@ mod tests { .manage(ColumnEncryption::new([3u8; 32])) .manage(PolicyV3Runtime::new(db.clone(), node.node_did(), node)) .manage(InvocationReplayCache::new(db.clone())) + .manage(policy) .manage(Config::default()); Ok(HookRouteFixture { diff --git a/tinycloud-node-server/src/routes/mod.rs b/tinycloud-node-server/src/routes/mod.rs index 7a6f52de..94cf9f1d 100644 --- a/tinycloud-node-server/src/routes/mod.rs +++ b/tinycloud-node-server/src/routes/mod.rs @@ -10984,6 +10984,7 @@ mod tests { rocket::routes![hooks::create_webhook, hooks::delete_webhook], ) .manage(ColumnEncryption::new(F43_WEBHOOK_KEY)) + .manage(crate::egress::EgressPolicy::test_loopback()) }) .await?; let tinycloud = f @@ -10996,6 +10997,7 @@ mod tests { tinycloud, HooksConfig::default(), ColumnEncryption::new(F43_WEBHOOK_KEY), + crate::egress::EgressPolicy::test_loopback(), )?; Ok((f, dispatcher)) } diff --git a/tinycloud-node-server/src/tunnel/connection.rs b/tinycloud-node-server/src/tunnel/connection.rs index 92021bde..6a8c4def 100644 --- a/tinycloud-node-server/src/tunnel/connection.rs +++ b/tinycloud-node-server/src/tunnel/connection.rs @@ -52,6 +52,10 @@ const AUTH_ACK_TIMEOUT: Duration = Duration::from_secs(10); /// heartbeats are tolerated before we give up on the socket. const IDLE_READ_TIMEOUT: Duration = Duration::from_secs(90); +/// Total budget for a loopback HTTP request, including its response body. +/// A stalled upstream must release its request task without closing the tunnel. +const UPSTREAM_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); + /// Default cap on the number of requests concurrently being reassembled /// (head received, body not yet complete) per tunnel connection. Bounds /// `pending`'s memory usage against a relay that opens many requests without @@ -455,7 +459,17 @@ async fn serve_requests( ) -> AttemptResult { let mut pending: HashMap = HashMap::new(); let upstream_addr = ctx.upstream_addr; - let http_client = reqwest::Client::new(); + let http_client = match reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .timeout(UPSTREAM_REQUEST_TIMEOUT) + .build() + { + Ok(client) => client, + Err(err) => { + tracing::warn!(%err, "tunnel: failed to build upstream HTTP client"); + return AttemptResult::Outcome(AttemptOutcome::TransportError); + } + }; let max_pending = max_pending_requests(); loop { diff --git a/tinycloud-node-server/src/webhook_dispatcher.rs b/tinycloud-node-server/src/webhook_dispatcher.rs index fb70859b..b1d52b59 100644 --- a/tinycloud-node-server/src/webhook_dispatcher.rs +++ b/tinycloud-node-server/src/webhook_dispatcher.rs @@ -1,9 +1,12 @@ -use crate::{config::HooksConfig, TinyCloud}; +use crate::{ + config::HooksConfig, + egress::{self, EgressPolicy}, + TinyCloud, +}; use anyhow::{Context, Result}; use hmac::{Hmac, Mac}; -use reqwest::{redirect::Policy, Client, Url}; use sha2::Sha256; -use std::{collections::HashMap, time::Duration}; +use std::{collections::HashMap, sync::Arc, time::Duration}; use time::{Duration as TimeDuration, OffsetDateTime}; use tinycloud_core::ColumnEncryption; @@ -19,6 +22,7 @@ pub struct WebhookDispatcher { encryption: ColumnEncryption, max_attempts: i64, timeout: Duration, + egress: Arc, } impl WebhookDispatcher { @@ -26,12 +30,15 @@ impl WebhookDispatcher { tinycloud: TinyCloud, config: HooksConfig, encryption: ColumnEncryption, + egress: Arc, ) -> Result { + config.validate().map_err(anyhow::Error::msg)?; Ok(Self { tinycloud, encryption, max_attempts: config.webhook_max_attempts.max(1) as i64, timeout: Duration::from_secs(config.webhook_timeout_seconds.max(1)), + egress, }) } @@ -130,32 +137,36 @@ impl WebhookDispatcher { }; let signature = sign_payload(&secret, delivery.payload_json.as_bytes())?; - let url = crate::routes::hooks::validate_webhook_callback_url(&delivery.callback_url) - .await - .map_err(|(_, message)| anyhow::anyhow!(message)); - let client = match url { - Ok(url) => self - .pinned_client(&url) - .context("building pinned webhook client")?, + let destination = match crate::routes::hooks::validate_webhook_callback_url( + &delivery.callback_url, + &self.egress, + ) + .await + { + Ok((_, destination)) => destination, Err(error) => { - self.record_terminal_failure( - delivery.id, - current_attempt, - &format!("callback URL rejected: {error}"), - ) - .await?; + if error.is_retryable() { + self.record_retryable_failure(delivery.id, current_attempt, error.to_string()) + .await?; + } else { + self.record_terminal_failure( + delivery.id, + current_attempt, + &format!("callback URL rejected: {error}"), + ) + .await?; + } return Ok(()); } }; - match client - .post(&delivery.callback_url) + match egress::post(self.egress.clone(), destination, self.timeout) .header("Content-Type", "application/json") .header("X-TinyCloud-Delivery-Id", &delivery.id) .header("X-TinyCloud-Event-Id", &delivery.event_id) .header("X-TinyCloud-Subscription-Id", &delivery.subscription_id) .header("X-TinyCloud-Signature", signature) - .body(delivery.payload_json.clone()) + .body(delivery.payload_json) .send() .await { @@ -174,34 +185,23 @@ impl WebhookDispatcher { .await?; } Err(error) => { - self.record_retryable_failure(delivery.id, current_attempt, error.to_string()) + if error.is_retryable() { + self.record_retryable_failure(delivery.id, current_attempt, error.to_string()) + .await?; + } else { + self.record_terminal_failure( + delivery.id, + current_attempt, + &format!("callback URL rejected: {error}"), + ) .await?; + } } } Ok(()) } - fn pinned_client(&self, url: &Url) -> Result { - let host = url.host_str().context("callback URL host missing")?; - let port = url - .port_or_known_default() - .context("callback URL port missing")?; - let address = std::net::ToSocketAddrs::to_socket_addrs(&(host, port)) - .context("resolving callback URL")? - .find(|address| { - !crate::routes::hooks::blocked_webhook_ip(address.ip()) - || (cfg!(test) && address.ip().is_loopback()) - }) - .context("callback URL has no public address")?; - Client::builder() - .timeout(self.timeout) - .redirect(Policy::none()) - .resolve(host, address) - .build() - .context("building pinned callback client") - } - async fn record_terminal_failure( &self, delivery_id: String, @@ -263,7 +263,7 @@ fn retry_delay_for_attempt(attempt_number: i64) -> TimeDuration { 2 => TimeDuration::seconds(30), 3 => TimeDuration::minutes(2), 4 => TimeDuration::minutes(10), - 5 => TimeDuration::hours(1), + 5.. => TimeDuration::hours(1), _ => TimeDuration::seconds(0), } } @@ -291,6 +291,7 @@ mod tests { service::{make_service_fn, service_fn}, Body, Request, Response, Server, StatusCode, }; + use reqwest::Url; use std::{convert::Infallible, net::TcpListener}; use tempfile::TempDir; use tinycloud_core::{ @@ -437,12 +438,79 @@ mod tests { Ok(()) } + #[test] + fn tc1132_later_attempts_keep_hour_backoff() { + for attempt in [5, 6, 7, 10, i64::MAX] { + assert_eq!( + retry_delay_for_attempt(attempt), + TimeDuration::hours(1), + "{attempt}" + ); + } + } + + #[tokio::test] + async fn tc1132_dispatcher_rejects_excessive_attempt_config() -> Result<()> { + let (tinycloud, _, encryption) = test_tinycloud().await?; + assert!(WebhookDispatcher::new( + tinycloud, + HooksConfig { + webhook_max_attempts: 11, + ..HooksConfig::default() + }, + encryption, + EgressPolicy::test_loopback() + ) + .is_err()); + Ok(()) + } + + #[tokio::test] + async fn tc1132_sixth_failure_schedules_hour_backoff() -> Result<()> { + let (tinycloud, db, encryption) = test_tinycloud().await?; + let (callback, _receiver) = + spawn_callback_server(StatusCode::INTERNAL_SERVER_ERROR).await?; + let id = insert_pending_delivery(&db, &encryption, &callback, 5).await?; + let before = OffsetDateTime::now_utc(); + WebhookDispatcher::new( + tinycloud, + HooksConfig { + webhook_max_attempts: 10, + ..HooksConfig::default() + }, + encryption, + EgressPolicy::test_loopback(), + )? + .dispatch_due_once() + .await?; + let row = hook_delivery::Entity::find_by_id(id) + .one(&db) + .await? + .unwrap(); + assert_eq!( + row.status, + tinycloud_core::db::HOOK_DELIVERY_STATUS_RETRYING + ); + assert_eq!(row.attempts, 6); + let next = OffsetDateTime::parse( + row.next_attempt_at.as_deref().unwrap(), + &time::format_description::well_known::Rfc3339, + )?; + assert!((next - before).whole_seconds() >= 3600); + assert!((next - before).whole_seconds() <= 3602); + Ok(()) + } #[tokio::test] async fn dispatches_due_delivery_and_marks_it_delivered() -> Result<()> { let (tinycloud, db, encryption) = test_tinycloud().await?; let (callback_url, mut receiver) = spawn_callback_server(StatusCode::OK).await?; let delivery_id = insert_pending_delivery(&db, &encryption, &callback_url, 0).await?; - let dispatcher = WebhookDispatcher::new(tinycloud, HooksConfig::default(), encryption)?; + let dispatcher = WebhookDispatcher::new( + tinycloud, + HooksConfig::default(), + encryption, + EgressPolicy::test_loopback(), + )?; dispatcher.dispatch_due_once().await?; @@ -483,6 +551,7 @@ mod tests { ..HooksConfig::default() }, encryption, + EgressPolicy::test_loopback(), )?; dispatcher.dispatch_due_once().await?; @@ -547,7 +616,12 @@ mod tests { ) .await? .with_encryption(Some(encryption.clone())); - let dispatcher = WebhookDispatcher::new(tinycloud, HooksConfig::default(), encryption)?; + let dispatcher = WebhookDispatcher::new( + tinycloud, + HooksConfig::default(), + encryption, + EgressPolicy::test_loopback(), + )?; dispatcher.dispatch_due_once().await?; @@ -808,6 +882,7 @@ mod tests { self.tinycloud.clone(), HooksConfig::default(), self.encryption.clone(), + EgressPolicy::test_loopback(), ) .unwrap() } diff --git a/tinycloud-node-server/tests/tunnel_service.rs b/tinycloud-node-server/tests/tunnel_service.rs index 472ab434..57639311 100644 --- a/tinycloud-node-server/tests/tunnel_service.rs +++ b/tinycloud-node-server/tests/tunnel_service.rs @@ -51,6 +51,7 @@ use tinycloud::{ protocol::{TunnelFrame, BODY_CHUNK_BYTES, CLOSE_STALE_SEQUENCE, CLOSE_SUPERSEDED}, }, }; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::{oneshot, watch}; /// Serializes tests that mutate the process-wide `TINYCLOUD_KEYS_SECRET` env @@ -1245,3 +1246,250 @@ impl Drop for EnvVarGuard { } } } + +/// Runs the real tunnel connection against a local relay, recording complete +/// response/error frame sequences for each request on the same WebSocket. +async fn proxy_local_requests( + upstream_addr: SocketAddr, + paths: Vec<&'static str>, + deadline: Duration, +) -> Vec> { + let observations = Arc::new(Mutex::new(RelayObservations::default())); + let (result_tx, result_rx) = oneshot::channel(); + let result_tx = Arc::new(tokio::sync::Mutex::new(Some(result_tx))); + let ws_handler = move |ws: WebSocketUpgrade, Path(_name): Path| { + let obs = observations.clone(); + let result_tx = result_tx.clone(); + let paths = paths.clone(); + async move { + ws.on_upgrade(move |mut socket| async move { + recv_auth_frame(&mut socket, &obs).await; + send_ack(&mut socket).await; + let mut results = Vec::new(); + for (index, path) in paths.into_iter().enumerate() { + let id = format!("req-{index}"); + for frame in [ + TunnelFrame::Request { + id: id.clone(), + method: "GET".into(), + path: path.into(), + headers: vec![( + "authorization".into(), + "Bearer tunnel-test-credential".into(), + )], + }, + TunnelFrame::RequestBody { + id, + chunk: String::new(), + done: true, + }, + ] { + socket + .send(WsMessage::Text(frame.encode())) + .await + .expect("send relay request"); + } + let mut frames = Vec::new(); + loop { + let message = socket.recv().await.expect("tunnel closed").unwrap(); + let WsMessage::Text(text) = message else { + continue; + }; + let frame = TunnelFrame::parse(&text).expect("tunnel frame"); + let done = matches!( + frame, + TunnelFrame::ResponseBody { done: true, .. } + | TunnelFrame::Error { .. } + ); + frames.push(frame); + if done { + break; + } + } + results.push(frames); + } + if let Some(tx) = result_tx.lock().await.take() { + let _ = tx.send(results); + } + let _ = socket.recv().await; + }) + } + }; + let app = Router::new().route("/v1/tunnel/:name", get(ws_handler)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let relay_addr = listener.local_addr().unwrap(); + let relay = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let data_root = tempdir().unwrap(); + seed_enabled_tunnel_state( + data_root.path(), + "egressnode", + &format!("http://{relay_addr}"), + ); + let (shutdown_tx, handle) = spawn_tunnel_client(data_root.path(), upstream_addr).await; + let result = tokio::time::timeout(deadline, result_rx).await; + stop_and_join(shutdown_tx, handle).await; + relay.abort(); + result + .expect("tunnel did not complete requests within the deadline") + .expect("relay dropped observations") +} + +async fn read_upstream_request(stream: &mut tokio::net::TcpStream) -> String { + tokio::time::timeout(Duration::from_secs(5), async { + let mut request = Vec::new(); + let mut buffer = [0; 1024]; + while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") { + let len = stream.read(&mut buffer).await.expect("read request"); + assert_ne!(len, 0, "request ended before headers"); + request.extend_from_slice(&buffer[..len]); + assert!(request.len() <= 65536); + } + String::from_utf8(request).unwrap() + }) + .await + .expect("upstream request headers timed out") +} + +#[tokio::test] +async fn tc1132_redirect_is_returned_without_requesting_target() { + let _env_lock = env_lock().await; + let _secret = EnvVarGuard::set( + "TINYCLOUD_KEYS_SECRET", + &base64::encode_config([41; 32], base64::URL_SAFE_NO_PAD), + ); + let target = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let target_addr = target.local_addr().unwrap(); + let location = format!("http://{target_addr}/redirected"); + let location_for_upstream = location.clone(); + let app = Router::new().fallback(any(move || { + let location = location_for_upstream.clone(); + async move { (StatusCode::FOUND, [("location", location)], "redirect body") } + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_addr = listener.local_addr().unwrap(); + let upstream = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let target_server = tokio::spawn(async move { + match tokio::time::timeout(Duration::from_secs(2), target.accept()).await { + Ok(Ok((mut stream, _))) => { + let request = read_upstream_request(&mut stream).await; + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 8\r\nConnection: close\r\n\r\nfollowed", + ) + .await + .unwrap(); + Some(request) + } + Err(_) => None, + Ok(Err(err)) => panic!("target accept failed: {err}"), + } + }); + let results = + proxy_local_requests(upstream_addr, vec!["/redirect"], Duration::from_secs(5)).await; + upstream.abort(); + let redirected_request = target_server.await.unwrap(); + assert!( + redirected_request.is_none(), + "redirect target must receive no request or credentials: {redirected_request:?}", + ); + assert!(matches!( + &results[0][0], + TunnelFrame::Response { id, status: 302, headers } + if id == "req-0" && headers.iter().any(|(name, value)| name == "location" && value == &location) + )); + assert!(matches!( + &results[0][1], + TunnelFrame::ResponseBody { id, chunk, done: true } + if id == "req-0" && base64::decode_config(chunk, base64::STANDARD).unwrap() == b"redirect body" + )); +} + +async fn stalled_upstream_times_out_and_socket_survives(send_headers: bool) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_addr = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (mut stalled, _) = listener.accept().await.unwrap(); + let request = read_upstream_request(&mut stalled).await; + assert_eq!(request.lines().next(), Some("GET /stall HTTP/1.1")); + if send_headers { + stalled + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\na") + .await + .unwrap(); + } + // The request timeout must close the stalled connection, including + // when response headers arrived but the body never finished. + let mut byte = [0]; + assert_eq!( + tokio::time::timeout(Duration::from_secs(35), stalled.read(&mut byte)) + .await + .expect("stalled request was not closed") + .expect("read stalled connection"), + 0, + ); + let (mut healthy, _) = listener.accept().await.unwrap(); + let request = read_upstream_request(&mut healthy).await; + assert_eq!(request.lines().next(), Some("GET /healthy HTTP/1.1")); + healthy + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok") + .await + .unwrap(); + }); + let start = Instant::now(); + let results = proxy_local_requests( + upstream_addr, + vec!["/stall", "/healthy"], + Duration::from_secs(35), + ) + .await; + server.await.unwrap(); + assert!(start.elapsed() < Duration::from_secs(35)); + let expected_prefix = if send_headers { + "failed to read upstream response body:" + } else { + "upstream request failed:" + }; + assert!( + matches!( + &results[0][0], + TunnelFrame::Error { id: Some(id), message } + if id == "req-0" && message.starts_with(expected_prefix) && message.contains("timed out") + ), + "expected the existing per-request timeout error, got {:?}", + results[0] + ); + assert!( + matches!( + &results[1][0], + TunnelFrame::Response { id, status: 200, .. } if id == "req-1" + ), + "the same tunnel must serve the next request" + ); + assert!(matches!( + &results[1][1], + TunnelFrame::ResponseBody { id, chunk, done: true } + if id == "req-1" && base64::decode_config(chunk, base64::STANDARD).unwrap() == b"ok" + )); +} + +#[tokio::test] +async fn tc1132_stalled_upstream_headers_time_out_without_closing_tunnel() { + let _env_lock = env_lock().await; + let _secret = EnvVarGuard::set( + "TINYCLOUD_KEYS_SECRET", + &base64::encode_config([42; 32], base64::URL_SAFE_NO_PAD), + ); + stalled_upstream_times_out_and_socket_survives(false).await; +} + +#[tokio::test] +async fn tc1132_stalled_upstream_body_time_out_without_closing_tunnel() { + let _env_lock = env_lock().await; + let _secret = EnvVarGuard::set( + "TINYCLOUD_KEYS_SECRET", + &base64::encode_config([43; 32], base64::URL_SAFE_NO_PAD), + ); + stalled_upstream_times_out_and_socket_survives(true).await; +}