open-standards/wai/wai-audit/src/tests.rs

//! The collector against the stream-audit corpus (`live-conformance/audit/`): every session
//! bundle posted gets the corpus's outcome, the stored sessions report exactly as the corpus's
//! served reports, byte for byte and signature, and the limits, CORS and CMCD intake hold.

use crate::collector::{Collector, Config};
use crate::http::{App, router};
use axum::body::Body;
use axum::extract::ConnectInfo;
use axum::http::{Request, StatusCode};
use ed25519_dalek::SigningKey;
use http_body_util::BodyExt;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use tower::ServiceExt;

fn corpus() -> Value {
    let p = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../live-conformance/audit/vectors.json");
    serde_json::from_slice(&std::fs::read(&p).unwrap_or_else(|e| panic!("{}: {e}", p.display()))).expect("JSON")
}

/// The corpus's owner key: SHA-256 of its label and index 0 (its README).
fn owner_key() -> SigningKey {
    let mut h = Sha256::new();
    h.update(b"wai:stream-audit/corpus/key\x00");
    h.update([0u8]);
    SigningKey::from_bytes(&h.finalize().into())
}

fn data_dir(name: &str) -> PathBuf {
    let d = std::env::temp_dir().join(format!("wai-audit-test-{}-{name}", std::process::id()));
    let _ = std::fs::remove_dir_all(&d);
    d
}

fn app(name: &str, per_minute: u32, max_bundle: usize) -> (axum::Router, PathBuf) {
    let doc = corpus();
    let channels = doc["channels"].as_array().unwrap().iter().map(|c| c.as_str().unwrap().to_string()).collect();
    let data = data_dir(name);
    let config = Config { channels, data: data.clone(), allow_origins: vec!["https://player.example".into()], max_bundle, max_batch: 64 << 10, per_minute };
    let app = Arc::new(App { collector: Arc::new(Collector::new(config, owner_key())), trusted: vec![] });
    (router(app), data)
}

async fn call(r: &axum::Router, method: &str, uri: &str, body: Vec<u8>, origin: Option<&str>) -> (StatusCode, axum::http::HeaderMap, Value) {
    let mut b = Request::builder().method(method).uri(uri);
    if let Some(o) = origin {
        b = b.header("origin", o);
    }
    let mut req = b.body(Body::from(body)).unwrap();
    req.extensions_mut().insert(ConnectInfo(SocketAddr::from(([192, 0, 2, 7], 40_000))));
    let res = r.clone().oneshot(req).await.unwrap();
    let (parts, body) = res.into_parts();
    let bytes = body.collect().await.unwrap().to_bytes();
    let v = if bytes.is_empty() { Value::Null } else { serde_json::from_slice(&bytes).unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&bytes).into())) };
    (parts.status, parts.headers, v)
}

fn bundle_bytes(v: &Value) -> Vec<u8> {
    match &v["bundle_text"] {
        Value::String(s) => s.clone().into_bytes(),
        _ => serde_json::to_vec(&v["bundle"]).unwrap(),
    }
}

#[tokio::test]
async fn every_corpus_session_gets_its_outcome_and_the_store_reports_as_the_corpus() {
    let doc = corpus();
    let (r, data) = app("corpus", 10_000, 4 << 20);
    for v in doc["sessions"].as_array().unwrap() {
        let (status, _, body) = call(&r, "POST", "/v1/sessions", bundle_bytes(v), None).await;
        let name = v["name"].as_str().unwrap();
        if name == "nothing_presented_again" {
            assert_eq!(status, StatusCode::CONFLICT, "{name}");
            assert_eq!(body["reason"], "audit_session_duplicate", "{name}");
        } else if v["outcome"]["outcome"] == "accept" {
            assert_eq!(status, StatusCode::CREATED, "{name}: {body}");
            assert_eq!(body["outcome"], v["outcome"], "{name}");
        } else {
            assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{name}: {body}");
            assert_eq!(body, v["outcome"], "{name}");
        }
    }
    // The same record again is stored once.
    let first = &doc["sessions"][0];
    let (status, _, body) = call(&r, "POST", "/v1/sessions", bundle_bytes(first), None).await;
    assert_eq!((status, body["stored"].as_str()), (StatusCode::OK, Some("again")));
    // The stored sessions' reports are the corpus's, byte for byte and signature.
    for name in ["news_east", "film_library", "fifty"] {
        let v = doc["reports"].as_array().unwrap().iter().find(|x| x["name"] == name).unwrap();
        let rep = &v["served"]["report"];
        let uri = format!("/v1/channels/{}/report?from={}&to={}", v["channel"].as_str().unwrap(), rep["from"], rep["to"]);
        let (status, _, body) = call(&r, "GET", &uri, vec![], None).await;
        assert_eq!(status, StatusCode::OK, "{name}: {body}");
        assert_eq!(body, v["served"], "{name}: the report served is not the corpus's");
    }
    let (status, _, _) = call(&r, "GET", "/v1/channels/other-channel/report?from=0&to=1", vec![], None).await;
    assert_eq!(status, StatusCode::NOT_FOUND);
    let (status, _, _) = call(&r, "GET", "/v1/channels/news-east/report?from=5&to=5", vec![], None).await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
    let (status, _, body) = call(&r, "GET", "/v1", vec![], None).await;
    assert_eq!((status, body["owner"].as_str()), (StatusCode::OK, doc["owner"].as_str()));
    let _ = std::fs::remove_dir_all(data);
}

