81e5788ce9
run.sh takes the directory of its labelled caches from WAI_CACHE_DIR, beside the default /opt/wai-cache, and a directory that was asked for and holds no labels file fails the run. wai http origin sends Date on every response. Without one, a cache has nothing to count an object's age from, and one such cache held every object as already stale and forwarded each request, so its store was never reached. Co-Authored-By: Claude Opus 5.5 <[email protected]>
902 lines
53 KiB
Rust
902 lines
53 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}");
|
||
assert!(h.lines().any(|l| l.starts_with("Date: ") && l.ends_with(" GMT")), "no Date in {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", "60000"]);
|
||
let request = b"GET /x.json HTTP/1.1\r\nHost: a\r\n\r\n";
|
||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
|
||
// The origin counts each connection on a thread of its own, so a third can overtake the idle
|
||
// two and one of them is refused instead. An attempt that went that way starts again.
|
||
let (idle, r) = loop {
|
||
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(100));
|
||
let r = raw(origin.port, request);
|
||
if r.starts_with("HTTP/1.1 503") || std::time::Instant::now() > deadline {
|
||
break (idle, r);
|
||
}
|
||
};
|
||
assert!(r.starts_with("HTTP/1.1 503"), "{r:?}");
|
||
// A refusal ends cleanly: closing with the request unread would end the connection in a reset,
|
||
// which a client reading to the close takes for a failure, after a valid answer. So the read
|
||
// itself must succeed here, and no attempt is retried.
|
||
for i in 0..200 {
|
||
let mut s = TcpStream::connect(("127.0.0.1", origin.port)).unwrap();
|
||
s.set_read_timeout(Some(std::time::Duration::from_secs(10))).unwrap();
|
||
let mut out = Vec::new();
|
||
let read = s.write_all(request).and_then(|()| s.read_to_end(&mut out));
|
||
let r = String::from_utf8_lossy(&out);
|
||
assert!(read.is_ok() && r.starts_with("HTTP/1.1 503") && r.contains("Retry-After: 1"), "refusal {i}: {read:?} {r:?}");
|
||
}
|
||
drop(idle);
|
||
let r = loop {
|
||
let r = raw(origin.port, request);
|
||
if r.starts_with("HTTP/1.1 200") || std::time::Instant::now() > deadline {
|
||
break r;
|
||
}
|
||
std::thread::sleep(std::time::Duration::from_millis(50));
|
||
};
|
||
assert!(r.starts_with("HTTP/1.1 200"), "{r:?}");
|
||
}
|
||
|
||
/// 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}");
|
||
}
|
||
|
||
/// `wai http receipts` (§8.1–§8.3) against the reference origin serving the corpus:
|
||
/// O3 fetched by its head proof, its group accepted on that cid, and its claims (the
|
||
/// ancestor's soft binding and a receipt claim with its parent fetched by link), each
|
||
/// signer confirmed under the pinned key log; O9c's group walked back two parents; a
|
||
/// layer not fetched inclusion-verified; and a walk past its bound refused.
|
||
#[cfg(feature = "key_discovery")]
|
||
#[test]
|
||
fn receipts_checks_an_index_over_http() {
|
||
let o = Origin::start(&corpus(), &[]);
|
||
let url = |id: &str| {
|
||
let p = object(id);
|
||
format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap())
|
||
};
|
||
let log = corpus().join(".well-known/wai-keys");
|
||
let run = wai(&["http", "receipts", "--url", &url("O3"), "--advertise", "wai.video.av1", "--pin-key-log", log.to_str().unwrap(),
|
||
"--key-origin", "https://origin.example"]);
|
||
let out = String::from_utf8_lossy(&run.stdout);
|
||
let v: serde_json::Value = serde_json::from_str(&out).unwrap_or_else(|e| panic!("{e}: {out}{}", String::from_utf8_lossy(&run.stderr)));
|
||
assert_eq!((v["fetch"].as_str(), v["held"].as_str()), (Some("head-proof"), Some("head-proof")), "{v}");
|
||
let d = &v["delivery"][0];
|
||
assert_eq!((d["evidence"].as_str(), d["established"].as_str()), (Some("delivered-verified"), Some("head-proof")), "{v}");
|
||
assert!(d["confirm"].as_str().unwrap().starts_with("confirmed origin=https://origin.example key_id=d1"), "{v}");
|
||
// O3's claims: a soft binding of O7, with no recovery run here, is about another
|
||
// object; a receipt claim with its parent; an energy claim checked at step 6 against
|
||
// the object's figure, and one that disagrees with it.
|
||
assert!(v["claims"][0]["refused"].as_str().unwrap().starts_with("receipt_not_about_object"), "{v}");
|
||
assert_eq!(v["claims"][1]["attests"], "receipt");
|
||
assert!(v["claims"][1]["parent"].is_string(), "{v}");
|
||
assert_eq!(v["claims"][2]["attests"], "object");
|
||
assert!(v["claims"][3]["refused"].as_str().unwrap().starts_with("energy_disagrees_with_delivery"), "{v}");
|
||
assert!(!run.status.success(), "a refused claim makes the exit status 1");
|
||
let run = wai(&["http", "receipts", "--url", &url("O9c"), "--advertise", "wai.video.av1"]);
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
|
||
assert!(run.status.success(), "{v}");
|
||
assert_eq!((v["delivery"][0]["walked"].as_u64(), v["delivery"][0]["admission"].as_str()), (Some(2), Some("accept")), "{v}");
|
||
let run = wai(&["http", "receipts", "--url", &url("O10L2"), "--held", "none"]);
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
|
||
assert_eq!((v["held"].as_str(), v["delivery"][0]["evidence"].as_str()), (Some("nothing"), Some("inclusion-verified")), "{v}");
|
||
let run = wai(&["http", "receipts", "--url", &url("O9c"), "--held", "none", "--max-walk", "1"]);
|
||
assert!(!run.status.success());
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
|
||
assert!(v["delivery"][0]["refused"].as_str().unwrap().starts_with("parent_not_held"), "{v}");
|
||
// The origin serves the index for revalidation and the status for a quarter of its
|
||
// freshness at most.
|
||
let cid = hex(&content_hash(&std::fs::read(object("O3")).unwrap()));
|
||
let (h, _) = get(o.port, &format!("/objects/{cid}.wai.receipts"), &[]);
|
||
assert!(h.contains("Cache-Control: no-cache, no-transform"), "{h}");
|
||
let (h, _) = get(o.port, "/.well-known/wai-key-status", &[]);
|
||
assert!(h.contains("Cache-Control: max-age=900, no-transform"), "{h}");
|
||
}
|
||
|
||
/// A test server for `wai http receipts`: each connection's request head is kept, and it
|
||
/// answers `mode`: `log` (the corpus's key log, 200), `redirect` (302 to another host),
|
||
/// `hang` (a status line and headers, then nothing), `stream` (a body that does not end).
|
||
#[cfg(feature = "key_discovery")]
|
||
fn test_server(mode: &'static str) -> (u16, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
|
||
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||
let port = l.local_addr().unwrap().port();
|
||
let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
|
||
let kept = seen.clone();
|
||
std::thread::spawn(move || {
|
||
for s in l.incoming().flatten() {
|
||
let kept = kept.clone();
|
||
std::thread::spawn(move || {
|
||
let mut s = s;
|
||
if mode == "plain" {
|
||
// What a client sends first, recorded as it arrives, answered in plain
|
||
// HTTP with the key log.
|
||
let _ = s.set_read_timeout(Some(std::time::Duration::from_secs(10)));
|
||
let mut first = [0u8; 512];
|
||
let n = s.read(&mut first).unwrap_or(0);
|
||
kept.lock().unwrap().push(String::from_utf8_lossy(&first[..n]).into_owned());
|
||
let body = std::fs::read(corpus().join(".well-known/wai-keys")).unwrap();
|
||
let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes());
|
||
let _ = s.write_all(&body);
|
||
return;
|
||
}
|
||
let mut head = Vec::new();
|
||
let mut b = [0u8; 1];
|
||
while !head.ends_with(b"\r\n\r\n") && s.read(&mut b).map_or(false, |n| n == 1) {
|
||
head.push(b[0]);
|
||
}
|
||
kept.lock().unwrap().push(String::from_utf8_lossy(&head).into_owned());
|
||
match mode {
|
||
"log" => {
|
||
let body = std::fs::read(corpus().join(".well-known/wai-keys")).unwrap();
|
||
let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes());
|
||
let _ = s.write_all(&body);
|
||
}
|
||
"biglog" => {
|
||
// The log padded past its cap (1 MiB), framed by Content-Length.
|
||
let mut body = std::fs::read(corpus().join(".well-known/wai-keys")).unwrap();
|
||
body.resize(1_048_621, b' ');
|
||
let _ = s.write_all(format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n", body.len()).as_bytes());
|
||
let _ = s.write_all(&body);
|
||
}
|
||
"redirect" => {
|
||
let _ = s.write_all(b"HTTP/1.1 302 Found\r\nLocation: https://mirror.example/.well-known/wai-keys\r\nContent-Length: 0\r\n\r\n");
|
||
}
|
||
"hang" => {
|
||
let _ = s.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n");
|
||
std::thread::sleep(std::time::Duration::from_secs(30));
|
||
}
|
||
_ => {
|
||
// A body that does not end before any deadline the test gives: 64 KiB
|
||
// every 5 ms, for longer than a minute, until the client hangs up.
|
||
let _ = s.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\r\n");
|
||
let chunk = vec![b' '; 64 << 10];
|
||
for _ in 0..(1 << 14) {
|
||
if s.write_all(&chunk).is_err() {
|
||
break;
|
||
}
|
||
std::thread::sleep(std::time::Duration::from_millis(5));
|
||
}
|
||
}
|
||
}
|
||
});
|
||
}
|
||
});
|
||
(port, seen)
|
||
}
|
||
|
||
/// §8.4's test aid: `--test-plaintext-connect` fetches the key log over plain HTTP from a
|
||
/// stated address, as the TLS path asks for it (revalidating, no redirect followed). What
|
||
/// it retrieves confirms this run's signers and is labelled `test-plaintext`, never
|
||
/// `fetched`, as anchoring nothing a verifier may rely on, and the command fails; a
|
||
/// redirect anchors nothing. `--connect` is gone.
|
||
#[cfg(feature = "key_discovery")]
|
||
#[test]
|
||
fn receipts_labels_a_plaintext_test_anchor_as_such() {
|
||
let o = Origin::start(&corpus(), &[]);
|
||
let p = object("O3");
|
||
let url = format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap());
|
||
let (kp, seen) = test_server("log");
|
||
let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
|
||
"--test-plaintext-connect", &format!("127.0.0.1:{kp}")]);
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&run.stderr)));
|
||
assert_eq!(v["anchor"]["path"], "test-plaintext", "{v}");
|
||
assert_eq!(v["anchor"]["relied_on"], false, "{v}");
|
||
assert!(v["anchor"]["note"].as_str().unwrap().contains("anchors nothing a verifier may rely on"), "{v}");
|
||
assert!(v["delivery"][0]["confirm"].as_str().unwrap().starts_with("confirmed origin=https://origin.example key_id=d1"), "{v}");
|
||
// Every confirmation names the test anchor, and no anchor anywhere is `fetched`.
|
||
let confirms: Vec<&str> = ["delivery", "claims"].iter().flat_map(|k| v[*k].as_array().unwrap().iter())
|
||
.filter_map(|x| x["confirm"].as_str()).collect();
|
||
assert!(confirms.iter().filter(|c| c.starts_with("confirmed")).count() >= 2, "{v}");
|
||
for c in confirms.iter().filter(|c| c.starts_with("confirmed")) {
|
||
assert!(c.contains(" anchor=test-plaintext"), "{c}");
|
||
}
|
||
let out = String::from_utf8_lossy(&run.stdout);
|
||
assert!(!out.contains("anchor=fetched") && !out.contains("\"fetched\""), "{v}");
|
||
let req = seen.lock().unwrap().join("");
|
||
assert!(req.starts_with("GET /.well-known/wai-keys HTTP/1.1\r\n"), "{req}");
|
||
assert!(req.contains("\r\nHost: origin.example\r\n") && req.contains("\r\nCache-Control: no-cache\r\n"), "{req}");
|
||
// A redirect is not followed, and the answer anchors nothing.
|
||
let (rp, seen) = test_server("redirect");
|
||
let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
|
||
"--test-plaintext-connect", &format!("127.0.0.1:{rp}")]);
|
||
assert!(!run.status.success());
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
|
||
assert_eq!(v["anchor"]["refused"], "key_log_not_from_origin", "{v}");
|
||
assert!(v["delivery"][0].get("confirm").is_none(), "{v}");
|
||
assert_eq!(seen.lock().unwrap().len(), 1, "the redirect was followed");
|
||
// A log past its cap is refused, read no further than one byte past it.
|
||
let (bp, _) = test_server("biglog");
|
||
let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
|
||
"--test-plaintext-connect", &format!("127.0.0.1:{bp}")]);
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap();
|
||
assert_eq!(v["anchor"]["refused"], "key_log_too_large", "{v}");
|
||
// The flag that let any server stand in for the origin is gone.
|
||
let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example", "--connect", &format!("127.0.0.1:{kp}")]);
|
||
assert!(!run.status.success() && String::from_utf8_lossy(&run.stderr).contains("--connect is gone"));
|
||
}
|
||
|
||
/// The test aid's anchor never lets the command succeed: on an object whose index passes
|
||
/// whole (O7), the same check exits 0 with the log pinned and 1 with it fetched over
|
||
/// plaintext from a stated address.
|
||
#[cfg(feature = "key_discovery")]
|
||
#[test]
|
||
fn receipts_never_succeeds_on_a_test_anchor() {
|
||
let o = Origin::start(&corpus(), &[]);
|
||
let p = object("O7");
|
||
let url = format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap());
|
||
let log = corpus().join(".well-known/wai-keys");
|
||
let pinned = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
|
||
"--pin-key-log", log.to_str().unwrap()]);
|
||
let v: serde_json::Value = serde_json::from_slice(&pinned.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&pinned.stderr)));
|
||
assert!(pinned.status.success(), "a pinned anchor on a whole index succeeds: {v}");
|
||
assert_eq!(v["anchor"]["path"], "pinned", "{v}");
|
||
let (kp, _) = test_server("log");
|
||
let test = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", "https://origin.example",
|
||
"--test-plaintext-connect", &format!("127.0.0.1:{kp}")]);
|
||
let w: serde_json::Value = serde_json::from_slice(&test.stdout).unwrap();
|
||
assert_eq!(w["anchor"]["path"], "test-plaintext", "{w}");
|
||
let strip = |x: &serde_json::Value| x.as_array().map(|d| json_without_confirm(d)).unwrap_or_default();
|
||
assert_eq!((strip(&w["delivery"]), &w["index"]), (strip(&v["delivery"]), &v["index"]), "the same check but for its anchor");
|
||
assert!(!test.status.success(), "a plaintext anchor never makes the command succeed: {w}");
|
||
}
|
||
|
||
/// Delivery verdicts with their confirmations dropped: the part an anchor does not
|
||
/// decide.
|
||
fn json_without_confirm(d: &[serde_json::Value]) -> Vec<serde_json::Value> {
|
||
d.iter().map(|x| { let mut x = x.clone(); x.as_object_mut().map(|m| m.remove("confirm")); x }).collect()
|
||
}
|
||
|
||
/// §8.4: without the test aid, the key log is retrieved over TLS from the origin itself,
|
||
/// so a server that speaks plain HTTP at the origin's address is never asked for it in
|
||
/// plaintext and anchors nothing; a build without TLS asks nothing at all.
|
||
#[cfg(feature = "key_discovery")]
|
||
#[test]
|
||
fn receipts_never_anchors_over_plaintext() {
|
||
let o = Origin::start(&corpus(), &[]);
|
||
let p = object("O3");
|
||
let url = format!("http://127.0.0.1:{}/{}", o.port, p.strip_prefix(corpus()).unwrap().to_str().unwrap());
|
||
let (kp, seen) = test_server("plain");
|
||
let run = wai(&["http", "receipts", "--url", &url, "--held", "none", "--key-origin", &format!("https://127.0.0.1:{kp}")]);
|
||
assert!(!run.status.success());
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&run.stderr)));
|
||
assert_eq!(v["anchor"]["refused"], "key_log_not_from_origin", "{v}");
|
||
assert!(v["anchor"].get("path").is_none(), "{v}");
|
||
assert!(v["delivery"][0].get("confirm").is_none(), "{v}");
|
||
let sent = seen.lock().unwrap().join("");
|
||
assert!(!sent.contains("GET "), "a plaintext request reached the server: {sent:?}");
|
||
// With TLS the client opens with a handshake, unless the host has no trust store to
|
||
// validate against, when it refuses before it connects.
|
||
#[cfg(feature = "cli_tls")]
|
||
assert!(sent.starts_with('\u{16}') || (sent.is_empty() && v["anchor"]["why"].as_str().unwrap().contains("TLS")),
|
||
"the client opened with a TLS handshake: {sent:?} {v}");
|
||
#[cfg(not(feature = "cli_tls"))]
|
||
{
|
||
assert!(sent.is_empty(), "a build without TLS connects to nothing: {sent:?}");
|
||
assert!(v["anchor"]["why"].as_str().unwrap().contains("TLS"), "{v}");
|
||
}
|
||
}
|
||
|
||
/// §8.2 over HTTP: a server that stops after its headers, or never ends its body, costs
|
||
/// `wai http receipts` no more than its deadline and the index's cap. The hanging server
|
||
/// holds its connection open for 30 s and the stream runs past a minute, so a client
|
||
/// without the deadline, or without the cap, reports something else.
|
||
#[cfg(feature = "key_discovery")]
|
||
#[test]
|
||
fn receipts_is_bounded_by_its_deadline_and_caps() {
|
||
let cid = "31abc1da2cdf927191f3f447f68508ead30d078690cb7bea2435ae3d1df5f4f3";
|
||
let (hp, _) = test_server("hang");
|
||
let run = wai(&["http", "receipts", "--held", "none", "--deadline-ms", "1500", "--url", &format!("http://127.0.0.1:{hp}/o/{cid}.wai")]);
|
||
assert!(!run.status.success());
|
||
assert!(String::from_utf8_lossy(&run.stderr).contains("deadline"), "{}", String::from_utf8_lossy(&run.stderr));
|
||
let (sp, _) = test_server("stream");
|
||
let run = wai(&["http", "receipts", "--held", "none", "--deadline-ms", "30000", "--url", &format!("http://127.0.0.1:{sp}/o/{cid}.wai")]);
|
||
assert!(!run.status.success());
|
||
let v: serde_json::Value = serde_json::from_slice(&run.stdout).unwrap_or_else(|e| panic!("{e}: {}", String::from_utf8_lossy(&run.stderr)));
|
||
assert!(v["index"].as_str().unwrap().starts_with("receipt_index_too_large"), "{v}");
|
||
}
|