diff --git a/server/src/main.rs b/server/src/main.rs index ab7a9a4..acce521 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -7,9 +7,10 @@ mod relay; mod ssrf; +use std::collections::hash_map::RandomState; use std::collections::HashMap; use std::convert::Infallible; -use std::hash::{Hash, Hasher}; +use std::hash::{BuildHasher, Hash, Hasher}; use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; @@ -36,6 +37,12 @@ use relay::{ const MAX_INFLIGHT_RELAY: usize = 32; const PROBE_TIMEOUT: Duration = Duration::from_secs(6); const PROBE_TTL: Duration = Duration::from_secs(300); +/// Bound how long a client may take to send its request headers — defeats +/// slow-loris connections that would otherwise hold a Tokio task open forever. +const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(15); +/// Hard ceiling on cached probe entries; past this we evict expired ones (and, +/// if still full, clear) so the cache can't grow without bound over a long run. +const PROBE_CACHE_MAX: usize = 4096; #[derive(Default)] struct Metrics { @@ -57,6 +64,10 @@ struct AppState { metrics: Metrics, relay_gate: Semaphore, probe_cache: Mutex>, + // Random per-process seed for the probe-cache key. DefaultHasher is + // deterministic across runs, so an attacker could engineer URL collisions; + // RandomState reseeds every boot, making the key unpredictable. + cache_hasher: RandomState, } #[tokio::main] @@ -75,28 +86,44 @@ async fn main() { metrics: Metrics::default(), relay_gate: Semaphore::new(MAX_INFLIGHT_RELAY), probe_cache: Mutex::new(HashMap::new()), + cache_hasher: RandomState::new(), }); let listener = TcpListener::bind(bind).await.expect("bind"); eprintln!("The Pondering Orb is gazing on http://{bind} (Ctrl-C to close the eye)"); - loop { - let (stream, _peer) = match listener.accept().await { - Ok(v) => v, - Err(e) => { - log_line("accept_error", &e.to_string(), None, None); - continue; - } - }; - let io = TokioIo::new(stream); - let st = state.clone(); - tokio::spawn(async move { - let svc = service_fn(move |req| handle(st.clone(), req)); - if let Err(e) = http1::Builder::new().serve_connection(io, svc).await { - // client disconnects are noise; log at debug granularity only - let _ = e; - } - }); + let serve = async { + loop { + let (stream, _peer) = match listener.accept().await { + Ok(v) => v, + Err(e) => { + log_line("accept_error", &e.to_string(), None, None); + continue; + } + }; + let io = TokioIo::new(stream); + let st = state.clone(); + tokio::spawn(async move { + let svc = service_fn(move |req| handle(st.clone(), req)); + let conn = http1::Builder::new() + .header_read_timeout(HEADER_READ_TIMEOUT) + .serve_connection(io, svc); + if let Err(e) = conn.await { + // client disconnects are noise; log at debug granularity only + let _ = e; + } + }); + } + }; + + // Honour the banner: a real Ctrl-C handler exits cleanly instead of relying + // on the default SIGINT kill (and keeps tokio's `signal` feature in use). + tokio::select! { + _ = serve => {} + _ = tokio::signal::ctrl_c() => { + eprintln!("\nThe eye closes."); + log_line("shutdown", "ctrl-c", None, None); + } } } @@ -217,8 +244,9 @@ async fn serve_probe(state: &AppState, req: &Request) -> Response return json(StatusCode::BAD_REQUEST, "{\"ok\":false,\"status\":0,\"latency_ms\":0}"), }; - // hash the URL for the cache key (never store the raw URL — token hygiene) - let mut hasher = std::collections::hash_map::DefaultHasher::new(); + // hash the URL for the cache key (never store the raw URL — token hygiene). + // Random per-process seed → keys aren't attacker-predictable across runs. + let mut hasher = state.cache_hasher.build_hasher(); raw.hash(&mut hasher); let key = hasher.finish(); @@ -244,6 +272,14 @@ async fn serve_probe(state: &AppState, req: &Request) -> Response= PROBE_CACHE_MAX { + cache.retain(|_, (ts, ..)| ts.elapsed() < PROBE_TTL); + if cache.len() >= PROBE_CACHE_MAX { + cache.clear(); + } + } cache.insert(key, (Instant::now(), outcome.ok, outcome.status, outcome.latency_ms)); } log_line("probe", if outcome.ok { "live" } else { "dead" }, parsed.host_str(), Some(outcome.latency_ms)); diff --git a/server/src/relay.rs b/server/src/relay.rs index 6a27a69..4bd6854 100644 --- a/server/src/relay.rs +++ b/server/src/relay.rs @@ -25,6 +25,12 @@ const MAX_MANIFEST: usize = 10 * 1024 * 1024; // 10 MB const MAX_SEGMENT: usize = 64 * 1024 * 1024; // 64 MB const MAX_REDIRECTS: u32 = 4; const UA: &str = "PonderingOrb/0.1 (+hardened-relay)"; +/// Bound connect + TLS handshake + request → response-headers. A stalling +/// upstream must never pin one of the relay's permits open indefinitely. +const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15); +/// Bound the body download too — the size cap stops a *large* body, but a +/// slow-drip body would otherwise hold the permit far longer than that implies. +const BODY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); #[derive(Debug)] pub enum RelayError { @@ -36,6 +42,7 @@ pub enum RelayError { Http(String), TooLarge, TooManyRedirects, + Timeout, Upstream(StatusCode), } @@ -50,6 +57,7 @@ impl std::fmt::Display for RelayError { RelayError::Http(e) => write!(f, "http: {e}"), RelayError::TooLarge => write!(f, "response exceeds size cap"), RelayError::TooManyRedirects => write!(f, "too many redirects"), + RelayError::Timeout => write!(f, "upstream timed out"), RelayError::Upstream(s) => write!(f, "upstream status {s}"), } } @@ -105,43 +113,49 @@ async fn fetch_once( .await .map_err(RelayError::Ssrf)?; - let tcp = TcpStream::connect(addr) - .await - .map_err(|e| RelayError::Connect(e.to_string()))?; - let _ = tcp.set_nodelay(true); + // Bound connect → TLS → request → response-headers as one unit. A stalling + // upstream cannot hold one of the 32 relay permits open past CONNECT_TIMEOUT. + let resp = tokio::time::timeout(CONNECT_TIMEOUT, async { + let tcp = TcpStream::connect(addr) + .await + .map_err(|e| RelayError::Connect(e.to_string()))?; + let _ = tcp.set_nodelay(true); - let server_name = rustls::pki_types::ServerName::try_from(host.clone()) - .map_err(|_| RelayError::Tls("invalid sni".into()))?; - let tls = connector - .connect(server_name, tcp) - .await - .map_err(|e| RelayError::Tls(e.to_string()))?; + let server_name = rustls::pki_types::ServerName::try_from(host.clone()) + .map_err(|_| RelayError::Tls("invalid sni".into()))?; + let tls = connector + .connect(server_name, tcp) + .await + .map_err(|e| RelayError::Tls(e.to_string()))?; - let io = TokioIo::new(tls); - let (mut sender, conn) = http1::handshake(io) - .await - .map_err(|e| RelayError::Http(e.to_string()))?; - tokio::spawn(async move { - let _ = conn.await; - }); + let io = TokioIo::new(tls); + let (mut sender, conn) = http1::handshake(io) + .await + .map_err(|e| RelayError::Http(e.to_string()))?; + tokio::spawn(async move { + let _ = conn.await; + }); - let mut pq = url.path().to_string(); - if let Some(q) = url.query() { - pq.push('?'); - pq.push_str(q); - } - let req = Request::builder() - .uri(pq) - .header(hyper::header::HOST, host.as_str()) - .header(hyper::header::USER_AGENT, UA) - .header(hyper::header::ACCEPT, "*/*") - .body(Empty::::new()) - .map_err(|e| RelayError::Http(e.to_string()))?; + let mut pq = url.path().to_string(); + if let Some(q) = url.query() { + pq.push('?'); + pq.push_str(q); + } + let req = Request::builder() + .uri(pq) + .header(hyper::header::HOST, host.as_str()) + .header(hyper::header::USER_AGENT, UA) + .header(hyper::header::ACCEPT, "*/*") + .body(Empty::::new()) + .map_err(|e| RelayError::Http(e.to_string()))?; - let resp = sender - .send_request(req) - .await - .map_err(|e| RelayError::Http(e.to_string()))?; + sender + .send_request(req) + .await + .map_err(|e| RelayError::Http(e.to_string())) + }) + .await + .map_err(|_| RelayError::Timeout)??; let status = resp.status(); let content_type = resp @@ -171,18 +185,48 @@ async fn fetch_once( } let limited = Limited::new(resp.into_body(), max_bytes); - let collected = limited.collect().await.map_err(|_| RelayError::TooLarge)?; + let collected = tokio::time::timeout(BODY_TIMEOUT, limited.collect()) + .await + .map_err(|_| RelayError::Timeout)? + .map_err(|_| RelayError::TooLarge)?; let body = collected.to_bytes(); Ok(Upstream { status, - content_type, + content_type: sanitize_content_type(&content_type), body, final_url: url.clone(), redirects: 0, }) } +/// Clamp the upstream-declared Content-Type to a known-safe media allowlist. +/// An attacker-controlled stream origin could otherwise label segment bytes as +/// `text/html` or `application/javascript`; if a tab ever navigates straight to +/// a /relay URL, that declared type (not just sniffing, which nosniff blocks) +/// would govern interpretation. Anything off-list collapses to octet-stream. +fn sanitize_content_type(ct: &str) -> String { + // Compare on the bare type, ignoring any `; charset=`/`; codecs=` parameters. + let base = ct.split(';').next().unwrap_or("").trim().to_ascii_lowercase(); + const ALLOWED: &[&str] = &[ + "application/vnd.apple.mpegurl", + "application/x-mpegurl", + "audio/mpegurl", + "audio/x-mpegurl", + "video/mp2t", + "video/mp4", + "video/iso.segment", + "audio/aac", + "audio/mp4", + "application/octet-stream", + ]; + if ALLOWED.contains(&base.as_str()) { + ct.to_string() + } else { + "application/octet-stream".to_string() + } +} + /// Public entry: validated GET with bounded, re-validated redirect following. pub async fn fetch_validated( connector: &TlsConnector, @@ -350,7 +394,7 @@ fn rewrite_tag_uris(line: &str, base: &Url) -> String { Ok(abs) => { let mut s = String::with_capacity(line.len() + 40); s.push_str(&line[..uri_start]); - s.push_str(&wrap_raw(&abs)); + s.push_str(&wrap(&abs)); s.push_str(&line[end..]); s } @@ -358,14 +402,12 @@ fn rewrite_tag_uris(line: &str, base: &Url) -> String { } } -/// "/relay?url=" +/// "/relay?url=". The percent-encoding leaves only +/// RFC3986-unreserved bytes, so the result is safe both on a bare URI line and +/// inside an already-quoted `URI="..."` tag attribute (no `"` can appear). fn wrap(abs: &Url) -> String { format!("/relay?url={}", pct(abs.as_str())) } -fn wrap_raw(abs: &Url) -> String { - // for inside an already-quoted attribute - wrap(abs) -} /// Minimal RFC3986 component percent-encoding (encode everything non-unreserved). fn pct(s: &str) -> String { diff --git a/server/src/ssrf.rs b/server/src/ssrf.rs index 1afc631..c8919c8 100644 --- a/server/src/ssrf.rs +++ b/server/src/ssrf.rs @@ -7,10 +7,23 @@ //! between check and connect cannot smuggle us onto an internal address. use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; +use std::sync::OnceLock; use hickory_resolver::config::{ResolverConfig, ResolverOpts}; use hickory_resolver::TokioAsyncResolver; +/// Built once on first use and reused for the process lifetime. Rebuilding the +/// resolver per request re-read /etc/resolv.conf and — worse — discarded +/// hickory's in-memory DNS cache, forcing a fresh lookup for every HLS segment. +static RESOLVER: OnceLock = OnceLock::new(); + +fn resolver() -> &'static TokioAsyncResolver { + RESOLVER.get_or_init(|| { + TokioAsyncResolver::tokio_from_system_conf() + .unwrap_or_else(|_| TokioAsyncResolver::tokio(ResolverConfig::default(), ResolverOpts::default())) + }) +} + #[derive(Debug)] pub enum SsrfError { Resolve(String), @@ -78,11 +91,7 @@ pub async fn resolve_and_validate(host: &str, port: u16) -> Result r, - Err(_) => TokioAsyncResolver::tokio(ResolverConfig::default(), ResolverOpts::default()), - }; - let lookup = resolver + let lookup = resolver() .lookup_ip(host) .await .map_err(|e| SsrfError::Resolve(e.to_string()))?; diff --git a/web/guide.js b/web/guide.js index 7eab78f..663a167 100644 --- a/web/guide.js +++ b/web/guide.js @@ -112,7 +112,7 @@ export class Guide { _renderRow(ch, idx) { const row = document.createElement('div'); - row.className = 'guide-row'; + row.className = 'guide-row' + (idx === this.focusRow ? ' focused' : ''); row.style.position = 'absolute'; row.style.top = (idx * ROW_H) + 'px'; row.style.left = '0'; row.style.right = '0'; @@ -189,12 +189,18 @@ export class Guide { _onKey(e) { const list = this.agg.channels; switch (e.key) { - case 'ArrowDown': e.preventDefault(); this.focusRow = Math.min(list.length-1, this.focusRow+1); this._scrollToFocus(); break; - case 'ArrowUp': e.preventDefault(); this.focusRow = Math.max(0, this.focusRow-1); this._scrollToFocus(); break; + case 'ArrowDown': e.preventDefault(); this.focusRow = Math.min(list.length-1, this.focusRow+1); this._moveFocus(); break; + case 'ArrowUp': e.preventDefault(); this.focusRow = Math.max(0, this.focusRow-1); this._moveFocus(); break; case 'Enter': e.preventDefault(); { const ch = list[this.focusRow]; if (ch){ this.onTune(ch); this.hide(); } } break; case 'Escape': e.preventDefault(); this.hide(); break; } } + // Scroll the focused row into view, then repaint so the .focused highlight + // tracks the cursor even when no scroll was needed (scroll event wouldn't fire). + _moveFocus() { + this._scrollToFocus(); + this._renderViewport(); + } _scrollToFocus() { const y = this.focusRow * ROW_H; if (y < this.scroll.scrollTop) this.scroll.scrollTop = y; diff --git a/web/player.js b/web/player.js index 3b6784c..3047367 100644 --- a/web/player.js +++ b/web/player.js @@ -27,6 +27,7 @@ export class Player { this._fallbackIdx = 0; this._wireControls(); this._wireActivity(); + this._wireBuffering(); } /** @@ -93,12 +94,31 @@ export class Player { if (this.hls) { try { this.hls.destroy(); } catch {} this.hls = null; } this.video.removeAttribute('src'); try { this.video.load(); } catch {} + this._setBuffering(false); // never leave the indicator stuck across a zap } destroy() { this._teardown(); } + // ---------- buffering / stall indicator ---------- + // Bridge the gap between the scrying static clearing and the first frame: + // without a hint, a slow-loading stream is indistinguishable from a dead one. + _wireBuffering() { + const v = this.video; + const on = () => this._setBuffering(true); + const off = () => this._setBuffering(false); + v.addEventListener('waiting', on); + v.addEventListener('stalled', on); + v.addEventListener('playing', off); + v.addEventListener('canplay', off); + v.addEventListener('pause', off); + v.addEventListener('emptied', off); + } + _setBuffering(on) { + if (this.els.screen) this.els.screen.classList.toggle('buffering', on); + } + // ---------- dead-stream "the orb clouds over" ---------- - _showNoSignal() { this.els.nosignal.hidden = false; } + _showNoSignal() { this.els.nosignal.hidden = false; this._setBuffering(false); } _hideNoSignal() { this.els.nosignal.hidden = true; } // ---------- live-aware chrome ---------- diff --git a/web/styles.css b/web/styles.css index 7343bd7..7e4b68e 100644 --- a/web/styles.css +++ b/web/styles.css @@ -91,6 +91,19 @@ img{display:block} transition:opacity .12s linear} .static.show{opacity:.9} +/* ===================== buffering / tuning ===================== */ +/* Corner hint shown while