#[tokio::test]
async fn cors_limits_and_cmcd() {
    let doc = corpus();
    let (r, data) = app("limits", 2, 2048);
    // The preflight answers the configured page origin, and no other.
    let (status, h, _) = call(&r, "OPTIONS", "/v1/sessions", vec![], Some("https://player.example")).await;
    assert_eq!(status, StatusCode::NO_CONTENT);
    assert_eq!(h.get("access-control-allow-origin").unwrap(), "https://player.example");
    assert_eq!(h.get("access-control-allow-methods").unwrap(), "POST");
    let (_, h, _) = call(&r, "OPTIONS", "/v1/sessions", vec![], Some("https://elsewhere.example")).await;
    assert!(h.get("access-control-allow-origin").is_none());
    // A bundle over the size limit is refused before it is read.
    let big = bundle_bytes(&doc["sessions"][0]);
    assert!(big.len() > 2048);
    let (status, _, _) = call(&r, "POST", "/v1/sessions", big, Some("https://player.example")).await;
    assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE);
    // An Event-mode batch is taken and stored; a line that is not one is refused.
    let batch = doc["sessions"][0]["cmcd"].as_str().unwrap().as_bytes().to_vec();
    let (status, h, _) = call(&r, "POST", "/v1/cmcd", batch.clone(), Some("https://player.example")).await;
    assert_eq!(status, StatusCode::NO_CONTENT);
    assert_eq!(h.get("access-control-allow-origin").unwrap(), "https://player.example");
    let stored: String = std::fs::read_dir(data.join("cmcd")).unwrap().flatten().map(|e| std::fs::read_to_string(e.path()).unwrap()).collect();
    assert_eq!(stored.trim_end(), std::str::from_utf8(&batch).unwrap());
    let (status, _, body) = call(&r, "POST", "/v1/cmcd", b"v=2,sid=\"x\"".to_vec(), None).await;
    assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY, "{body}");
    // Two requests a minute: the two batches. The oversized bundle is refused before its
    // handler runs and counts nothing; the third batch waits.
    let (status, _, body) = call(&r, "POST", "/v1/cmcd", batch, None).await;
    assert_eq!(status, StatusCode::TOO_MANY_REQUESTS, "{body}");
    assert_eq!(json!(body["error"].is_string()), json!(true));
    let _ = std::fs::remove_dir_all(data);
}

#[test]
fn clients_are_counted_by_address_or_ipv6_64() {
    use axum::http::HeaderMap;
    let peer = |s: &str| s.parse::<SocketAddr>().unwrap();
    let none = HeaderMap::new();
    assert_eq!(crate::http::client(&[], peer("[2001:db8:1:2:3:4:5:6]:1"), &none), "2001:db8:1:2::/64");
    assert_eq!(crate::http::client(&[], peer("[::ffff:192.0.2.1]:1"), &none), "192.0.2.1");
    let mut fwd = HeaderMap::new();
    fwd.insert("x-forwarded-for", "198.51.100.9, 10.0.0.1".parse().unwrap());
    let proxy = "127.0.0.1".parse().unwrap();
    assert_eq!(crate::http::client(&[proxy], peer("127.0.0.1:1"), &fwd), "198.51.100.9");
    assert_eq!(crate::http::client(&[], peer("127.0.0.1:1"), &fwd), "127.0.0.1");
}