sheathe_package/
origin.rs

1//! JIT / origin mode — package on HTTP request (Phase 5).
2//!
3//! A minimal HTTP/1.1 origin that serves:
4//! - `GET /health` → `ok`
5//! - `GET /package?input=<path>&…` → runs [`crate::package`] into a temp dir and
6//!   returns `master.m3u8` or `manifest.mpd` based on `Accept` / `format=`
7//! - `GET /out/<rel>` → serves a previously packaged object from the origin cache
8//!
9//! Optional HTTPS (`--tls-cert` / `--tls-key`) is rustls; optional Basic auth
10//! is `--auth user:password`. Not a production CDN.
11
12use crate::{PackageOptions, PresentationMode, package};
13use anyhow::{Context, Result, bail};
14use std::collections::HashMap;
15use std::fs;
16use std::io::{Read, Write};
17use std::net::{TcpListener, TcpStream};
18use std::path::{Path, PathBuf};
19use std::sync::{Arc, Mutex};
20use std::thread;
21
22/// Configuration for [`serve`].
23#[derive(Debug, Clone)]
24pub struct OriginConfig {
25    /// Bind address, e.g. `127.0.0.1:8787`.
26    pub bind: String,
27    /// Directory holding source media (path sandbox for `input=`).
28    pub media_root: PathBuf,
29    /// Cache directory for packaged outputs.
30    pub cache_dir: PathBuf,
31    /// Default segment duration seconds.
32    pub segment_duration: f64,
33    /// Optional TLS certificate PEM (enables HTTPS).
34    pub tls_cert: Option<PathBuf>,
35    /// Optional TLS private key PEM.
36    pub tls_key: Option<PathBuf>,
37    /// Optional `user:password` for HTTP Basic auth.
38    pub basic_auth: Option<String>,
39}
40
41impl Default for OriginConfig {
42    fn default() -> Self {
43        Self {
44            bind: "127.0.0.1:8787".into(),
45            media_root: PathBuf::from("."),
46            cache_dir: PathBuf::from("/tmp/sheathe-origin"),
47            segment_duration: 6.0,
48            tls_cert: None,
49            tls_key: None,
50            basic_auth: None,
51        }
52    }
53}
54
55/// Run the origin until the process is killed. Spawns one thread per connection.
56pub fn serve(cfg: OriginConfig) -> Result<()> {
57    fs::create_dir_all(&cfg.cache_dir)?;
58    let listener = TcpListener::bind(&cfg.bind).with_context(|| format!("bind {}", cfg.bind))?;
59    serve_listener(listener, cfg)
60}
61
62fn serve_listener(listener: TcpListener, cfg: OriginConfig) -> Result<()> {
63    let tls = match (&cfg.tls_cert, &cfg.tls_key) {
64        (None, None) => None,
65        (Some(cert), Some(key)) => Some(crate::tls::server_config(cert, key)?),
66        _ => bail!("origin TLS requires both --tls-cert and --tls-key"),
67    };
68    let scheme = if tls.is_some() { "https" } else { "http" };
69    let addr = listener.local_addr().map(|a| a.to_string()).unwrap_or_else(|_| cfg.bind.clone());
70    eprintln!("sheathe origin listening on {scheme}://{addr}/");
71    let cfg = Arc::new(cfg);
72    let lock = Arc::new(Mutex::new(()));
73    for conn in listener.incoming() {
74        let stream = conn.context("accept")?;
75        let cfg = Arc::clone(&cfg);
76        let lock = Arc::clone(&lock);
77        let tls = tls.clone();
78        thread::spawn(move || {
79            if let Err(e) = dispatch_client(stream, tls, &cfg, &lock) {
80                eprintln!("origin: {e:#}");
81            }
82        });
83    }
84    Ok(())
85}
86
87fn dispatch_client(
88    stream: TcpStream,
89    tls: Option<Arc<rustls::ServerConfig>>,
90    cfg: &OriginConfig,
91    lock: &Mutex<()>,
92) -> Result<()> {
93    match tls {
94        Some(tls) => {
95            let conn = rustls::ServerConnection::new(tls).context("TLS server")?;
96            let mut tls_stream = rustls::StreamOwned::new(conn, stream);
97            handle_client(&mut tls_stream, cfg, lock)
98        }
99        None => {
100            let mut stream = stream;
101            handle_client(&mut stream, cfg, lock)
102        }
103    }
104}
105
106fn handle_client<S: Read + Write>(
107    stream: &mut S,
108    cfg: &OriginConfig,
109    lock: &Mutex<()>,
110) -> Result<()> {
111    let mut buf = [0u8; 8192];
112    let n = stream.read(&mut buf)?;
113    let req = String::from_utf8_lossy(&buf[..n]);
114    if let Some(auth) = &cfg.basic_auth
115        && !authorized(&req, auth)
116    {
117        return respond(
118            stream,
119            401,
120            "text/plain",
121            b"unauthorized\n",
122            "WWW-Authenticate: Basic realm=\"sheathe\"\r\n",
123        );
124    }
125    let line = req.lines().next().unwrap_or("");
126    let mut parts = line.split_whitespace();
127    let method = parts.next().unwrap_or("");
128    let target = parts.next().unwrap_or("/");
129    if method != "GET" && method != "HEAD" {
130        return respond(stream, 405, "text/plain", b"method not allowed", "");
131    }
132
133    let (path, query) = target.split_once('?').unwrap_or((target, ""));
134    let q = parse_query(query);
135
136    match path {
137        "/health" | "/healthz" => respond(stream, 200, "text/plain", b"ok\n", ""),
138        "/package" => {
139            let input = q.get("input").map(String::as_str).context("missing input=")?;
140            let media = resolve_media(&cfg.media_root, input)?;
141            let format = q.get("format").map(String::as_str).unwrap_or("hls");
142            let key = cache_key(&media, format, cfg.segment_duration);
143            let out_dir = cfg.cache_dir.join(&key);
144            {
145                let _g = lock.lock().unwrap_or_else(|e| e.into_inner());
146                if !out_dir.join("master.m3u8").exists() && !out_dir.join("manifest.mpd").exists() {
147                    let opts = PackageOptions {
148                        out_dir: out_dir.clone(),
149                        segment_duration: cfg.segment_duration,
150                        dash: format == "dash" || format == "both",
151                        hls: format == "hls" || format == "both",
152                        presentation: PresentationMode::Vod,
153                        ..PackageOptions::default()
154                    };
155                    package(&[media], &opts)?;
156                }
157            }
158            let body_path = if format == "dash" {
159                out_dir.join("manifest.mpd")
160            } else {
161                out_dir.join("master.m3u8")
162            };
163            let body =
164                fs::read(&body_path).with_context(|| format!("read {}", body_path.display()))?;
165            let ctype = if format == "dash" {
166                "application/dash+xml"
167            } else {
168                "application/vnd.apple.mpegurl"
169            };
170            respond(stream, 200, ctype, &body, "")
171        }
172        p if p.starts_with("/out/") => {
173            let rel = &p["/out/".len()..];
174            let path = cfg.cache_dir.join(rel);
175            // Prevent path escape.
176            let canon_cache =
177                cfg.cache_dir.canonicalize().unwrap_or_else(|_| cfg.cache_dir.clone());
178            let canon = path.canonicalize().with_context(|| format!("missing {rel}"))?;
179            if !canon.starts_with(&canon_cache) {
180                bail!("path escape");
181            }
182            let body = fs::read(&canon)?;
183            respond(stream, 200, guess_ctype(rel), &body, "")
184        }
185        _ => respond(
186            stream,
187            404,
188            "text/plain",
189            b"not found\nGET /health | /package?input=FILE&format=hls|dash | /out/<cache-rel>\n",
190            "",
191        ),
192    }
193}
194
195fn resolve_media(root: &Path, input: &str) -> Result<PathBuf> {
196    let p = Path::new(input);
197    let full = if p.is_absolute() { p.to_path_buf() } else { root.join(p) };
198    let canon = full.canonicalize().with_context(|| format!("input not found: {input}"))?;
199    let root_canon = root.canonicalize().unwrap_or_else(|_| root.to_path_buf());
200    if !canon.starts_with(&root_canon) && !Path::new(input).is_absolute() {
201        bail!("input escapes media_root");
202    }
203    Ok(canon)
204}
205
206fn cache_key(media: &Path, format: &str, seg: f64) -> String {
207    use std::collections::hash_map::DefaultHasher;
208    use std::hash::{Hash, Hasher};
209    let mut h = DefaultHasher::new();
210    media.hash(&mut h);
211    format.hash(&mut h);
212    seg.to_bits().hash(&mut h);
213    format!("{:016x}", h.finish())
214}
215
216fn parse_query(q: &str) -> HashMap<String, String> {
217    let mut m = HashMap::new();
218    for pair in q.split('&').filter(|s| !s.is_empty()) {
219        if let Some((k, v)) = pair.split_once('=') {
220            m.insert(url_decode(k), url_decode(v));
221        }
222    }
223    m
224}
225
226fn url_decode(s: &str) -> String {
227    let mut out = String::new();
228    let b = s.as_bytes();
229    let mut i = 0;
230    while i < b.len() {
231        match b[i] {
232            b'+' => {
233                out.push(' ');
234                i += 1;
235            }
236            b'%' if i + 2 < b.len() => {
237                let hex = &s[i + 1..i + 3];
238                if let Ok(v) = u8::from_str_radix(hex, 16) {
239                    out.push(v as char);
240                    i += 3;
241                } else {
242                    out.push('%');
243                    i += 1;
244                }
245            }
246            c => {
247                out.push(c as char);
248                i += 1;
249            }
250        }
251    }
252    out
253}
254
255fn guess_ctype(path: &str) -> &'static str {
256    if path.ends_with(".m3u8") {
257        "application/vnd.apple.mpegurl"
258    } else if path.ends_with(".mpd") {
259        "application/dash+xml"
260    } else if path.ends_with(".mp4") || path.ends_with(".m4s") {
261        "video/mp4"
262    } else if path.ends_with(".ts") {
263        "video/mp2t"
264    } else {
265        "application/octet-stream"
266    }
267}
268
269fn authorized(req: &str, user_pass: &str) -> bool {
270    let want = b64_basic(user_pass.as_bytes());
271    req.lines().any(|line| {
272        let Some((name, value)) = line.split_once(':') else { return false };
273        if !name.eq_ignore_ascii_case("authorization") {
274            return false;
275        }
276        let value = value.trim();
277        value
278            .strip_prefix("Basic ")
279            .or_else(|| value.strip_prefix("basic "))
280            .is_some_and(|b64| b64.trim() == want)
281    })
282}
283
284fn b64_basic(raw: &[u8]) -> String {
285    const A: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
286    let mut b64 = String::new();
287    for chunk in raw.chunks(3) {
288        let x = [chunk[0], *chunk.get(1).unwrap_or(&0), *chunk.get(2).unwrap_or(&0)];
289        let n = (u32::from(x[0]) << 16) | (u32::from(x[1]) << 8) | u32::from(x[2]);
290        b64.push(A[(n >> 18 & 0x3f) as usize] as char);
291        b64.push(A[(n >> 12 & 0x3f) as usize] as char);
292        b64.push(if chunk.len() > 1 { A[(n >> 6 & 0x3f) as usize] as char } else { '=' });
293        b64.push(if chunk.len() > 2 { A[(n & 0x3f) as usize] as char } else { '=' });
294    }
295    b64
296}
297
298fn respond<S: Write>(
299    stream: &mut S,
300    status: u16,
301    ctype: &str,
302    body: &[u8],
303    extra_headers: &str,
304) -> Result<()> {
305    let reason = match status {
306        200 => "OK",
307        401 => "Unauthorized",
308        404 => "Not Found",
309        405 => "Method Not Allowed",
310        _ => "Error",
311    };
312    let header = format!(
313        "HTTP/1.1 {status} {reason}\r\nContent-Type: {ctype}\r\nContent-Length: {}\r\nConnection: close\r\nAccess-Control-Allow-Origin: *\r\n{extra_headers}\r\n",
314        body.len()
315    );
316    stream.write_all(header.as_bytes())?;
317    stream.write_all(body)?;
318    Ok(())
319}
320
321#[cfg(test)]
322mod tests {
323    use super::*;
324    use std::net::SocketAddr;
325    use std::time::Duration;
326
327    fn spawn_origin(cfg: OriginConfig) -> SocketAddr {
328        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
329        let addr = listener.local_addr().unwrap();
330        thread::spawn(move || {
331            let _ = serve_listener(listener, cfg);
332        });
333        // Accept loop is ready as soon as bind succeeded.
334        addr
335    }
336
337    fn http_exchange(addr: SocketAddr, req: &str) -> String {
338        let mut s = TcpStream::connect(addr).unwrap();
339        s.set_read_timeout(Some(Duration::from_secs(2))).ok();
340        s.set_write_timeout(Some(Duration::from_secs(2))).ok();
341        s.write_all(req.as_bytes()).unwrap();
342        let mut buf = Vec::new();
343        s.read_to_end(&mut buf).ok();
344        String::from_utf8_lossy(&buf).into_owned()
345    }
346
347    #[test]
348    fn health_over_plain_http() {
349        let cfg = OriginConfig { bind: "127.0.0.1:0".into(), ..OriginConfig::default() };
350        let addr = spawn_origin(cfg);
351        let resp = http_exchange(
352            addr,
353            "GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
354        );
355        assert!(resp.contains("HTTP/1.1 200"), "{resp}");
356        assert!(resp.contains("ok"), "{resp}");
357    }
358
359    #[test]
360    fn basic_auth_rejects_and_accepts() {
361        let cfg = OriginConfig {
362            bind: "127.0.0.1:0".into(),
363            basic_auth: Some("alice:secret".into()),
364            ..OriginConfig::default()
365        };
366        let addr = spawn_origin(cfg);
367        let denied = http_exchange(
368            addr,
369            "GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n",
370        );
371        assert!(denied.contains("401"), "{denied}");
372        assert!(denied.contains("WWW-Authenticate: Basic"), "{denied}");
373
374        let cred = b64_basic(b"alice:secret");
375        let ok = http_exchange(
376            addr,
377            &format!(
378                "GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\nAuthorization: Basic {cred}\r\nConnection: close\r\n\r\n"
379            ),
380        );
381        assert!(ok.contains("HTTP/1.1 200"), "{ok}");
382    }
383
384    #[test]
385    fn tls_requires_both_cert_and_key() {
386        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
387        let cfg = OriginConfig {
388            tls_cert: Some(PathBuf::from("/tmp/sheathe-missing.pem")),
389            ..OriginConfig::default()
390        };
391        let err = serve_listener(listener, cfg).unwrap_err();
392        assert!(err.to_string().contains("both"), "{err:#}");
393    }
394
395    #[test]
396    fn health_over_https() {
397        let ck = rcgen::generate_simple_self_signed(["localhost".into()]).unwrap();
398        let dir = std::env::temp_dir().join(format!("sheathe-origin-tls-{}", std::process::id()));
399        fs::create_dir_all(&dir).unwrap();
400        let cert_path = dir.join("cert.pem");
401        let key_path = dir.join("key.pem");
402        fs::write(&cert_path, ck.cert.pem()).unwrap();
403        fs::write(&key_path, ck.key_pair.serialize_pem()).unwrap();
404
405        let cfg = OriginConfig {
406            bind: "127.0.0.1:0".into(),
407            tls_cert: Some(cert_path.clone()),
408            tls_key: Some(key_path),
409            ..OriginConfig::default()
410        };
411        let addr = spawn_origin(cfg);
412
413        let mut roots = rustls::RootCertStore::empty();
414        let pem = fs::read(&cert_path).unwrap();
415        for cert in rustls_pemfile::certs(&mut pem.as_slice()).flatten() {
416            roots.add(cert).unwrap();
417        }
418        let config =
419            rustls::ClientConfig::builder().with_root_certificates(roots).with_no_client_auth();
420        let tcp = TcpStream::connect(addr).unwrap();
421        tcp.set_read_timeout(Some(Duration::from_secs(2))).ok();
422        let name = rustls::pki_types::ServerName::try_from("localhost").unwrap();
423        let conn = rustls::ClientConnection::new(Arc::new(config), name).unwrap();
424        let mut tls = rustls::StreamOwned::new(conn, tcp);
425        tls.write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
426            .unwrap();
427        let mut buf = Vec::new();
428        tls.read_to_end(&mut buf).ok();
429        let resp = String::from_utf8_lossy(&buf);
430        assert!(resp.contains("HTTP/1.1 200"), "{resp}");
431        assert!(resp.contains("ok"), "{resp}");
432    }
433}
sheathe_package/origin.rs

sheathe_package/
origin.rs
