open-standards/joule-code/joule-code-rs/crates/joule-moq-ingest/tests/auth_and_ice.rs

//! Refusals at the WHIP and WHEP resources, by RFC 9725, `draft-ietf-wish-whep-05` and
//! RFC 6750; trickle ICE and an ICE restart over PATCH (RFC 9725 §4.3), with the media
//! carried across the restart.

mod common;

use common::*;
use ed25519_dalek::SigningKey;
use joule_moq_ingest::frame::Codec;
use joule_moq_ingest::room::build;
use joule_moq_ingest::sdp;
use std::time::Duration;
use str0m::media::Direction;
use str0m::IceConnectionState;
use tokio::time::timeout;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn tokens_without_the_power_are_refused() {
    timeout(Duration::from_secs(120), refusals()).await.expect("the test finishes in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn trickle_ice_and_an_ice_restart_over_patch() {
    timeout(Duration::from_secs(120), ice()).await.expect("the test finishes in time");
}

async fn post_whip(w: &World, auth: Option<&str>, ct: &str, body: &[u8]) -> Resp {
    let a = auth.map(bearer);
    let mut h = vec![("content-type", ct)];
    if let Some(a) = &a {
        h.push(("authorization", a.as_str()));
    }
    http("POST", &format!("{}/whip", w.base()), &h, body).await
}

async fn refusals() {
    let w = World::start(|_| {}).await;
    let o = offer(&[Codec::Avc, Codec::Opus], Direction::SendOnly, true, true).await;
    let ob = o.offer.as_bytes();

    // No token: 401, the challenge without an error (RFC 6750 §3.1).
    let r = post_whip(&w, None, "application/sdp", ob).await;
    assert_eq!(r.status, 401);
    assert_eq!(r.header("www-authenticate"), Some("Bearer realm=\"wai-ingest\""));

    // A token that is not a grant: 401 invalid_token.
    let r = post_whip(&w, Some("bm90LWEtZ3JhbnQ"), "application/sdp", ob).await;
    assert_eq!(r.status, 401);
    assert!(r.header("www-authenticate").unwrap().contains("error=\"invalid_token\""));

    // A viewer's grant, its key proved, has no publish power: 403 insufficient_scope.
    let whip = format!("{}/whip", w.base());
    let viewer = SigningKey::from_bytes(&[0x61; 32]);
    let va = w.auth(&viewer, 1);
    let r = va.send("POST", &whip, &[("content-type", "application/sdp")], ob).await;
    assert_eq!(r.status, 403, "{}", r.text());
    assert!(r.header("www-authenticate").unwrap().contains("error=\"insufficient_scope\""));
    assert!(r.text().contains("publish_not_permitted"));
    assert_eq!(r.header("content-type"), Some("application/problem+json"));

    // A grant past its window, and one of a room the gateway does not hold.
    let s = now_us() / 1_000_000;
    let old = build::grant(&w.descriptor, &w.authority, &w.publisher.verifying_key().to_bytes(), 0, s - 7200, s - 3600);
    let r = post_whip(&w, Some(&World::token(&old)), "application/sdp", ob).await;
    assert_eq!(r.status, 401);
    assert!(r.text().contains("outside_grant_window"), "{}", r.text());
    let other_authority = SigningKey::from_bytes(&[0x77; 32]);
    let other = build::room(&other_authority, now_us());
    let og = build::grant(&other, &other_authority, &w.publisher.verifying_key().to_bytes(), 0, s - 60, s + 60);
    let r = post_whip(&w, Some(&World::token(&og)), "application/sdp", ob).await;
    assert_eq!(r.status, 401);
    assert!(r.text().contains("room_unknown"), "{}", r.text());

    // A good grant, its key proved, but not an SDP body: 415.
    let pa = w.auth(&w.publisher, 0);
    let r = pa.send("POST", &whip, &[("content-type", "text/plain")], ob).await;
    assert_eq!(r.status, 415);

    // An offer with a data channel only: 422.
    let r = pa.send("POST", &whip, &[("content-type", "application/sdp")], b"v=0\r\no=- 1 1 IN IP4 127.0.0.1\r\ns=-\r\nt=0 0\r\nm=application 9 UDP/DTLS/SCTP webrtc-datachannel\r\na=mid:0\r\n").await;
    assert_eq!(r.status, 422);

    // WHEP: an observer's grant may not subscribe; the viewer's may not go above its
    // max_layer.
    let observer = SigningKey::from_bytes(&[0x62; 32]);
    let oa = w.auth(&observer, 2);
    let po = offer(&[Codec::Avc, Codec::Opus], Direction::RecvOnly, true, true).await;
    let r = oa.send("POST", &w.whep_url(), &[("content-type", "application/sdp")], po.offer.as_bytes()).await;
    assert_eq!(r.status, 403);
    assert!(r.text().contains("subscribe_not_permitted"));
    let r = va.send("POST", &format!("{}?layer=1", w.whep_url()), &[("content-type", "application/sdp")], po.offer.as_bytes()).await;
    assert_eq!(r.status, 403);
    assert!(r.text().contains("layer_not_permitted"));

    // Discovery and CORS.
    let r = http("HEAD", &w.whep_url(), &[], b"").await;
    assert_eq!(r.status, 200);
    assert_eq!(r.header("link"), Some("<urn:ietf:params:whep>; rel=\"profile\""));
    let r = http("OPTIONS", &format!("{}/whip", w.base()), &[], b"").await;
    assert_eq!((r.status, r.header("accept-post")), (200, Some("application/sdp")));

    // No session was opened by any of these.
    assert_eq!(w.gateway.gateway.records().len(), 0);
    w.stop().await;
}

async fn ice() {
    let w = World::start(|_| {}).await;
    let a = w.auth(&w.publisher, 0);
    let p = publish(&w, &a).await;
    let watcher = watch(&w).await;
    let url = format!("{}{}", w.base(), p.location);
    let mut sent = Sent::default();
    send(&p, &mut sent, 1, 15).await;

    let frag = "a=ice-ufrag:{u}\r\na=ice-pwd:{p}\r\nm=audio 9 UDP/TLS/RTP/SAVPF 111\r\na=mid:0\r\na=candidate:9 1 udp 2130706431 127.0.0.1 9 typ host\r\n";
    let (ufrag, pwd) = (sdp::attr(&p_offer_creds(&p), "ice-ufrag").unwrap().to_string(), sdp::attr(&p_offer_creds(&p), "ice-pwd").unwrap().to_string());
    let trickle = frag.replace("{u}", &ufrag).replace("{p}", &pwd);
    let ct = "application/trickle-ice-sdpfrag";

    // A PATCH names the ICE session it is for: missing, 428; another, 412.
    let r = a.send("PATCH", &url, &[("content-type", ct)], trickle.as_bytes()).await;
    assert_eq!(r.status, 428);
    let r = a.send("PATCH", &url, &[("content-type", ct), ("if-match", "\"not-this-one\"")], trickle.as_bytes()).await;
    assert_eq!(r.status, 412);
    // Trickle ICE: 204, no ETag.
    let etag = format!("\"{}\"", p.etag);
    let r = a.send("PATCH", &url, &[("content-type", ct), ("if-match", &etag)], trickle.as_bytes()).await;
    assert_eq!(r.status, 204);
    assert!(r.header("etag").is_none());
    // Another principal's grant, its own key proved: 403.
    let other = w.auth(&SigningKey::from_bytes(&[0x63; 32]), 0);
    let r = other.send("PATCH", &url, &[("content-type", ct), ("if-match", "*")], trickle.as_bytes()).await;
    assert_eq!(r.status, 403);

    // An ICE restart: new credentials, If-Match *; 200 with the gateway's new ones and a
    // new entity-tag.
    let connected_before = p.count(|e| matches!(e, PeerEvent::Ice(IceConnectionState::Connected) | PeerEvent::Ice(IceConnectionState::Completed)));
    let new_offer = p.restart_offer().await;
    let body = sdp::client_restart_frag(&new_offer);
    let r = a.send("PATCH", &url, &[("content-type", ct), ("if-match", "*")], body.as_bytes()).await;
    assert_eq!(r.status, 200, "{}", r.text());
    assert_eq!(r.header("content-type"), Some(ct));
    let new_etag = r.header("etag").expect("a new entity-tag").to_string();
    assert_ne!(new_etag, etag);
    let f = sdp::parse_frag(&r.text()).expect("the answer is a fragment");
    assert!(f.end_of_candidates && !f.candidates.is_empty());
    assert_ne!(f.ufrag.as_deref(), sdp::attr(&p.answer, "ice-ufrag"), "the gateway's credentials changed");
    p.restarted(r.text()).await.expect("the client applies the restart");
    assert!(
        until(STEP, || p.count(|e| matches!(e, PeerEvent::Ice(IceConnectionState::Connected) | PeerEvent::Ice(IceConnectionState::Completed))) > connected_before).await,
        "ICE connects again"
    );
    // The old entity-tag is stale now.
    let r = a.send("PATCH", &url, &[("content-type", ct), ("if-match", &etag)], trickle.as_bytes()).await;
    assert_eq!(r.status, 412);

    // The session's resource asks for the proof too: a PATCH or a DELETE that presents the
    // grant alone is refused (live-rooms §10.5.4 step 1), and the session stays.
    let bare = bearer(&a.token());
    let r = http("PATCH", &url, &[("content-type", ct), ("authorization", &bare), ("if-match", "*")], trickle.as_bytes()).await;
    assert_eq!(r.status, 401);
    assert!(r.text().contains("pop_missing"), "{}", r.text());
    let r = http("DELETE", &url, &[("authorization", &bare)], b"").await;
    assert_eq!(r.status, 401);
    assert!(r.text().contains("pop_missing"), "{}", r.text());

    // Media after the restart reaches the relay with the rest.
    send(&p, &mut sent, 1, 15).await;
    assert!(received(&w, &p, sent.video.len() + sent.audio.len()).await, "every frame reached the gateway");
    assert_eq!(delete(&w, &p.location, &a).await.status, 200);
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await);
    settle(&watcher, Duration::from_millis(500)).await;
    let video = watcher.tracks.iter().find(|t| t.kind == 2).unwrap();
    let audio = watcher.tracks.iter().find(|t| t.kind == 1).unwrap();
    let (vf, _) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
    let (af, _) = verified(&watcher, audio, &w.gateway_key, Codec::Opus);
    same(&vf, &sent.video, "every video frame, before and after the restart");
    same(&af, &sent.audio, "every audio packet, before and after the restart");
    p.close();
    w.stop().await;
}

/// The client's own ICE credentials: the gateway holds them as the offer's.
fn p_offer_creds(p: &Peer) -> String {
    p.offer.clone()
}