622 lines
36 KiB
Rust
622 lines
36 KiB
Rust
//! The `wai http` commands run as a user runs them (`--features http --test cli_http`):
|
|
//!
|
|
//! - `wai http publish` writes `<cid>.wai`, its head proof and its ETag sidecar, named by
|
|
//! the content hash, and refuses an envelope with a byte after it (§5.1);
|
|
//! - `wai http codecs` prints a sequence's union and floor (§4.2);
|
|
//! - `wai http origin` serves an object with §5.3's headers, byte ranges and `If-Range`,
|
|
//! and `wai http bench` fetches it through the byte-range procedure (§7) and writes a
|
|
//! record with the pick, the bytes by category and energy stated as not measured;
|
|
//! - the origin reads at most 16 KiB of a request head, gives a whole head one deadline
|
|
//! from the accept and a whole connection one lifetime, caps the connections it serves
|
|
//! at once (and those of one peer only when asked), percent-decodes `/__hits` as bytes,
|
|
//! and refuses a path segment with `\`, `:`, NUL or a control byte. Every origin a test
|
|
//! starts is stopped when the test ends, whether or not it passed.
|
|
|
|
use std::io::{BufRead, BufReader, Read, Write};
|
|
use std::net::TcpStream;
|
|
use std::path::{Path, PathBuf};
|
|
use std::process::{Command, Output, Stdio};
|
|
|
|
use wai::http_binding::head_proof::HeadProof;
|
|
use wai::http_binding::{content_hash, hex};
|
|
|
|
fn wai(args: &[&str]) -> Output {
|
|
Command::new(env!("CARGO_BIN_EXE_wai")).args(args).output().expect("run wai")
|
|
}
|
|
|
|
fn corpus() -> PathBuf {
|
|
Path::new(env!("CARGO_MANIFEST_DIR")).join("../http-conformance")
|
|
}
|
|
|
|
fn object(id: &str) -> PathBuf {
|
|
let listing: serde_json::Value = serde_json::from_slice(&std::fs::read(corpus().join("objects.json")).unwrap()).unwrap();
|
|
corpus().join(listing[id]["path"].as_str().unwrap())
|
|
}
|
|
|
|
fn scratch(name: &str) -> PathBuf {
|
|
let d = Path::new(env!("CARGO_TARGET_TMPDIR")).join("cli_http").join(name);
|
|
let _ = std::fs::remove_dir_all(&d);
|
|
std::fs::create_dir_all(&d).unwrap();
|
|
d
|
|
}
|
|
|
|
#[test]
|
|
fn publish_names_by_content_hash() {
|
|
let out = scratch("publish");
|
|
let src = object("O3");
|
|
let run = wai(&["http", "publish", src.to_str().unwrap(), "--out", out.to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let env = std::fs::read(&src).unwrap();
|
|
let cid = hex(&content_hash(&env));
|
|
assert_eq!(std::fs::read(out.join(format!("{cid}.wai"))).unwrap(), env);
|
|
assert_eq!(std::fs::read(out.join(format!("{cid}.wai.hp"))).unwrap(), HeadProof::build(&env).unwrap().to_bytes());
|
|
assert_eq!(std::fs::read_to_string(out.join(format!("{cid}.wai.etag"))).unwrap(), format!("\"{cid}\"\n"));
|
|
let line = String::from_utf8_lossy(&run.stdout);
|
|
assert!(line.starts_with(&cid) && line.contains(&format!("n={}", env.len())) && line.contains("wai2.video.av1+meta.hdr10plus"), "{line}");
|
|
// One byte after the envelope: not an object.
|
|
let bad = corpus().join("bad/O6.wai");
|
|
let run = wai(&["http", "publish", bad.to_str().unwrap(), "--out", out.to_str().unwrap()]);
|
|
assert!(!run.status.success());
|
|
assert!(String::from_utf8_lossy(&run.stderr).contains("object_length_mismatch"));
|
|
}
|
|
|
|
#[test]
|
|
fn codecs_prints_union_and_floor() {
|
|
let run = wai(&["http", "codecs", object("O9a").to_str().unwrap(), object("O9b").to_str().unwrap(), object("O9c").to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
assert_eq!(String::from_utf8_lossy(&run.stdout), "union\twai2.video.av1|video.int_hyper\nfloor\twai2.video.av1\n");
|
|
}
|
|
|
|
/// A running `wai http origin`, stopped when dropped, so a failing assertion never leaves
|
|
/// it running.
|
|
struct Origin {
|
|
child: std::process::Child,
|
|
port: u16,
|
|
}
|
|
|
|
impl Origin {
|
|
fn start(root: &Path, extra: &[&str]) -> Origin {
|
|
let mut child = Command::new(env!("CARGO_BIN_EXE_wai"))
|
|
.args(["http", "origin", "--root", root.to_str().unwrap(), "--port", "0"])
|
|
.args(extra)
|
|
.stdout(Stdio::piped())
|
|
.spawn()
|
|
.unwrap();
|
|
let mut first = String::new();
|
|
BufReader::new(child.stdout.take().unwrap()).read_line(&mut first).unwrap();
|
|
let port: u16 = first.trim().rsplit(':').next().unwrap().trim_end_matches('/').parse().unwrap();
|
|
Origin { child, port }
|
|
}
|
|
}
|
|
|
|
impl Drop for Origin {
|
|
fn drop(&mut self) {
|
|
let _ = self.child.kill();
|
|
let _ = self.child.wait();
|
|
}
|
|
}
|
|
|
|
/// Send raw request bytes; return what the origin answered before closing.
|
|
fn raw(port: u16, request: &[u8]) -> String {
|
|
let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
|
|
s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
|
|
s.write_all(request).unwrap();
|
|
let mut out = Vec::new();
|
|
let _ = s.read_to_end(&mut out);
|
|
String::from_utf8_lossy(&out).into_owned()
|
|
}
|
|
|
|
/// One request to the origin: status line and headers, and the body.
|
|
fn get(port: u16, path: &str, extra: &[(&str, &str)]) -> (String, Vec<u8>) {
|
|
let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
|
|
let mut req = format!("GET {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n");
|
|
for (k, v) in extra {
|
|
req.push_str(&format!("{k}: {v}\r\n"));
|
|
}
|
|
req.push_str("\r\n");
|
|
s.write_all(req.as_bytes()).unwrap();
|
|
let mut raw = Vec::new();
|
|
s.read_to_end(&mut raw).unwrap();
|
|
let end = raw.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
|
|
(String::from_utf8_lossy(&raw[..end]).into_owned(), raw[end..].to_vec())
|
|
}
|
|
|
|
#[test]
|
|
fn origin_serves_and_bench_fetches() {
|
|
let root = scratch("origin");
|
|
let src = object("OD");
|
|
let env = std::fs::read(&src).unwrap();
|
|
let pubd = wai(&["http", "publish", src.to_str().unwrap(), "--out", root.to_str().unwrap()]);
|
|
assert!(pubd.status.success());
|
|
let cid = hex(&content_hash(&env));
|
|
let origin = Origin::start(&root, &[]);
|
|
let port = origin.port;
|
|
let path = format!("/{cid}.wai");
|
|
|
|
let (h, body) = get(port, &path, &[]);
|
|
assert!(h.starts_with("HTTP/1.1 200") && body == env, "{h}");
|
|
for want in ["Cache-Control: public, max-age=31536000, immutable, no-transform", &format!("ETag: \"{cid}\""), "Accept-Ranges: bytes", "Access-Control-Allow-Origin: *"] {
|
|
assert!(h.contains(want), "{want} missing from {h}");
|
|
}
|
|
assert!(!h.contains("Vary") && !h.contains("Content-Encoding"), "{h}");
|
|
let (h, body) = get(port, &path, &[("Range", "bytes=1009-3593")]);
|
|
assert!(h.starts_with("HTTP/1.1 206") && h.contains("Content-Range: bytes 1009-3593/53794") && body == env[1009..3594], "{h}");
|
|
let (h, _) = get(port, &path, &[("Range", "bytes=1009-3593"), ("If-Range", "\"not-this\"")]);
|
|
assert!(h.starts_with("HTTP/1.1 200"), "{h}");
|
|
let (h, _) = get(port, &path, &[("Range", "bytes=53794-")]);
|
|
assert!(h.starts_with("HTTP/1.1 416") && h.contains("Content-Range: bytes */53794"), "{h}");
|
|
|
|
let rec = root.join("bench.json");
|
|
let url = format!("http://127.0.0.1:{port}{path}#h=1009&n=53794");
|
|
let run = wai(&["http", "bench", "--url", &url, "--runs", "3", "--advertise", "wai.video.av1", "--out", rec.to_str().unwrap()]);
|
|
drop(origin);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let r: serde_json::Value = serde_json::from_slice(&std::fs::read(&rec).unwrap()).unwrap();
|
|
// Appendix D's plan 1: the head, the proof and `low`'s rest.
|
|
assert_eq!(r["pick"], 0);
|
|
assert_eq!(r["body_bytes_media"]["median"], 1009.0 + 2585.0);
|
|
assert_eq!(r["body_bytes_proof"]["median"], 245.0);
|
|
assert_eq!(r["requests"], 3);
|
|
assert_eq!(r["energy"], "omitted: not measured");
|
|
assert_eq!(r["envelope_len"], 53_794);
|
|
}
|
|
|
|
|
|
/// The origin reads at most 16 KiB of a request head: a 400 MiB header is answered 431
|
|
/// and the connection closed long before it is sent.
|
|
#[test]
|
|
fn origin_refuses_an_oversized_head() {
|
|
let root = scratch("origin-big");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &[]);
|
|
let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
|
|
s.set_read_timeout(Some(std::time::Duration::from_secs(20))).unwrap();
|
|
let mut w = s.try_clone().unwrap();
|
|
let t0 = std::time::Instant::now();
|
|
let writer = std::thread::spawn(move || {
|
|
let _ = w.write_all(b"GET /x.json HTTP/1.1\r\nHost: a\r\nX-Fill: ");
|
|
let chunk = vec![b'a'; 1 << 20];
|
|
let mut sent = 0usize;
|
|
while sent < 400 << 20 {
|
|
if w.write_all(&chunk).is_err() {
|
|
break;
|
|
}
|
|
sent += chunk.len();
|
|
}
|
|
sent
|
|
});
|
|
let mut head = [0u8; 64];
|
|
let n = s.read(&mut head).unwrap_or(0);
|
|
let answer = String::from_utf8_lossy(&head[..n]).into_owned();
|
|
drop(s);
|
|
let sent = writer.join().unwrap();
|
|
assert!(answer.starts_with("HTTP/1.1 431"), "{answer:?}");
|
|
assert!(sent < 400 << 20, "the origin read the whole 400 MiB");
|
|
assert!(t0.elapsed() < std::time::Duration::from_secs(30));
|
|
// And it still serves.
|
|
assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
|
|
}
|
|
|
|
/// `/__hits?path=` is percent-decoded as bytes: an escape before a non-ASCII character, or
|
|
/// an invalid UTF-8 sequence, is answered, not a panic.
|
|
#[test]
|
|
fn origin_decodes_hits_as_bytes() {
|
|
let root = scratch("origin-hits");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &["--count-hits"]);
|
|
for q in ["%aé", "%ff%fe", "%", "%4", "%e9%"] {
|
|
let r = raw(origin.port, format!("GET /__hits?path={q} HTTP/1.1\r\nHost: a\r\n\r\n").as_bytes());
|
|
assert!(r.starts_with("HTTP/1.1 200"), "{q}: {r:?}");
|
|
}
|
|
let _ = raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
|
|
let r = raw(origin.port, b"GET /__hits?path=%2Fx.json HTTP/1.1\r\nHost: a\r\n\r\n");
|
|
assert!(r.ends_with("\r\n\r\n1"), "{r:?}");
|
|
}
|
|
|
|
/// A path segment holding `\`, `:`, NUL or a control byte, or `.`/`..`, names nothing:
|
|
/// 400.
|
|
#[test]
|
|
fn origin_refuses_backslash_drive_and_nul_paths() {
|
|
let root = scratch("origin-paths");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &[]);
|
|
for path in [&b"/..\\..\\etc\\passwd"[..], b"/C:/Windows/win.ini", b"/x.json:stream", b"/a\x00b", b"/a\x01b", b"/./x.json", b"/../x.json", b"//x.json", b"/x.json/"] {
|
|
let mut req = b"GET ".to_vec();
|
|
req.extend_from_slice(path);
|
|
req.extend_from_slice(b" HTTP/1.1\r\nHost: a\r\n\r\n");
|
|
let r = raw(origin.port, &req);
|
|
assert!(r.starts_with("HTTP/1.1 400"), "{}: {r:?}", String::from_utf8_lossy(path));
|
|
}
|
|
assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
|
|
}
|
|
|
|
/// At most `--max-connections` at once (one more is answered 503), and a connection that
|
|
/// sends nothing is answered 408 and closed `--read-timeout-ms` after its accept.
|
|
#[test]
|
|
fn origin_caps_connections_and_times_out() {
|
|
let root = scratch("origin-cap");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &["--max-connections", "2", "--read-timeout-ms", "700"]);
|
|
let idle: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(("127.0.0.1", origin.port)).unwrap()).collect();
|
|
std::thread::sleep(std::time::Duration::from_millis(200));
|
|
let r = raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
|
|
assert!(r.starts_with("HTTP/1.1 503"), "{r:?}");
|
|
let t0 = std::time::Instant::now();
|
|
for mut c in idle {
|
|
// Told 408, and closed.
|
|
c.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
|
|
let mut out = Vec::new();
|
|
let _ = c.read_to_end(&mut out);
|
|
assert!(out.starts_with(b"HTTP/1.1 408"), "{:?}", String::from_utf8_lossy(&out));
|
|
}
|
|
assert!(t0.elapsed() < std::time::Duration::from_secs(5));
|
|
assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
|
|
}
|
|
|
|
/// N1: a whole request head has one deadline from the accept, so connections that send a
|
|
/// byte at a time cannot hold the connection cap: a GET succeeds within about one head
|
|
/// deadline while they keep trickling.
|
|
#[test]
|
|
fn origin_survives_trickling_heads() {
|
|
let root = scratch("origin-trickle");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &["--max-connections", "4", "--read-timeout-ms", "1000"]);
|
|
let port = origin.port;
|
|
let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let tricklers: Vec<_> = (0..4).map(|_| {
|
|
let stop = stop.clone();
|
|
std::thread::spawn(move || {
|
|
let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
|
|
let mut sent = 0;
|
|
// One byte every 100 ms, never ending the head, until the origin closes.
|
|
for b in b"GET /x.json HTTP/1.1\r\nX-Slow: ".iter().chain(std::iter::repeat(&b'a')) {
|
|
if stop.load(std::sync::atomic::Ordering::SeqCst) || s.write_all(&[*b]).is_err() {
|
|
break;
|
|
}
|
|
sent += 1;
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
|
}
|
|
sent
|
|
})
|
|
}).collect();
|
|
std::thread::sleep(std::time::Duration::from_millis(200));
|
|
let t0 = std::time::Instant::now();
|
|
let mut answer = String::new();
|
|
while t0.elapsed() < std::time::Duration::from_secs(5) {
|
|
answer = raw(port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
|
|
if answer.starts_with("HTTP/1.1 200") {
|
|
break;
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(50));
|
|
}
|
|
let waited = t0.elapsed();
|
|
stop.store(true, std::sync::atomic::Ordering::SeqCst);
|
|
for t in tricklers {
|
|
t.join().unwrap();
|
|
}
|
|
assert!(answer.starts_with("HTTP/1.1 200"), "{answer:?}");
|
|
assert!(waited < std::time::Duration::from_millis(2500), "the GET waited {waited:?}");
|
|
}
|
|
|
|
/// R3-3: with its defaults (no cap per peer), eight connections that trickle a head and
|
|
/// reconnect the moment they are closed never shut a legitimate GET out: on a loopback
|
|
/// listener every client has their address.
|
|
#[test]
|
|
fn origin_serves_past_reconnecting_tricklers() {
|
|
let root = scratch("origin-reconnect");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &["--read-timeout-ms", "1000"]);
|
|
let port = origin.port;
|
|
let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let tricklers: Vec<_> = (0..8).map(|_| {
|
|
let stop = stop.clone();
|
|
std::thread::spawn(move || {
|
|
while !stop.load(std::sync::atomic::Ordering::SeqCst) {
|
|
let Ok(mut s) = TcpStream::connect(("127.0.0.1", port)) else { continue };
|
|
for b in b"GET /x.json HTTP/1.1\r\nX-Slow: ".iter().chain(std::iter::repeat(&b'a')) {
|
|
if stop.load(std::sync::atomic::Ordering::SeqCst) || s.write_all(&[*b]).is_err() {
|
|
break;
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(50));
|
|
}
|
|
}
|
|
})
|
|
}).collect();
|
|
std::thread::sleep(std::time::Duration::from_millis(300));
|
|
let t0 = std::time::Instant::now();
|
|
let mut answers = Vec::new();
|
|
while t0.elapsed() < std::time::Duration::from_secs(3) {
|
|
answers.push(raw(port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").lines().next().unwrap_or("").to_string());
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
|
}
|
|
stop.store(true, std::sync::atomic::Ordering::SeqCst);
|
|
for t in tricklers {
|
|
t.join().unwrap();
|
|
}
|
|
assert!(answers.len() >= 10 && answers.iter().all(|a| a.starts_with("HTTP/1.1 200")), "{answers:?}");
|
|
}
|
|
|
|
/// R3-3: a connection lives at most `--max-connection-ms` from its accept, so a peer that
|
|
/// reads a large response slowly is cut off then, not served for as long as it keeps
|
|
/// reading.
|
|
#[test]
|
|
fn origin_cuts_a_slow_reader_at_the_lifetime() {
|
|
let root = scratch("origin-slow-reader");
|
|
let size = 32usize << 20;
|
|
std::fs::write(root.join("big.bin"), vec![7u8; size]).unwrap();
|
|
let origin = Origin::start(&root, &["--max-connection-ms", "1500"]);
|
|
let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
|
|
s.write_all(b"GET /big.bin HTTP/1.1\r\nHost: a\r\n\r\n").unwrap();
|
|
s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
|
|
let (t0, mut got, mut buf) = (std::time::Instant::now(), Vec::new(), [0u8; 4096]);
|
|
// Slowly, past the lifetime: 4 KiB every 100 ms.
|
|
while t0.elapsed() < std::time::Duration::from_millis(2500) {
|
|
match s.read(&mut buf) {
|
|
Ok(0) | Err(_) => break,
|
|
Ok(n) => got.extend_from_slice(&buf[..n]),
|
|
}
|
|
std::thread::sleep(std::time::Duration::from_millis(100));
|
|
}
|
|
// Then as fast as it comes: what was already sent, and the end of the connection.
|
|
let ended = loop {
|
|
match s.read(&mut buf) {
|
|
Ok(0) => break true,
|
|
Ok(n) => got.extend_from_slice(&buf[..n]),
|
|
Err(e) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => break false,
|
|
Err(_) => break true,
|
|
}
|
|
};
|
|
assert!(got.starts_with(b"HTTP/1.1 200"), "{:?}", String::from_utf8_lossy(&got[..got.len().min(64)]));
|
|
assert!(ended, "the connection was still open");
|
|
assert!(got.len() < size, "the whole response arrived: the connection outlived its lifetime");
|
|
}
|
|
|
|
/// N1: at most `--max-per-peer` connections from one address, when it is given; one more
|
|
/// is answered 503.
|
|
#[test]
|
|
fn origin_per_peer_limit() {
|
|
let root = scratch("origin-peer");
|
|
std::fs::write(root.join("x.json"), b"{}").unwrap();
|
|
let origin = Origin::start(&root, &["--max-per-peer", "2", "--read-timeout-ms", "2000"]);
|
|
let idle: Vec<TcpStream> = (0..2).map(|_| TcpStream::connect(("127.0.0.1", origin.port)).unwrap()).collect();
|
|
std::thread::sleep(std::time::Duration::from_millis(200));
|
|
let r = raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n");
|
|
assert!(r.starts_with("HTTP/1.1 503"), "{r:?}");
|
|
drop(idle);
|
|
std::thread::sleep(std::time::Duration::from_millis(200));
|
|
assert!(raw(origin.port, b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n").starts_with("HTTP/1.1 200"));
|
|
}
|
|
|
|
/// N3: `wai http bench` times the base from the session's `base_ready`: with the origin
|
|
/// holding the enhancement's bytes back, the base's time does not include the delay, and
|
|
/// the time to complete does.
|
|
#[test]
|
|
fn bench_times_the_base_apart_from_its_refinement() {
|
|
use wai::http_binding::head::{parse_head, HeadParse};
|
|
let root = scratch("bench-delay");
|
|
let src = object("O8");
|
|
let env = std::fs::read(&src).unwrap();
|
|
assert!(wai(&["http", "publish", src.to_str().unwrap(), "--out", root.to_str().unwrap()]).status.success());
|
|
let HeadParse::Head(h) = parse_head(&env, env.len() as u64).unwrap() else { panic!("a head") };
|
|
let enh = h.components[1];
|
|
let delay = format!("{}-{}:1500", enh.start, enh.start + enh.len - 1);
|
|
let origin = Origin::start(&root, &["--delay-range", &delay]);
|
|
let rec = root.join("bench.json");
|
|
let url = format!("http://127.0.0.1:{}/{}.wai#h={}&n={}", origin.port, hex(&content_hash(&env)), h.covered_len(), env.len());
|
|
let run = wai(&["http", "bench", "--url", &url, "--runs", "2", "--advertise", "wai.video.avc", "--out", rec.to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let r: serde_json::Value = serde_json::from_slice(&std::fs::read(&rec).unwrap()).unwrap();
|
|
let base = r["ms_to_verified"]["median"].as_f64().unwrap();
|
|
let all = r["ms_to_complete"]["median"].as_f64().unwrap();
|
|
assert!(base < 1000.0, "the base's time includes the enhancement's delay: {base} ms");
|
|
assert!(all >= 1500.0, "the enhancement was not delayed: {all} ms");
|
|
// With a refinement deadline shorter than the delay, the enhancement is abandoned.
|
|
let run = wai(&["http", "bench", "--url", &url, "--runs", "1", "--advertise", "wai.video.avc", "--refinement-deadline-ms", "1", "--out", rec.to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let r: serde_json::Value = serde_json::from_slice(&std::fs::read(&rec).unwrap()).unwrap();
|
|
assert_eq!(r["refinements_abandoned"], serde_json::json!([1]));
|
|
assert!(r["ms_to_complete"]["median"].as_f64().unwrap() < 1000.0);
|
|
drop(origin);
|
|
}
|
|
|
|
/// A classical playlist for the O9 segments (three of 640 ms), and an objects list over
|
|
/// them, written in a scratch directory with the corpus objects beside them.
|
|
fn o9_presentation(name: &str) -> PathBuf {
|
|
let d = scratch(name);
|
|
std::fs::write(d.join("classical.m3u8"), "#EXTM3U\n#EXT-X-VERSION:7\n#EXT-X-TARGETDURATION:1\n#EXT-X-MEDIA-SEQUENCE:0\n#EXTINF:0.640,\nc0.m4s\n#EXTINF:0.640,\nc1.m4s\n#EXTINF:0.640,\nc2.m4s\n#EXT-X-ENDLIST\n").unwrap();
|
|
let objs: Vec<String> = ["O9a", "O9b", "O9c"].iter().map(|id| object(id).to_str().unwrap().to_string()).collect();
|
|
let list = serde_json::json!({
|
|
"uri_prefix": "objects/",
|
|
"variants": [{ "uri": "wai-180.m3u8", "classical": "classical.m3u8", "resolution": [320, 180], "frame_rate_milli": 25000,
|
|
"segments": objs.iter().map(|o| serde_json::json!({ "object": o })).collect::<Vec<_>>() }],
|
|
});
|
|
std::fs::write(d.join("list.json"), serde_json::to_vec_pretty(&list).unwrap()).unwrap();
|
|
d
|
|
}
|
|
|
|
/// `wai http hls` writes the WAI media playlist aligned with the classical one, and
|
|
/// `wai.m3u8` with the union, the floor and computed bandwidths; `wai http mpd` writes
|
|
/// `wai.mpd` on the same timeline.
|
|
#[test]
|
|
fn hls_and_mpd_write_aligned_playlists() {
|
|
let d = o9_presentation("hls");
|
|
let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let media = std::fs::read_to_string(d.join("wai-180.m3u8")).unwrap();
|
|
let classical = std::fs::read_to_string(d.join("classical.m3u8")).unwrap();
|
|
wai::http_binding::hls::check_adjacent(&media, &classical).unwrap();
|
|
let cid = hex(&content_hash(&std::fs::read(object("O9b")).unwrap()));
|
|
// A WAI1 is listed with its n alone; a WAI2 with its h too (§5.4).
|
|
assert!(media.contains(&format!("#EXTINF:0.640,\nobjects/{cid}.wai#n=18004\n")), "{media}");
|
|
assert!(media.contains("wai#n=21034&h=1009\n"), "{media}");
|
|
let master = std::fs::read_to_string(d.join("wai.m3u8")).unwrap();
|
|
assert_eq!(master, "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=262925,AVERAGE-BANDWIDTH=239246,CODECS=\"wai2.video.av1|video.int_hyper\",X-WAI-FLOOR=\"wai2.video.av1\",RESOLUTION=320x180,FRAME-RATE=25.000\nwai-180.m3u8\n");
|
|
let run = wai(&["http", "mpd", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.join("wai.mpd").to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let mpd = std::fs::read_to_string(d.join("wai.mpd")).unwrap();
|
|
assert!(mpd.contains("<Representation id=\"wai-180\" bandwidth=\"262925\">") && mpd.contains("<S t=\"1280\" d=\"640\"/>"), "{mpd}");
|
|
// An object list that does not cover the classical timeline is refused: three
|
|
// objects for two classical segments.
|
|
std::fs::write(d.join("classical.m3u8"), classical.replacen("#EXTINF:0.640,\nc2.m4s\n", "", 1)).unwrap();
|
|
let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
|
|
assert!(!run.status.success());
|
|
}
|
|
|
|
/// `wai http mpd --classical-mpd` takes the classical MPD's period, timescale, `t` and
|
|
/// `d`, leaves a gap's `S` and `SegmentURL` out, and checks the result adjacent to the
|
|
/// classical MPD; a list that does not cover the classical timeline is refused.
|
|
#[test]
|
|
fn mpd_takes_the_classical_mpd_timeline_and_leaves_gaps_out() {
|
|
let d = o9_presentation("mpd-classical");
|
|
let classical = "<?xml version=\"1.0\"?>\n<MPD xmlns=\"urn:mpeg:dash:schema:mpd:2011\" type=\"static\">\n <Period id=\"main\" start=\"PT10S\">\n <AdaptationSet id=\"7\" mimeType=\"video/mp4\">\n <SegmentTemplate timescale=\"90000\" media=\"c$Number$.m4s\">\n <SegmentTimeline><S t=\"0\" d=\"57600\" r=\"2\"/></SegmentTimeline>\n </SegmentTemplate>\n <Representation id=\"v\" bandwidth=\"1\"/>\n </AdaptationSet>\n </Period>\n</MPD>\n";
|
|
std::fs::write(d.join("classical.mpd"), classical).unwrap();
|
|
let mut list: serde_json::Value = serde_json::from_slice(&std::fs::read(d.join("list.json")).unwrap()).unwrap();
|
|
list["variants"][0]["segments"][1]["gap"] = serde_json::json!(true);
|
|
std::fs::write(d.join("list.json"), serde_json::to_vec(&list).unwrap()).unwrap();
|
|
let args = |out: &str| vec!["http".to_string(), "mpd".into(), "--objects".into(), d.join("list.json").to_str().unwrap().into(), "--out".into(),
|
|
d.join(out).to_str().unwrap().into(), "--classical-mpd".into(), d.join("classical.mpd").to_str().unwrap().into()];
|
|
let run = wai(&args("wai.mpd").iter().map(String::as_str).collect::<Vec<_>>());
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let mpd = std::fs::read_to_string(d.join("wai.mpd")).unwrap();
|
|
assert!(mpd.contains("<Period id=\"main\" start=\"PT10S\">") && mpd.contains("<SegmentList timescale=\"90000\">"), "{mpd}");
|
|
assert!(mpd.contains("<S t=\"0\" d=\"57600\"/>") && mpd.contains("<S t=\"115200\" d=\"57600\"/>") && !mpd.contains("t=\"57600\""), "{mpd}");
|
|
assert_eq!(mpd.matches("<SegmentURL ").count(), 2, "{mpd}");
|
|
wai::http_binding::dash_read::check_dash_adjacent(&mpd, classical).unwrap();
|
|
std::fs::write(d.join("classical.mpd"), classical.replace("r=\"2\"", "r=\"1\"")).unwrap();
|
|
let run = wai(&args("short.mpd").iter().map(String::as_str).collect::<Vec<_>>());
|
|
assert!(!run.status.success());
|
|
assert!(String::from_utf8_lossy(&run.stderr).contains("3 segments, and the classical timeline has 2"), "{}", String::from_utf8_lossy(&run.stderr));
|
|
}
|
|
|
|
/// `wai http seal-playlist` writes `mr/<manifest_hash>.json` and
|
|
/// `mr/<receipt_hash>.json`; `check-playlist` reports it signed, refuses a changed
|
|
/// playlist, and holds a revision to the head it chains to, walking the revisions
|
|
/// between.
|
|
#[cfg(feature = "video_manifest")]
|
|
#[test]
|
|
fn seal_and_check_playlist() {
|
|
let d = o9_presentation("seal");
|
|
let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let pl = d.join("wai-180.m3u8");
|
|
let key = "07".repeat(32);
|
|
let run = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--signer-id", "did:example:cli", "--content-id", "o9"]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let bytes = std::fs::read(&pl).unwrap();
|
|
let name = format!("{}.json", blake3::hash(&bytes).to_hex());
|
|
let receipt = d.join("mr").join(&name);
|
|
assert!(String::from_utf8_lossy(&run.stdout).starts_with(receipt.to_str().unwrap()));
|
|
let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", receipt.to_str().unwrap(), "--classical", d.join("classical.m3u8").to_str().unwrap()]);
|
|
assert!(ok.status.success(), "{}", String::from_utf8_lossy(&ok.stdout));
|
|
assert!(String::from_utf8_lossy(&ok.stdout).starts_with("signed\t"));
|
|
let mut changed = bytes.clone();
|
|
let at = changed.len() - 20;
|
|
changed[at] ^= 1;
|
|
std::fs::write(d.join("changed.m3u8"), &changed).unwrap();
|
|
let bad = wai(&["http", "check-playlist", d.join("changed.m3u8").to_str().unwrap(), "--receipt", receipt.to_str().unwrap()]);
|
|
assert_eq!(bad.status.code(), Some(1));
|
|
assert_eq!(String::from_utf8_lossy(&bad.stdout), "refused\tplaylist_receipt_mismatch\n");
|
|
// With no grant, one of no ceilings, written null; under the publisher's grant, its
|
|
// ceilings; a ceiling past 2^53 - 1 is not sealed.
|
|
assert!(String::from_utf8_lossy(&std::fs::read(&receipt).unwrap()).contains("\"joule_ceiling_micro\":null"));
|
|
let grant = d.join("grant.json");
|
|
std::fs::write(&grant, format!("{{\"grant_hash\":\"{}\",\"capability\":\"video.manifest\",\"joule_ceiling_micro\":5000000,\"funds_ceiling\":null}}", "ab".repeat(32))).unwrap();
|
|
let granted = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--grant", grant.to_str().unwrap(), "--out", d.join("granted").to_str().unwrap()]);
|
|
assert!(granted.status.success(), "{}", String::from_utf8_lossy(&granted.stderr));
|
|
let file = String::from_utf8(std::fs::read(d.join("granted/mr").join(&name)).unwrap()).unwrap();
|
|
assert!(file.contains("\"joule_ceiling_micro\":5000000") && file.contains(&"ab".repeat(32)), "{file}");
|
|
let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", d.join("granted/mr").join(&name).to_str().unwrap()]);
|
|
assert!(ok.status.success(), "{}", String::from_utf8_lossy(&ok.stdout));
|
|
std::fs::write(&grant, format!("{{\"grant_hash\":\"{}\",\"capability\":\"video.manifest\",\"joule_ceiling_micro\":9007199254740992,\"funds_ceiling\":null}}", "ab".repeat(32))).unwrap();
|
|
let past = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--grant", grant.to_str().unwrap(), "--out", d.join("past").to_str().unwrap()]);
|
|
assert!(!past.status.success());
|
|
// Each receipt is also written under its receipt hash.
|
|
let hash_of = |out: &std::process::Output| String::from_utf8_lossy(&out.stdout).trim_end().rsplit(' ').next().unwrap().to_string();
|
|
let r1_hash = hash_of(&run);
|
|
assert_eq!(std::fs::read(d.join("mr").join(format!("{r1_hash}.json"))).unwrap(), std::fs::read(&receipt).unwrap());
|
|
// A second revision sealed with the first as its parent checks against it; the
|
|
// first fetched again is the head unchanged.
|
|
let run = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--parent", receipt.to_str().unwrap(), "--out", d.join("r2").to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let r2 = d.join("r2/mr").join(&name);
|
|
let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", r2.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
|
|
assert!(ok.status.success(), "{}", String::from_utf8_lossy(&ok.stdout));
|
|
assert!(String::from_utf8_lossy(&ok.stdout).contains("\tchained (0 parents walked)"));
|
|
let same = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", receipt.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
|
|
assert!(same.status.success(), "{}", String::from_utf8_lossy(&same.stdout));
|
|
// A third revision, chained to the second, checked against the first: the second is
|
|
// fetched by its link from the third's directory; without it the walk stops.
|
|
let r2_file = d.join("r2/mr").join(format!("{}.json", hash_of(&run)));
|
|
let run = wai(&["http", "seal-playlist", pl.to_str().unwrap(), "--key", &key, "--parent", r2_file.to_str().unwrap(), "--out", d.join("r2").to_str().unwrap()]);
|
|
assert!(run.status.success(), "{}", String::from_utf8_lossy(&run.stderr));
|
|
let ok = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", r2.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
|
|
assert!(String::from_utf8_lossy(&ok.stdout).contains("\tchained (1 parents walked)"), "{}", String::from_utf8_lossy(&ok.stdout));
|
|
std::fs::remove_file(&r2_file).unwrap();
|
|
let bad = wai(&["http", "check-playlist", pl.to_str().unwrap(), "--receipt", r2.to_str().unwrap(), "--parent", receipt.to_str().unwrap()]);
|
|
assert_eq!(String::from_utf8_lossy(&bad.stdout), "refused\tplaylist_receipt_parent_missing\n");
|
|
}
|
|
|
|
/// The origin serves a playlist to be revalidated on every use, and its receipt
|
|
/// immutable.
|
|
#[test]
|
|
fn origin_serves_playlists_revalidated() {
|
|
let d = o9_presentation("origin-playlists");
|
|
let run = wai(&["http", "hls", "--objects", d.join("list.json").to_str().unwrap(), "--out", d.to_str().unwrap()]);
|
|
assert!(run.status.success());
|
|
std::fs::write(d.join("wai.mpd"), "<MPD/>\n").unwrap();
|
|
let o = Origin::start(&d, &[]);
|
|
let (h, body) = get(o.port, "/wai-180.m3u8", &[]);
|
|
assert!(h.contains("Content-Type: application/vnd.apple.mpegurl\r\n") && h.contains("Cache-Control: no-cache, no-transform\r\n"), "{h}");
|
|
assert_eq!(body, std::fs::read(d.join("wai-180.m3u8")).unwrap());
|
|
let (h, _) = get(o.port, "/wai.mpd", &[]);
|
|
assert!(h.contains("Content-Type: application/dash+xml\r\n") && h.contains("Cache-Control: no-cache, no-transform\r\n"), "{h}");
|
|
}
|
|
|
|
/// The replayed live edge (§11.4): the playlist as its current revision; a blocking
|
|
/// reload for a part answered by a revision that lists it; a hinted part's alias held
|
|
/// until a revision lists it, and answered with the object whose content hash is the cid
|
|
/// that revision lists; an alias served with another part's bytes, which that cid
|
|
/// refuses; and a hint no revision ever fills, not found once the edge ends. Nothing here
|
|
/// reads a clock: each answer is checked by what it holds.
|
|
#[test]
|
|
fn origin_replays_a_live_edge() {
|
|
let root = corpus();
|
|
let revs: Vec<String> = (0..4).map(|k| format!("live/wai-r{k}.m3u8")).collect();
|
|
let texts: Vec<Vec<u8>> = revs.iter().map(|r| std::fs::read(root.join(r)).unwrap()).collect();
|
|
let edge = format!("live/wai-720.m3u8={}", revs.join(","));
|
|
let o = Origin::start(&root, &["--live-edge", &edge, "--live-step-ms", "400", "--live-swap", "720/1201.7.wai=720/1201.8.wai"]);
|
|
let (h, body) = get(o.port, "/live/wai-720.m3u8", &[]);
|
|
assert!(h.starts_with("HTTP/1.1 200") && h.contains("Cache-Control: no-cache, no-transform\r\n"), "{h}");
|
|
assert!(texts.contains(&body), "the playlist is one of its revisions");
|
|
// A blocking reload more than two segments past the edge, or a part with no segment,
|
|
// is refused at once (RFC 8216bis §6.2.5.2), not held.
|
|
// (A held reload is answered 200 with a revision; 400 is the refusal.)
|
|
for q in ["_HLS_msn=1300", "_HLS_msn=1300&_HLS_part=0", "_HLS_part=1", "_HLS_msn=x", "_HLS_msn=-1"] {
|
|
let (h, _) = get(o.port, &format!("/live/wai-720.m3u8?{q}"), &[]);
|
|
assert!(h.starts_with("HTTP/1.1 400"), "{q}: {h}");
|
|
}
|
|
// Part 2 of segment 1201 is first listed by revision 1.
|
|
let (_, body) = get(o.port, "/live/wai-720.m3u8?_HLS_msn=1201&_HLS_part=2", &[]);
|
|
let p = wai::http_binding::hls::parse_media_playlist(&String::from_utf8(body.clone()).unwrap()).unwrap();
|
|
assert!(wai::http_binding::hls::listed_cid(&p, "720/1201.2.wai").is_some(), "the blocking reload lists the part");
|
|
assert!(texts[1..].contains(&body));
|
|
// Revision 0 hints 1200.5; revision 1 lists it with its cid.
|
|
let r1 = wai::http_binding::hls::parse_media_playlist(&String::from_utf8(texts[1].clone()).unwrap()).unwrap();
|
|
let (h, part) = get(o.port, "/live/720/1200.5.wai", &[]);
|
|
assert!(h.starts_with("HTTP/1.1 200"), "{h}");
|
|
assert_eq!(Some(content_hash(&part)), wai::http_binding::hls::listed_cid(&r1, "720/1200.5.wai"));
|
|
// Served with another part's bytes: the cid the playlist lists refuses them.
|
|
let r2 = wai::http_binding::hls::parse_media_playlist(&String::from_utf8(texts[2].clone()).unwrap()).unwrap();
|
|
let (_, swapped) = get(o.port, "/live/720/1201.7.wai", &[]);
|
|
assert_ne!(Some(content_hash(&swapped)), wai::http_binding::hls::listed_cid(&r2, "720/1201.7.wai"));
|
|
assert_eq!(Some(content_hash(&swapped)), wai::http_binding::hls::listed_cid(&r2, "720/1201.8.wai"));
|
|
// Revision 2 hints 1202.0, which no revision fills: not found when the edge ends,
|
|
// and the playlist is then its last revision, which ends.
|
|
let (h, _) = get(o.port, "/live/720/1202.0.wai", &[]);
|
|
assert!(h.starts_with("HTTP/1.1 404"), "{h}");
|
|
let (_, body) = get(o.port, "/live/wai-720.m3u8", &[]);
|
|
assert_eq!(body, texts[3]);
|
|
// A byte range of a part, under its alias.
|
|
let (h, b) = get(o.port, "/live/720/1200.5.wai", &[("Range", "bytes=0-7")]);
|
|
assert!(h.starts_with("HTTP/1.1 206") && b == part[..8], "{h}");
|
|
}
|