feat: initial release of The Pondering Orb
A retro CRT-themed web TV that channel-surfs live HLS streams through a security-hardened, same-origin Rust relay. Front end (vanilla JS + HLS.js, no framework/build step): - SourceProvider abstraction + canonical Channel/Program model so public streams and future self-hosted media share one UI (MockProvider built first to keep the UI provably source-agnostic; IptvOrgProvider for live data) - Red/black phosphor CRT theme, virtualized 1994-style guide (The Almanac), cable-box zapping with 7-segment OSD, scrying-static transitions - Theater/Dim/Ambient immersion modes, dead-stream "orb clouds over" handling - The Lexicon in-world naming with mundane-mode; full a11y + reduced-motion - HLS.js pinned + Subresource-Integrity verified Rust relay (hyper 1.x + tokio-rustls/ring + hickory-resolver): - Serves the SPA and proxies HLS from one origin (no CORS, untainted canvas) - SSRF defense: resolve -> reject private/loopback/link-local/metadata IPs -> connect to the validated IP (no re-resolve, defeats DNS-rebind), fail-closed - HTTPS-only egress, re-validated redirects, manifest URL rewrite, size/ concurrency caps, strict CSP + security headers, path-traversal guard - /metrics observability + structured logs (host only, never raw URLs) Verified: all SSRF probes rejected, CSP clean, real Apple HLS manifest rewritten, ~39k iptv-org channels fetched end-to-end, tests green, zero warnings. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Generated
+1184
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,35 @@
|
||||
[package]
|
||||
name = "pondering-orb"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
description = "The Pondering Orb — hardened same-origin static server + HLS relay for a retro CRT web TV."
|
||||
license = "MIT"
|
||||
|
||||
[[bin]]
|
||||
name = "pondering-orb"
|
||||
path = "src/main.rs"
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "io-util", "time", "signal", "sync", "fs"] }
|
||||
hyper = { version = "1", features = ["server", "client", "http1"] }
|
||||
hyper-util = { version = "0.1", features = ["server", "client", "client-legacy", "http1", "tokio"] }
|
||||
http-body-util = "0.1"
|
||||
bytes = "1"
|
||||
# TLS to upstream stream origins (https-only egress).
|
||||
# Use the `ring` provider (pure-ish, no cmake/C toolchain) rather than the
|
||||
# aws-lc-rs default, so the build stays self-contained.
|
||||
rustls = { version = "0.23", default-features = false, features = ["ring", "std", "tls12", "logging"] }
|
||||
tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12", "logging"] }
|
||||
webpki-roots = "0.26"
|
||||
# DNS resolution so we can validate the *resolved* IP before connecting (SSRF defense)
|
||||
hickory-resolver = "0.24"
|
||||
# robust URL parse + relative-join for manifest rewriting
|
||||
url = "2"
|
||||
|
||||
[profile.release]
|
||||
# reproducible-ish, lean binary (CI-gate task hardens this further)
|
||||
opt-level = "z"
|
||||
lto = true
|
||||
codegen-units = 1
|
||||
strip = true
|
||||
panic = "abort"
|
||||
@@ -0,0 +1,330 @@
|
||||
//! The Pondering Orb — same-origin static server + hardened HLS relay.
|
||||
//!
|
||||
//! Binds 127.0.0.1 by default. Serves web/ AND proxies streams from the same
|
||||
//! origin so the SPA needs no CORS and the ambient canvas never taints. The
|
||||
//! relay is the sole egress point; all SSRF defense lives in ssrf.rs.
|
||||
|
||||
mod relay;
|
||||
mod ssrf;
|
||||
|
||||
use std::convert::Infallible;
|
||||
use std::net::SocketAddr;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
|
||||
use bytes::Bytes;
|
||||
use http_body_util::Full;
|
||||
use hyper::body::Incoming;
|
||||
use hyper::server::conn::http1;
|
||||
use hyper::service::service_fn;
|
||||
use hyper::{Request, Response, StatusCode};
|
||||
use hyper_util::rt::TokioIo;
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::Semaphore;
|
||||
use tokio_rustls::TlsConnector;
|
||||
use url::Url;
|
||||
|
||||
use relay::{
|
||||
build_tls_connector, fetch_validated, looks_like_manifest, max_for, rewrite_manifest,
|
||||
RelayError,
|
||||
};
|
||||
|
||||
const MAX_INFLIGHT_RELAY: usize = 32;
|
||||
|
||||
#[derive(Default)]
|
||||
struct Metrics {
|
||||
relay_ok: AtomicU64,
|
||||
relay_ssrf_block: AtomicU64,
|
||||
relay_upstream_fail: AtomicU64,
|
||||
relay_ratelimit: AtomicU64,
|
||||
relay_bad_request: AtomicU64,
|
||||
}
|
||||
|
||||
struct AppState {
|
||||
web_root: PathBuf,
|
||||
tls: TlsConnector,
|
||||
metrics: Metrics,
|
||||
relay_gate: Semaphore,
|
||||
}
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
let bind: SocketAddr = std::env::var("ORB_BIND")
|
||||
.ok()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or_else(|| "127.0.0.1:8088".parse().unwrap());
|
||||
|
||||
let web_root = locate_web_root();
|
||||
log_line("startup", &format!("web_root={}", web_root.display()), None, None);
|
||||
|
||||
let state = Arc::new(AppState {
|
||||
web_root,
|
||||
tls: build_tls_connector(),
|
||||
metrics: Metrics::default(),
|
||||
relay_gate: Semaphore::new(MAX_INFLIGHT_RELAY),
|
||||
});
|
||||
|
||||
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;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
fn locate_web_root() -> PathBuf {
|
||||
if let Ok(p) = std::env::var("ORB_WEB_ROOT") {
|
||||
return PathBuf::from(p);
|
||||
}
|
||||
for cand in ["web", "../web", "./web"] {
|
||||
let p = PathBuf::from(cand);
|
||||
if p.join("index.html").exists() {
|
||||
return p.canonicalize().unwrap_or(p);
|
||||
}
|
||||
}
|
||||
PathBuf::from("web")
|
||||
}
|
||||
|
||||
async fn handle(
|
||||
state: Arc<AppState>,
|
||||
req: Request<Incoming>,
|
||||
) -> Result<Response<Full<Bytes>>, Infallible> {
|
||||
let path = req.uri().path().to_string();
|
||||
let resp = match path.as_str() {
|
||||
"/relay" => serve_relay(&state, &req).await,
|
||||
"/metrics" => serve_metrics(&state),
|
||||
"/healthz" => text(StatusCode::OK, "ok"),
|
||||
_ => serve_static(&state, &path).await,
|
||||
};
|
||||
Ok(with_security_headers(resp, &path))
|
||||
}
|
||||
|
||||
// ----------------------------- /relay ------------------------------------
|
||||
|
||||
async fn serve_relay(state: &AppState, req: &Request<Incoming>) -> Response<Full<Bytes>> {
|
||||
let query = req.uri().query().unwrap_or("");
|
||||
let raw = url::form_urlencoded::parse(query.as_bytes())
|
||||
.find(|(k, _)| k == "url")
|
||||
.map(|(_, v)| v.into_owned());
|
||||
|
||||
let Some(raw) = raw else {
|
||||
state.metrics.relay_bad_request.fetch_add(1, Ordering::Relaxed);
|
||||
return text(StatusCode::BAD_REQUEST, "missing url param");
|
||||
};
|
||||
|
||||
let parsed = match Url::parse(&raw) {
|
||||
Ok(u) => u,
|
||||
Err(_) => {
|
||||
state.metrics.relay_bad_request.fetch_add(1, Ordering::Relaxed);
|
||||
return text(StatusCode::BAD_REQUEST, "malformed url");
|
||||
}
|
||||
};
|
||||
if parsed.scheme() != "https" {
|
||||
state.metrics.relay_bad_request.fetch_add(1, Ordering::Relaxed);
|
||||
log_line("relay_block", "non-https", parsed.host_str(), None);
|
||||
return text(StatusCode::BAD_REQUEST, "only https upstreams allowed");
|
||||
}
|
||||
|
||||
// concurrency cap
|
||||
let _permit = match state.relay_gate.try_acquire() {
|
||||
Ok(p) => p,
|
||||
Err(_) => {
|
||||
state.metrics.relay_ratelimit.fetch_add(1, Ordering::Relaxed);
|
||||
return text(StatusCode::SERVICE_UNAVAILABLE, "relay busy");
|
||||
}
|
||||
};
|
||||
|
||||
let is_manifest_hint = parsed.path().to_ascii_lowercase().ends_with(".m3u8");
|
||||
let cap = max_for(is_manifest_hint);
|
||||
let started = Instant::now();
|
||||
|
||||
match fetch_validated(&state.tls, &parsed, cap).await {
|
||||
Ok(up) => {
|
||||
let ms = started.elapsed().as_millis() as u64;
|
||||
state.metrics.relay_ok.fetch_add(1, Ordering::Relaxed);
|
||||
if looks_like_manifest(&up.final_url, &up.content_type) {
|
||||
let text_body = String::from_utf8_lossy(&up.body);
|
||||
let rewritten = rewrite_manifest(&text_body, &up.final_url);
|
||||
log_line("relay_ok", &format!("manifest redirects={}", up.redirects), parsed.host_str(), Some(ms));
|
||||
return bytes_resp(
|
||||
StatusCode::OK,
|
||||
"application/vnd.apple.mpegurl",
|
||||
Bytes::from(rewritten),
|
||||
);
|
||||
}
|
||||
log_line("relay_ok", &format!("segment redirects={}", up.redirects), parsed.host_str(), Some(ms));
|
||||
bytes_resp(StatusCode::OK, &up.content_type, up.body)
|
||||
}
|
||||
Err(e) => {
|
||||
let ms = started.elapsed().as_millis() as u64;
|
||||
match &e {
|
||||
RelayError::Ssrf(_) => {
|
||||
state.metrics.relay_ssrf_block.fetch_add(1, Ordering::Relaxed);
|
||||
log_line("relay_ssrf_block", &e.to_string(), parsed.host_str(), Some(ms));
|
||||
text(StatusCode::FORBIDDEN, "blocked by relay policy")
|
||||
}
|
||||
_ => {
|
||||
state.metrics.relay_upstream_fail.fetch_add(1, Ordering::Relaxed);
|
||||
log_line("relay_upstream_fail", &e.to_string(), parsed.host_str(), Some(ms));
|
||||
text(StatusCode::BAD_GATEWAY, "upstream unavailable")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ----------------------------- /metrics ----------------------------------
|
||||
|
||||
fn serve_metrics(state: &AppState) -> Response<Full<Bytes>> {
|
||||
let m = &state.metrics;
|
||||
let body = format!(
|
||||
"# HELP orb_relay_total Relay requests by outcome.\n\
|
||||
# TYPE orb_relay_total counter\n\
|
||||
orb_relay_total{{event=\"ok\"}} {}\n\
|
||||
orb_relay_total{{event=\"ssrf_block\"}} {}\n\
|
||||
orb_relay_total{{event=\"upstream_fail\"}} {}\n\
|
||||
orb_relay_total{{event=\"ratelimit\"}} {}\n\
|
||||
orb_relay_total{{event=\"bad_request\"}} {}\n",
|
||||
m.relay_ok.load(Ordering::Relaxed),
|
||||
m.relay_ssrf_block.load(Ordering::Relaxed),
|
||||
m.relay_upstream_fail.load(Ordering::Relaxed),
|
||||
m.relay_ratelimit.load(Ordering::Relaxed),
|
||||
m.relay_bad_request.load(Ordering::Relaxed),
|
||||
);
|
||||
bytes_resp(StatusCode::OK, "text/plain; version=0.0.4", Bytes::from(body))
|
||||
}
|
||||
|
||||
// ----------------------------- static ------------------------------------
|
||||
|
||||
async fn serve_static(state: &AppState, req_path: &str) -> Response<Full<Bytes>> {
|
||||
let rel = if req_path == "/" { "index.html" } else { req_path.trim_start_matches('/') };
|
||||
|
||||
// Path-traversal guard: join, canonicalize, ensure still under web_root.
|
||||
let joined = state.web_root.join(rel);
|
||||
let canon = match joined.canonicalize() {
|
||||
Ok(c) => c,
|
||||
Err(_) => return text(StatusCode::NOT_FOUND, "not found"),
|
||||
};
|
||||
if !canon.starts_with(&state.web_root) {
|
||||
return text(StatusCode::FORBIDDEN, "forbidden");
|
||||
}
|
||||
|
||||
match tokio::fs::read(&canon).await {
|
||||
Ok(bytes) => {
|
||||
let ct = content_type_for(&canon);
|
||||
// Inject the relay marker so the SPA auto-selects the live provider.
|
||||
if canon.file_name().and_then(|n| n.to_str()) == Some("index.html") {
|
||||
let html = String::from_utf8_lossy(&bytes).replacen(
|
||||
"<head>",
|
||||
"<head>\n<meta name=\"orb-relay\" content=\"1\">",
|
||||
1,
|
||||
);
|
||||
return bytes_resp(StatusCode::OK, ct, Bytes::from(html));
|
||||
}
|
||||
bytes_resp(StatusCode::OK, ct, Bytes::from(bytes))
|
||||
}
|
||||
Err(_) => text(StatusCode::NOT_FOUND, "not found"),
|
||||
}
|
||||
}
|
||||
|
||||
fn content_type_for(p: &Path) -> &'static str {
|
||||
match p.extension().and_then(|e| e.to_str()) {
|
||||
Some("html") => "text/html; charset=utf-8",
|
||||
Some("js") | Some("mjs") => "text/javascript; charset=utf-8",
|
||||
Some("css") => "text/css; charset=utf-8",
|
||||
Some("json") => "application/json",
|
||||
Some("svg") => "image/svg+xml",
|
||||
Some("png") => "image/png",
|
||||
Some("jpg") | Some("jpeg") => "image/jpeg",
|
||||
Some("webp") => "image/webp",
|
||||
Some("ico") => "image/x-icon",
|
||||
Some("woff2") => "font/woff2",
|
||||
Some("map") => "application/json",
|
||||
_ => "application/octet-stream",
|
||||
}
|
||||
}
|
||||
|
||||
// ----------------------------- helpers -----------------------------------
|
||||
|
||||
fn with_security_headers(mut resp: Response<Full<Bytes>>, path: &str) -> Response<Full<Bytes>> {
|
||||
let h = resp.headers_mut();
|
||||
// Strict, same-origin CSP. img/media relaxed only where the design requires:
|
||||
// - img-src https: → remote channel logos (images can't execute)
|
||||
// - media-src blob: → HLS.js feeds segments to <video> via MSE blobs
|
||||
// All scripting/styling/XHR stay 'self'; the relay is the only egress.
|
||||
h.insert(
|
||||
"Content-Security-Policy",
|
||||
hyper::header::HeaderValue::from_static(
|
||||
"default-src 'none'; script-src 'self'; style-src 'self'; img-src 'self' https: data:; \
|
||||
media-src 'self' blob:; connect-src 'self'; font-src 'self'; object-src 'none'; \
|
||||
base-uri 'none'; frame-src 'none'; form-action 'none'; frame-ancestors 'none'",
|
||||
),
|
||||
);
|
||||
h.insert("X-Content-Type-Options", hyper::header::HeaderValue::from_static("nosniff"));
|
||||
h.insert("Referrer-Policy", hyper::header::HeaderValue::from_static("no-referrer"));
|
||||
h.insert("X-Frame-Options", hyper::header::HeaderValue::from_static("DENY"));
|
||||
h.insert(
|
||||
"Permissions-Policy",
|
||||
hyper::header::HeaderValue::from_static("camera=(), microphone=(), geolocation=(), interest-cohort=()"),
|
||||
);
|
||||
// HSTS is only meaningful over TLS; harmless on loopback but include for parity.
|
||||
h.insert(
|
||||
"Strict-Transport-Security",
|
||||
hyper::header::HeaderValue::from_static("max-age=31536000"),
|
||||
);
|
||||
let _ = path;
|
||||
resp
|
||||
}
|
||||
|
||||
fn bytes_resp(status: StatusCode, content_type: &str, body: Bytes) -> Response<Full<Bytes>> {
|
||||
Response::builder()
|
||||
.status(status)
|
||||
.header(hyper::header::CONTENT_TYPE, content_type)
|
||||
.body(Full::new(body))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn text(status: StatusCode, msg: &str) -> Response<Full<Bytes>> {
|
||||
bytes_resp(status, "text/plain; charset=utf-8", Bytes::from(msg.to_string()))
|
||||
}
|
||||
|
||||
/// Structured one-line JSON log. We log host (not full URL) + outcome — query
|
||||
/// strings can carry stream tokens, so the raw URL is never recorded.
|
||||
fn log_line(event: &str, detail: &str, host: Option<&str>, latency_ms: Option<u64>) {
|
||||
let ts = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
let host = host.unwrap_or("-");
|
||||
match latency_ms {
|
||||
Some(ms) => println!(
|
||||
"{{\"ts\":{ts},\"event\":\"{event}\",\"host\":\"{host}\",\"latency_ms\":{ms},\"detail\":\"{}\"}}",
|
||||
esc(detail)
|
||||
),
|
||||
None => println!(
|
||||
"{{\"ts\":{ts},\"event\":\"{event}\",\"host\":\"{host}\",\"detail\":\"{}\"}}",
|
||||
esc(detail)
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
fn esc(s: &str) -> String {
|
||||
s.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', " ")
|
||||
}
|
||||
@@ -0,0 +1,303 @@
|
||||
//! Hardened HLS relay.
|
||||
//!
|
||||
//! - https-only upstreams
|
||||
//! - every hop SSRF-validated (resolve → reject internal IPs → connect to the
|
||||
//! exact validated IP, no re-resolve)
|
||||
//! - bounded redirect following, each redirect re-validated
|
||||
//! - response size caps
|
||||
//! - HLS manifests rewritten so every nested URL (variants, segments, keys)
|
||||
//! flows back through this same relay
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use bytes::Bytes;
|
||||
use http_body_util::{BodyExt, Empty, Limited};
|
||||
use hyper::client::conn::http1;
|
||||
use hyper::{Request, StatusCode};
|
||||
use hyper_util::rt::TokioIo;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio_rustls::TlsConnector;
|
||||
use url::Url;
|
||||
|
||||
use crate::ssrf::{resolve_and_validate, SsrfError};
|
||||
|
||||
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)";
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum RelayError {
|
||||
Ssrf(SsrfError),
|
||||
NotHttps,
|
||||
BadUrl,
|
||||
Connect(String),
|
||||
Tls(String),
|
||||
Http(String),
|
||||
TooLarge,
|
||||
TooManyRedirects,
|
||||
Upstream(StatusCode),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for RelayError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
RelayError::Ssrf(e) => write!(f, "ssrf: {e}"),
|
||||
RelayError::NotHttps => write!(f, "only https upstreams allowed"),
|
||||
RelayError::BadUrl => write!(f, "malformed url"),
|
||||
RelayError::Connect(e) => write!(f, "connect: {e}"),
|
||||
RelayError::Tls(e) => write!(f, "tls: {e}"),
|
||||
RelayError::Http(e) => write!(f, "http: {e}"),
|
||||
RelayError::TooLarge => write!(f, "response exceeds size cap"),
|
||||
RelayError::TooManyRedirects => write!(f, "too many redirects"),
|
||||
RelayError::Upstream(s) => write!(f, "upstream status {s}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct Upstream {
|
||||
pub status: StatusCode,
|
||||
pub content_type: String,
|
||||
pub body: Bytes,
|
||||
pub final_url: Url,
|
||||
pub redirects: u32,
|
||||
}
|
||||
|
||||
/// Build the shared rustls/TLS connector (webpki roots, explicit ring provider).
|
||||
pub fn build_tls_connector() -> TlsConnector {
|
||||
let mut roots = rustls::RootCertStore::empty();
|
||||
roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
|
||||
let config = rustls::ClientConfig::builder_with_provider(Arc::new(
|
||||
rustls::crypto::ring::default_provider(),
|
||||
))
|
||||
.with_safe_default_protocol_versions()
|
||||
.expect("tls protocol versions")
|
||||
.with_root_certificates(roots)
|
||||
.with_no_client_auth();
|
||||
TlsConnector::from(Arc::new(config))
|
||||
}
|
||||
|
||||
fn is_https(u: &Url) -> bool {
|
||||
u.scheme() == "https"
|
||||
}
|
||||
|
||||
/// One validated HTTPS GET (no redirect following — caller loops).
|
||||
async fn fetch_once(
|
||||
connector: &TlsConnector,
|
||||
url: &Url,
|
||||
max_bytes: usize,
|
||||
) -> Result<Upstream, RelayError> {
|
||||
if !is_https(url) {
|
||||
return Err(RelayError::NotHttps);
|
||||
}
|
||||
let host = url.host_str().ok_or(RelayError::BadUrl)?.to_string();
|
||||
let port = url.port_or_known_default().unwrap_or(443);
|
||||
|
||||
// SSRF: resolve ourselves, validate, connect to that exact IP.
|
||||
let addr = resolve_and_validate(&host, port)
|
||||
.await
|
||||
.map_err(RelayError::Ssrf)?;
|
||||
|
||||
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 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::<Bytes>::new())
|
||||
.map_err(|e| RelayError::Http(e.to_string()))?;
|
||||
|
||||
let resp = sender
|
||||
.send_request(req)
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(e.to_string()))?;
|
||||
|
||||
let status = resp.status();
|
||||
let content_type = resp
|
||||
.headers()
|
||||
.get(hyper::header::CONTENT_TYPE)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("application/octet-stream")
|
||||
.to_string();
|
||||
let location = resp
|
||||
.headers()
|
||||
.get(hyper::header::LOCATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|s| s.to_string());
|
||||
|
||||
// Redirect: return an empty body, let the caller re-validate + re-fetch.
|
||||
if status.is_redirection() {
|
||||
if let Some(loc) = location {
|
||||
let next = url.join(&loc).map_err(|_| RelayError::BadUrl)?;
|
||||
return Ok(Upstream {
|
||||
status,
|
||||
content_type,
|
||||
body: Bytes::new(),
|
||||
final_url: next,
|
||||
redirects: 0,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let limited = Limited::new(resp.into_body(), max_bytes);
|
||||
let collected = limited.collect().await.map_err(|_| RelayError::TooLarge)?;
|
||||
let body = collected.to_bytes();
|
||||
|
||||
Ok(Upstream {
|
||||
status,
|
||||
content_type,
|
||||
body,
|
||||
final_url: url.clone(),
|
||||
redirects: 0,
|
||||
})
|
||||
}
|
||||
|
||||
/// Public entry: validated GET with bounded, re-validated redirect following.
|
||||
pub async fn fetch_validated(
|
||||
connector: &TlsConnector,
|
||||
start: &Url,
|
||||
max_bytes: usize,
|
||||
) -> Result<Upstream, RelayError> {
|
||||
let mut current = start.clone();
|
||||
let mut redirects = 0u32;
|
||||
loop {
|
||||
let up = fetch_once(connector, ¤t, max_bytes).await?;
|
||||
if up.status.is_redirection() && !up.final_url.as_str().is_empty() {
|
||||
redirects += 1;
|
||||
if redirects > MAX_REDIRECTS {
|
||||
return Err(RelayError::TooManyRedirects);
|
||||
}
|
||||
current = up.final_url;
|
||||
continue;
|
||||
}
|
||||
if up.status.is_server_error() || up.status.is_client_error() {
|
||||
return Err(RelayError::Upstream(up.status));
|
||||
}
|
||||
return Ok(Upstream { redirects, ..up });
|
||||
}
|
||||
}
|
||||
|
||||
/// Heuristic: is this an HLS manifest we should rewrite?
|
||||
pub fn looks_like_manifest(url: &Url, content_type: &str) -> bool {
|
||||
let ct = content_type.to_ascii_lowercase();
|
||||
ct.contains("mpegurl")
|
||||
|| ct.contains("application/x-mpegurl")
|
||||
|| ct.contains("vnd.apple.mpegurl")
|
||||
|| url.path().to_ascii_lowercase().ends_with(".m3u8")
|
||||
}
|
||||
|
||||
/// Rewrite every nested URI in an HLS manifest so it routes back through /relay.
|
||||
/// Relative URIs are resolved against `base` first. Never executes content.
|
||||
pub fn rewrite_manifest(text: &str, base: &Url) -> String {
|
||||
let mut out = String::with_capacity(text.len() + text.len() / 4);
|
||||
for raw in text.split_inclusive('\n') {
|
||||
let line = raw.trim_end_matches(['\n', '\r']);
|
||||
let nl = &raw[line.len()..];
|
||||
|
||||
if line.is_empty() {
|
||||
out.push_str(raw);
|
||||
continue;
|
||||
}
|
||||
if line.starts_with('#') {
|
||||
out.push_str(&rewrite_tag_uris(line, base));
|
||||
out.push_str(nl);
|
||||
continue;
|
||||
}
|
||||
// bare URI line (variant playlist or media segment)
|
||||
match base.join(line) {
|
||||
Ok(abs) => {
|
||||
out.push_str(&wrap(&abs));
|
||||
}
|
||||
Err(_) => out.push_str(line), // leave untouched if unparseable
|
||||
}
|
||||
out.push_str(nl);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Rewrite URI="..." attributes inside EXT-X-* tags (KEY, MEDIA, MAP, I-FRAME…).
|
||||
fn rewrite_tag_uris(line: &str, base: &Url) -> String {
|
||||
const NEEDLE: &str = "URI=\"";
|
||||
let Some(start) = line.find(NEEDLE) else {
|
||||
return line.to_string();
|
||||
};
|
||||
let uri_start = start + NEEDLE.len();
|
||||
let Some(rel_end) = line[uri_start..].find('"') else {
|
||||
return line.to_string();
|
||||
};
|
||||
let end = uri_start + rel_end;
|
||||
let uri = &line[uri_start..end];
|
||||
match base.join(uri) {
|
||||
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(&line[end..]);
|
||||
s
|
||||
}
|
||||
Err(_) => line.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
/// "/relay?url=<pct-encoded absolute>"
|
||||
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 {
|
||||
const UNRESERVED: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_.~";
|
||||
let mut out = String::with_capacity(s.len() * 3);
|
||||
for &b in s.as_bytes() {
|
||||
if UNRESERVED.contains(&b) {
|
||||
out.push(b as char);
|
||||
} else {
|
||||
out.push('%');
|
||||
out.push(hex((b >> 4) & 0xf));
|
||||
out.push(hex(b & 0xf));
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
fn hex(n: u8) -> char {
|
||||
match n {
|
||||
0..=9 => (b'0' + n) as char,
|
||||
_ => (b'A' + (n - 10)) as char,
|
||||
}
|
||||
}
|
||||
|
||||
pub const fn max_for(is_manifest: bool) -> usize {
|
||||
if is_manifest {
|
||||
MAX_MANIFEST
|
||||
} else {
|
||||
MAX_SEGMENT
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
//! SSRF defense — the whole reason the relay exists.
|
||||
//!
|
||||
//! Every upstream URL is attacker-influenced (it comes from public iptv-org
|
||||
//! playlists). We resolve the host ourselves, reject any resolved IP that is
|
||||
//! private / loopback / link-local / cloud-metadata / IPv6-local, and then
|
||||
//! connect to *that exact validated IP* — never re-resolving — so a DNS-rebind
|
||||
//! between check and connect cannot smuggle us onto an internal address.
|
||||
|
||||
use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
|
||||
|
||||
use hickory_resolver::config::{ResolverConfig, ResolverOpts};
|
||||
use hickory_resolver::TokioAsyncResolver;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum SsrfError {
|
||||
Resolve(String),
|
||||
NoSafeAddress,
|
||||
Blocked(IpAddr),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for SsrfError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
SsrfError::Resolve(e) => write!(f, "dns resolve failed: {e}"),
|
||||
SsrfError::NoSafeAddress => write!(f, "no non-forbidden address for host"),
|
||||
SsrfError::Blocked(ip) => write!(f, "address blocked by SSRF policy: {ip}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
impl std::error::Error for SsrfError {}
|
||||
|
||||
/// True if this IP must never be reached by the relay.
|
||||
pub fn is_forbidden_ip(ip: IpAddr) -> bool {
|
||||
match ip {
|
||||
IpAddr::V4(v4) => is_forbidden_v4(v4),
|
||||
IpAddr::V6(v6) => is_forbidden_v6(v6),
|
||||
}
|
||||
}
|
||||
|
||||
fn is_forbidden_v4(ip: Ipv4Addr) -> bool {
|
||||
ip.is_private() // 10/8, 172.16/12, 192.168/16
|
||||
|| ip.is_loopback() // 127/8
|
||||
|| ip.is_link_local() // 169.254/16 (incl. 169.254.169.254 metadata)
|
||||
|| ip.is_broadcast() // 255.255.255.255
|
||||
|| ip.is_documentation() // 192.0.2/24, 198.51.100/24, 203.0.113/24
|
||||
|| ip.is_unspecified() // 0.0.0.0
|
||||
|| ip.octets()[0] == 0 // 0.0.0.0/8
|
||||
|| (ip.octets()[0] == 100 && (ip.octets()[1] & 0xC0) == 64) // 100.64/10 CGNAT/Tailscale
|
||||
|| (ip.octets()[0] & 0xF0) == 0xF0 // 240/4 reserved (+ multicast handled below)
|
||||
|| ip.is_multicast() // 224/4
|
||||
}
|
||||
|
||||
fn is_forbidden_v6(ip: Ipv6Addr) -> bool {
|
||||
// Reject IPv4-mapped/compat addresses by re-checking the embedded v4.
|
||||
if let Some(v4) = ip.to_ipv4_mapped() {
|
||||
return is_forbidden_v4(v4);
|
||||
}
|
||||
if let Some(v4) = ip.to_ipv4() {
|
||||
return is_forbidden_v4(v4);
|
||||
}
|
||||
ip.is_loopback() // ::1
|
||||
|| ip.is_unspecified() // ::
|
||||
|| ip.is_multicast() // ff00::/8
|
||||
|| (ip.segments()[0] & 0xfe00) == 0xfc00 // fc00::/7 unique-local
|
||||
|| (ip.segments()[0] & 0xffc0) == 0xfe80 // fe80::/10 link-local
|
||||
}
|
||||
|
||||
/// Resolve `host` and return one validated, connectable address.
|
||||
/// Rejects the whole host if ANY resolved address is forbidden (fail-closed:
|
||||
/// prevents a host that returns one public + one internal record from slipping
|
||||
/// through on a later lookup).
|
||||
pub async fn resolve_and_validate(host: &str, port: u16) -> Result<SocketAddr, SsrfError> {
|
||||
// A bare IP literal in the URL must also be validated.
|
||||
if let Ok(ip) = host.parse::<IpAddr>() {
|
||||
if is_forbidden_ip(ip) {
|
||||
return Err(SsrfError::Blocked(ip));
|
||||
}
|
||||
return Ok(SocketAddr::new(ip, port));
|
||||
}
|
||||
|
||||
let resolver = match TokioAsyncResolver::tokio_from_system_conf() {
|
||||
Ok(r) => r,
|
||||
Err(_) => TokioAsyncResolver::tokio(ResolverConfig::default(), ResolverOpts::default()),
|
||||
};
|
||||
let lookup = resolver
|
||||
.lookup_ip(host)
|
||||
.await
|
||||
.map_err(|e| SsrfError::Resolve(e.to_string()))?;
|
||||
|
||||
let mut chosen: Option<IpAddr> = None;
|
||||
for ip in lookup.iter() {
|
||||
if is_forbidden_ip(ip) {
|
||||
return Err(SsrfError::Blocked(ip)); // fail-closed
|
||||
}
|
||||
if chosen.is_none() {
|
||||
chosen = Some(ip);
|
||||
}
|
||||
}
|
||||
match chosen {
|
||||
Some(ip) => Ok(SocketAddr::new(ip, port)),
|
||||
None => Err(SsrfError::NoSafeAddress),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn blocks_metadata_and_private() {
|
||||
for s in [
|
||||
"169.254.169.254", "127.0.0.1", "10.0.0.1", "192.168.1.1",
|
||||
"172.16.5.4", "0.0.0.0", "100.100.100.100", "::1",
|
||||
"fe80::1", "fc00::1", "::ffff:127.0.0.1",
|
||||
] {
|
||||
let ip: IpAddr = s.parse().unwrap();
|
||||
assert!(is_forbidden_ip(ip), "should block {s}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn allows_public() {
|
||||
for s in ["1.1.1.1", "104.21.7.81", "8.8.8.8", "2606:4700::1111"] {
|
||||
let ip: IpAddr = s.parse().unwrap();
|
||||
assert!(!is_forbidden_ip(ip), "should allow {s}");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user