open-standards/joule-code/joule-code-rs/crates/joule-moq-ingest/src/gateway.rs

//! The gateway: its rooms, its key, its sessions, and the HTTP resources of WHIP (RFC
//! 9725) and WHEP (`draft-ietf-wish-whep-05`).
//!
//! | resource | methods |
//! |---|---|
//! | `/whip` | `POST` an offer: `201` with the answer, `Location`, `ETag`; `OPTIONS`; `GET`, `HEAD` |
//! | `/whip/s/<id>` | `PATCH` trickle ICE or an ICE restart; `DELETE`; `GET`, `HEAD`; `OPTIONS` |
//! | `/whep/<room>/<owner>[?layer=<n>]` | `POST` an offer for the owner's tracks; `HEAD` (discovery); `OPTIONS`; `GET` |
//! | `/whep/s/<id>` | `PATCH`; `DELETE`; `GET`, `HEAD`; `OPTIONS` |
//!
//! Every request but `OPTIONS` carries `Authorization: Bearer <token>`, the token a room
//! grant (`WRG1`) in base64url without padding, and beside it `WAI-Room-Proof`, a proof of
//! possession of the grant's key over a single-use challenge the gateway issued in
//! `WAI-Room-Challenge` (live-rooms §10.5.3; `extensions/live-ingest.md` §3). A grant is
//! public, every participant of its room holds it, so the grant alone proves nothing.

use crate::config::GatewayConfig;
use crate::package::Signer;
use crate::record::{HopSink, KeepAndAppend, SignedRecord};
use crate::relay::Link;
use crate::room::{Admitted, Reason, Room};
use crate::sdp::{self, Frag};
use base64::Engine;
use bytes::Bytes;
use ed25519_dalek::{SigningKey, VerifyingKey};
use http_body_util::{BodyExt, Full, Limited};
use hyper::body::Incoming;
use hyper::header::{HeaderName, HeaderValue};
use hyper::{Method, Request, Response, StatusCode};
use joule_moq::grant_proof::{self, GrantProof, HEADER_PROOF};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::sync::{mpsc, oneshot};

pub const SDP: &str = "application/sdp";
pub const SDPFRAG: &str = "application/trickle-ice-sdpfrag";
pub const PROBLEM: &str = "application/problem+json";
const REALM: &str = "wai-ingest";

/// An answer to one request.
#[derive(Clone, Debug)]
pub struct Reply {
    pub status: StatusCode,
    pub headers: Vec<(HeaderName, String)>,
    pub body: Vec<u8>,
}

impl Reply {
    pub fn new(status: StatusCode) -> Reply {
        Reply { status, headers: Vec::new(), body: Vec::new() }
    }
    pub fn header(mut self, k: HeaderName, v: impl Into<String>) -> Reply {
        self.headers.push((k, v.into()));
        self
    }
    pub fn body(mut self, content_type: &str, b: impl Into<Vec<u8>>) -> Reply {
        self.headers.push((hyper::header::CONTENT_TYPE, content_type.to_string()));
        self.body = b.into();
        self
    }
}

/// A refusal, as a problem details object (RFC 9457; RFC 9725 §4.1), with the reason
/// code `extensions/live-ingest.md` §8 gives it.
pub fn problem(status: StatusCode, reason: &str, detail: &str) -> Reply {
    let body = serde_json::json!({
        "detail": detail,
        "reason": reason,
        "status": status.as_u16(),
        "title": status.canonical_reason().unwrap_or(""),
        "type": "about:blank",
    });
    Reply::new(status).body(PROBLEM, serde_json::to_vec(&body).unwrap())
}

/// `401` with `WWW-Authenticate` (RFC 6750 §3): no token, or one that is not good here.
pub fn unauthorized(reason: Option<&str>) -> Reply {
    let (v, r, d) = match reason {
        None => (format!("Bearer realm=\"{REALM}\""), "token_missing", "a room grant token is required (RFC 6750 §2.1)"),
        Some(code) => (format!("Bearer realm=\"{REALM}\", error=\"invalid_token\", error_description=\"{code}\""), code, "the room grant token is not good here"),
    };
    let mut p = problem(StatusCode::UNAUTHORIZED, r, d);
    p.headers.push((hyper::header::WWW_AUTHENTICATE, v));
    p
}

/// `403` with `error="insufficient_scope"` (RFC 6750 §3.1): a good grant without the
/// power the request needs.
pub fn forbidden(reason: &str, detail: &str) -> Reply {
    let mut p = problem(StatusCode::FORBIDDEN, reason, detail);
    p.headers.push((hyper::header::WWW_AUTHENTICATE, format!("Bearer realm=\"{REALM}\", error=\"insufficient_scope\", error_description=\"{reason}\"")));
    p
}

/// The microsecond clock of the gateway, monotonic from its start, in Unix time.
#[derive(Clone, Copy, Debug)]
pub struct Clock {
    base: Instant,
    base_unix_us: u64,
}

impl Clock {
    pub fn new() -> Clock {
        let base_unix_us = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_micros() as u64);
        Clock { base: Instant::now(), base_unix_us }
    }
    pub fn unix_us(&self) -> u64 {
        self.unix_at(Instant::now())
    }
    pub fn unix_at(&self, i: Instant) -> u64 {
        match i.checked_duration_since(self.base) {
            Some(d) => self.base_unix_us + d.as_micros() as u64,
            None => self.base_unix_us.saturating_sub(self.base.duration_since(i).as_micros() as u64),
        }
    }
}

impl Default for Clock {
    fn default() -> Self {
        Clock::new()
    }
}

/// Where a challenge may be used (live-rooms §10.5.3: "the challenge it holds for the
/// session or the request"): at an endpoint, by a request that opens a session; or at one
/// session's resource, by that session's next request.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum Scope {
    Endpoint,
    Session(String),
}

/// The challenges the gateway issued and has not consumed, each with where it may be used
/// and when it was issued, and those it consumed, kept so a proof presented again is
/// refused `pop_replayed` rather than `pop_wrong_challenge` (live-rooms §10.5.4 step 6).
pub struct Challenges {
    issued: Mutex<HashMap<[u8; 32], (Scope, Instant)>>,
    consumed: Mutex<HashMap<[u8; 32], Instant>>,
    ttl: Duration,
    max: usize,
}

impl Challenges {
    pub fn new(ttl: Duration, max: usize) -> Challenges {
        Challenges { issued: Mutex::new(HashMap::new()), consumed: Mutex::new(HashMap::new()), ttl, max: max.max(1) }
    }

    /// Keep `m` within its bound: drop what is older than `keep`, then the oldest.
    fn prune<V>(m: &mut HashMap<[u8; 32], V>, when: impl Fn(&V) -> Instant, keep: Duration, max: usize) {
        m.retain(|_, v| when(v).elapsed() <= keep);
        while m.len() >= max {
            let Some(oldest) = m.iter().min_by_key(|(_, v)| when(v)).map(|(k, _)| *k) else { break };
            m.remove(&oldest);
        }
    }

    /// A fresh challenge, from the operating system's generator, for `scope`.
    pub fn issue(&self, scope: Scope) -> [u8; 32] {
        use rand_core_06::RngCore;
        let mut c = [0u8; 32];
        rand_core_06::OsRng.fill_bytes(&mut c);
        let mut m = self.issued.lock().unwrap();
        Self::prune(&mut m, |v| v.1, self.ttl, self.max);
        m.insert(c, (scope, Instant::now()));
        c
    }

    /// Whether `c` is a challenge issued for `scope`, not consumed, and not past its time.
    pub fn held(&self, c: &[u8; 32], scope: &Scope) -> bool {
        self.issued.lock().unwrap().get(c).is_some_and(|(s, at)| s == scope && at.elapsed() <= self.ttl)
    }

    /// Consume `c`: it is good for one request.
    pub fn consume(&self, c: &[u8; 32]) {
        self.issued.lock().unwrap().remove(c);
        let mut m = self.consumed.lock().unwrap();
        // A consumed challenge is kept twice its lifetime: past it, a proof over it would
        // be refused as stale anyway.
        Self::prune(&mut m, |v| *v, self.ttl * 2, self.max);
        m.insert(*c, Instant::now());
    }

    pub fn is_consumed(&self, c: &[u8; 32]) -> bool {
        self.consumed.lock().unwrap().contains_key(c)
    }
}

/// The proof step of [`Gateway::authorize`] (live-rooms §10.5.3, §10.5.4), on its own: a
/// grant's digest and principal as this gateway's reader gives them, the `WAI-Room-Proof`
/// field's value if the request carried one, whether the gateway holds a challenge for the
/// request's scope, whether it consumed one before, and the request's context. The
/// challenge the proof names, accepted; the caller consumes it.
pub fn check_http_proof(
    digest: &[u8; 32],
    principal: &[u8; 32],
    header: Option<&str>,
    held: impl Fn(&[u8; 32]) -> bool,
    consumed: impl Fn(&[u8; 32]) -> bool,
    context: &str,
) -> Result<[u8; 32], grant_proof::ProofRefusal> {
    let proof: Option<Vec<u8>> = header.map(|t| grant_proof::from_base64url(t.trim().trim_end_matches('=')).unwrap_or_default());
    let presented = proof.as_deref().and_then(|p| GrantProof::read(p).ok()).map(|p| p.challenge);
    // The challenge held for this request: the proof's own, where the gateway issued it
    // for this scope and it is still good; otherwise one no proof names, so step 6 tells a
    // challenge consumed before from any other.
    let challenge = match presented {
        Some(c) if held(&c) => c,
        Some(mut c) => {
            c[0] ^= 0xff;
            c
        }
        None => [0; 32],
    };
    grant_proof::check_proof(digest, principal, proof.as_deref(), &challenge, consumed, context).map(|p| p.challenge)
}

/// What a session's HTTP resource asks of its task.
pub enum Cmd {
    Patch { frag: Frag, if_match: Option<String>, reply: oneshot::Sender<Reply> },
    Delete { reply: oneshot::Sender<()> },
}

#[derive(Clone)]
pub struct Handle {
    pub tx: mpsc::Sender<Cmd>,
    /// SHA-256 of the token the session was opened with: its resource answers that
    /// token's holder alone.
    pub token_hash: [u8; 32],
    /// Frames the session has taken from the WebRTC stack (WHIP) or handed to it (WHEP).
    pub frames: Arc<AtomicU64>,
}

/// The gateway's counters.
#[derive(Debug, Default)]
pub struct Stats {
    pub whip_opened: AtomicU64,
    pub whep_opened: AtomicU64,
    pub refused: AtomicU64,
}

pub struct Gateway {
    pub cfg: GatewayConfig,
    pub signer: Signer,
    pub rooms: HashMap<[u8; 32], Arc<Room>>,
    pub link: Link,
    pub clock: Clock,
    /// The first hops whose receipts a WHEP session accepts.
    pub trust: Vec<VerifyingKey>,
    pub whip: Mutex<HashMap<String, Handle>>,
    pub whep: Mutex<HashMap<String, Handle>>,
    pub whip_active: AtomicUsize,
    pub whep_active: AtomicUsize,
    pub stats: Stats,
    /// The proof-of-possession challenges (live-rooms §10.5.3).
    pub challenges: Challenges,
    kept: Arc<KeepAndAppend>,
    pub sink: Arc<dyn HopSink>,
}

/// `did:key` of an Ed25519 key (WAI jwp-receipts §7.1): `did:key:z`, then base58btc of
/// `0xed 0x01` and the key.
pub fn did_key(key: &[u8; 32]) -> String {
    const ALPHABET: &[u8] = b"123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz";
    let mut bytes = vec![0xed, 0x01];
    bytes.extend_from_slice(key);
    let mut digits: Vec<u8> = Vec::new();
    for &b in &bytes {
        let mut carry = b as u32;
        for d in digits.iter_mut() {
            carry += (*d as u32) << 8;
            *d = (carry % 58) as u8;
            carry /= 58;
        }
        while carry > 0 {
            digits.push((carry % 58) as u8);
            carry /= 58;
        }
    }
    let zeros = bytes.iter().take_while(|&&b| b == 0).count();
    let mut s = String::from("did:key:z");
    s.extend(std::iter::repeat_n('1', zeros));
    s.extend(digits.iter().rev().map(|&d| ALPHABET[d as usize] as char));
    s
}

pub fn random_hex(n: usize) -> String {
    use rand_core_06::RngCore;
    let mut b = vec![0u8; n];
    rand_core_06::OsRng.fill_bytes(&mut b);
    hex::encode(b)
}

impl Gateway {
    pub fn new(cfg: GatewayConfig) -> Result<Gateway, String> {
        let key = match (&cfg.signing.key_seed, &cfg.signing.key_path) {
            (Some(s), _) => SigningKey::from_bytes(s),
            (None, Some(p)) => {
                let b = std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()))?;
                let seed: [u8; 32] = b.get(..32).and_then(|s| s.try_into().ok()).ok_or_else(|| format!("{}: not 32 bytes", p.display()))?;
                SigningKey::from_bytes(&seed)
            }
            (None, None) => SigningKey::generate(&mut rand_core_06::OsRng),
        };
        let pk = key.verifying_key().to_bytes();
        let signer_id = if cfg.signing.signer_id.is_empty() { did_key(&pk) } else { cfg.signing.signer_id.clone() };
        let mut rooms = HashMap::new();
        for r in &cfg.rooms {
            let bytes = match (&r.descriptor, &r.descriptor_path) {
                (Some(b), _) => b.clone(),
                (None, Some(p)) => std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()))?,
                (None, None) => return Err("a room names no descriptor".into()),
            };
            let mut revoked = Vec::new();
            for h in &r.revoked {
                let d: [u8; 32] = hex::decode(h).map_err(|e| format!("a revoked digest: {e}"))?.try_into().map_err(|_| "a revoked digest is not 32 bytes".to_string())?;
                revoked.push(d);
            }
            let room = Room::new(&bytes, revoked).map_err(|e| format!("a room's descriptor is refused: {e}"))?;
            rooms.insert(room.id, Arc::new(room));
        }
        let mut trust = Vec::new();
        for t in &cfg.whep.trust {
            let b: [u8; 32] = hex::decode(t).map_err(|e| format!("a trusted key: {e}"))?.try_into().map_err(|_| "a trusted key is not 32 bytes".to_string())?;
            trust.push(crate::room::strict_key(&b).map_err(|e| format!("a trusted key: {e}"))?);
        }
        if trust.is_empty() {
            trust.push(key.verifying_key());
        }
        let link = Link::from_config(&cfg.relay)?;
        let kept = Arc::new(KeepAndAppend { kept: Mutex::new(Vec::new()), path: cfg.summary.out_path.clone() });
        let origin_label = cfg.signing.origin_label.clone();
        Ok(Gateway {
            signer: Signer { key: Arc::new(key), signer_id, origin_label },
            rooms,
            link,
            clock: Clock::new(),
            trust,
            whip: Mutex::new(HashMap::new()),
            whep: Mutex::new(HashMap::new()),
            whip_active: AtomicUsize::new(0),
            whep_active: AtomicUsize::new(0),
            stats: Stats::default(),
            challenges: Challenges::new(Duration::from_secs(cfg.http.challenge_ttl_s), cfg.http.max_challenges),
            sink: kept.clone(),
            kept,
            cfg,
        })
    }

    pub fn public_key(&self) -> [u8; 32] {
        self.signer.key.verifying_key().to_bytes()
    }

    /// Every session record so far.
    pub fn records(&self) -> Vec<SignedRecord> {
        self.kept.kept.lock().unwrap().clone()
    }

    pub fn reserve(&self, whep: bool) -> bool {
        let (n, max) = if whep { (&self.whep_active, self.cfg.limits.max_whep_sessions) } else { (&self.whip_active, self.cfg.limits.max_whip_sessions) };
        if n.fetch_add(1, Ordering::SeqCst) >= max {
            n.fetch_sub(1, Ordering::SeqCst);
            return false;
        }
        true
    }

    pub fn release(&self, whep: bool) {
        if whep { &self.whep_active } else { &self.whip_active }.fetch_sub(1, Ordering::SeqCst);
    }

    pub fn busy(&self) -> Reply {
        problem(StatusCode::SERVICE_UNAVAILABLE, "session_limit", "the gateway holds as many sessions as it is configured for").header(hyper::header::RETRY_AFTER, self.cfg.limits.retry_after_s.to_string())
    }

    /// The bearer token of a request (RFC 6750 §2.1), decoded.
    fn token(req: &Request<Incoming>) -> Result<Vec<u8>, Reply> {
        let Some(v) = req.headers().get(hyper::header::AUTHORIZATION).and_then(|v| v.to_str().ok()) else { return Err(unauthorized(None)) };
        let mut parts = v.splitn(2, ' ');
        let (Some(scheme), Some(t)) = (parts.next(), parts.next()) else { return Err(unauthorized(None)) };
        if !scheme.eq_ignore_ascii_case("bearer") {
            return Err(unauthorized(None));
        }
        let t = t.trim().trim_end_matches('=');
        base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(t).map_err(|_| unauthorized(Some("token_malformed")))
    }

    /// Admit a token: the room it names, then live-rooms §4.3 and the gateway's checks;
    /// `opening` for a request that opens a session (its grant's single-use nonce).
    pub fn admit(&self, token: &[u8], opening: bool) -> Result<(Arc<Room>, Admitted), Reply> {
        let room_id: Option<[u8; 32]> = token.get(5..37).and_then(|s| s.try_into().ok());
        let Some(room) = room_id.and_then(|r| self.rooms.get(&r)) else {
            // Refused for what a reader finds first (live-rooms §4.3 step 1), or for
            // naming a room the gateway does not hold.
            let reason = if token.len() < 4 {
                Reason::Truncated
            } else if &token[..4] != b"WRG1" {
                Reason::Malformed
            } else if token.len() < 37 {
                Reason::Truncated
            } else {
                Reason::RoomUnknown
            };
            return Err(unauthorized(Some(reason.code())));
        };
        room.admit(token, self.clock.unix_us(), &self.cfg.http.origin, opening).map(|a| (room.clone(), a)).map_err(|r| unauthorized(Some(r.code())))
    }

    /// The authority a request addressed: its `:authority`, or its `Host` (live-rooms
    /// §10.5.1: a verifier compares the target with the authority and path it received).
    fn authority(req: &Request<Incoming>) -> String {
        req.uri()
            .authority()
            .map(|a| a.as_str().to_string())
            .or_else(|| req.headers().get(hyper::header::HOST).and_then(|v| v.to_str().ok()).map(str::to_string))
            .unwrap_or_default()
    }

    /// The grant a request presents, admitted (`opening` for a request that opens a
    /// session), and its proof of possession, checked as live-rooms §10.5.4 checks it with
    /// this gateway's reading of the grant: its digest and its principal, a challenge held
    /// for `scope`, the challenges consumed, and the request's method and target. A proof
    /// that passes consumes its challenge. Every refusal is `401` with
    /// `error="invalid_token"`; the caller's response carries a fresh challenge.
    pub fn authorize(&self, req: &Request<Incoming>, scope: &Scope, opening: bool) -> Result<(Arc<Room>, Admitted, Vec<u8>), Reply> {
        let token = Self::token(req)?;
        let (room, ad) = self.admit(&token, opening)?;
        let header = req.headers().get(HEADER_PROOF).map(|v| v.to_str().unwrap_or(""));
        let ctx = grant_proof::context(req.method().as_str(), &Self::authority(req), req.uri().path());
        match check_http_proof(&ad.digest, &ad.grant.principal, header, |c| self.challenges.held(c, scope), |c| self.challenges.is_consumed(c), &ctx) {
            Ok(c) => {
                self.challenges.consume(&c);
                Ok((room, ad, token))
            }
            Err(r) => Err(unauthorized(Some(r.code()))),
        }
    }

    fn cors(&self, mut r: Reply) -> Reply {
        if !self.cfg.http.cors_allow_origin.is_empty() {
            r.headers.push((hyper::header::ACCESS_CONTROL_ALLOW_ORIGIN, self.cfg.http.cors_allow_origin.clone()));
            r.headers.push((hyper::header::ACCESS_CONTROL_EXPOSE_HEADERS, "Location, ETag, Link, Retry-After, WWW-Authenticate, Accept-Post, Accept-Patch, WAI-Room-Challenge".into()));
        }
        r
    }

    fn options(&self, endpoint: bool) -> Reply {
        let mut r = Reply::new(StatusCode::OK);
        if endpoint {
            r = r.header(HeaderName::from_static("accept-post"), SDP);
        } else {
            r = r.header(HeaderName::from_static("accept-patch"), SDPFRAG);
        }
        if !self.cfg.http.cors_allow_origin.is_empty() {
            r = r
                .header(hyper::header::ACCESS_CONTROL_ALLOW_METHODS, "GET, HEAD, POST, PATCH, DELETE, OPTIONS")
                .header(hyper::header::ACCESS_CONTROL_ALLOW_HEADERS, "Authorization, Content-Type, If-Match, WAI-Room-Proof")
                .header(hyper::header::ACCESS_CONTROL_MAX_AGE, "600");
        }
        r
    }

    async fn body(&self, req: Request<Incoming>) -> Result<Vec<u8>, Reply> {
        let limit = self.cfg.http.max_body_bytes;
        let t = Duration::from_millis(self.cfg.http.request_timeout_ms);
        match tokio::time::timeout(t, Limited::new(req.into_body(), limit).collect()).await {
            Ok(Ok(c)) => Ok(c.to_bytes().to_vec()),
            Ok(Err(_)) => Err(problem(StatusCode::PAYLOAD_TOO_LARGE, "body_too_large", "the request body passes the gateway's bound")),
            Err(_) => Err(problem(StatusCode::REQUEST_TIMEOUT, "body_timeout", "the request body did not arrive in time")),
        }
    }

    fn content_type_is(req: &Request<Incoming>, t: &str) -> bool {
        req.headers().get(hyper::header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).is_some_and(|v| v.split(';').next().unwrap_or("").trim().eq_ignore_ascii_case(t))
    }

    fn session(&self, whep: bool, id: &str) -> Option<Handle> {
        if whep { &self.whep } else { &self.whip }.lock().unwrap().get(id).cloned()
    }

    /// Frames a session has carried so far, while it runs.
    pub fn session_frames(&self, whep: bool, id: &str) -> Option<u64> {
        self.session(whep, id).map(|h| h.frames.load(Ordering::Relaxed))
    }

    pub fn forget(&self, whep: bool, id: &str) {
        if whep { &self.whep } else { &self.whip }.lock().unwrap().remove(id);
    }

    /// One request.
    pub async fn handle(self: &Arc<Self>, req: Request<Incoming>) -> Reply {
        let path = req.uri().path().trim_end_matches('/').to_string();
        let query = req.uri().query().unwrap_or("").to_string();
        let seg: Vec<&str> = path.split('/').filter(|s| !s.is_empty()).collect();
        let m = req.method().clone();
        let presented = req.headers().contains_key(hyper::header::AUTHORIZATION) && m != Method::OPTIONS;
        let scope = match seg.as_slice() {
            ["whip" | "whep", "s", id] => Scope::Session(id.to_string()),
            _ => Scope::Endpoint,
        };
        // A request that carries a grant on an endpoint's GET or HEAD proves it too; one
        // that carries none is a discovery probe (WHEP §4.2), answered without one.
        if presented && matches!(m, Method::GET | Method::HEAD) && scope == Scope::Endpoint
            && let Err(r) = self.authorize(&req, &scope, false)
        {
            return self.respond(r, presented, &scope);
        }
        let r = match (seg.as_slice(), &m) {
            // A session's resource first: `/whep/s/<id>` would otherwise read as a room.
            (["whip" | "whep", "s", _], &Method::OPTIONS) => self.options(false),
            ([kind @ ("whip" | "whep"), "s", id], _) => {
                let whep = *kind == "whep";
                let id = id.to_string();
                self.on_session(req, whep, &id).await
            }
            (["whip"], &Method::OPTIONS) | (["whep", _, _], &Method::OPTIONS) => self.options(true),
            (["whip"], &Method::POST) => self.post_whip(req).await,
            (["whip"], &Method::GET) | (["whip"], &Method::HEAD) => Reply::new(StatusCode::NO_CONTENT),
            (["whep", room, owner], &Method::POST) => {
                let (room, owner) = (room.to_string(), owner.to_string());
                self.post_whep(req, room, owner, &query).await
            }
            (["whep", _, _], &Method::HEAD) | (["whep", _, _], &Method::GET) => Reply::new(StatusCode::OK)
                .header(hyper::header::CONTENT_TYPE, SDP)
                .header(hyper::header::LINK, "<urn:ietf:params:whep>; rel=\"profile\""),
            (["whip"], _) | (["whep", _, _], _) => Reply::new(StatusCode::METHOD_NOT_ALLOWED).header(hyper::header::ALLOW, "GET, HEAD, POST, OPTIONS"),
            _ => problem(StatusCode::NOT_FOUND, "not_found", "no such resource"),
        };
        self.respond(r, presented, &scope)
    }

    /// The response, with a fresh challenge where the request carried a grant
    /// (live-rooms §10.5.3: on a `401`, and on each response, for the client's next
    /// request): one for the session a `201` opened, the session's own for a response at
    /// its resource, an endpoint's otherwise. A session a DELETE ended takes none.
    fn respond(&self, mut r: Reply, presented: bool, scope: &Scope) -> Reply {
        if r.status.is_client_error() || r.status.is_server_error() {
            self.stats.refused.fetch_add(1, Ordering::Relaxed);
        }
        let ended = matches!(scope, Scope::Session(id) if r.status == StatusCode::OK && self.session(false, id).is_none() && self.session(true, id).is_none());
        if presented && !ended {
            let next = if r.status == StatusCode::CREATED {
                r.headers
                    .iter()
                    .find(|(k, _)| *k == hyper::header::LOCATION)
                    .and_then(|(_, v)| v.rsplit('/').next())
                    .map_or(Scope::Endpoint, |id| Scope::Session(id.to_string()))
            } else {
                scope.clone()
            };
            let c = self.challenges.issue(next);
            r.headers.push((HeaderName::from_static("wai-room-challenge"), grant_proof::base64url(&c)));
        }
        self.cors(r)
    }

    async fn post_whip(self: &Arc<Self>, req: Request<Incoming>) -> Reply {
        let (room, ad, token) = match self.authorize(&req, &Scope::Endpoint, true) {
            Ok(x) => x,
            Err(r) => return r,
        };
        if !Self::content_type_is(&req, SDP) {
            return problem(StatusCode::UNSUPPORTED_MEDIA_TYPE, "not_sdp", "an offer is application/sdp");
        }
        let body = match self.body(req).await {
            Ok(b) => b,
            Err(r) => return r,
        };
        let Ok(offer) = String::from_utf8(body) else { return problem(StatusCode::BAD_REQUEST, "offer_malformed", "the offer is not UTF-8") };
        crate::whip::create(self.clone(), room, ad, crate::room::sha256(&[&token]), offer).await
    }

    async fn post_whep(self: &Arc<Self>, req: Request<Incoming>, room_hex: String, owner: String, query: &str) -> Reply {
        let (room, ad, token) = match self.authorize(&req, &Scope::Endpoint, true) {
            Ok(x) => x,
            Err(r) => return r,
        };
        if room_hex != hex::encode(room.id) {
            return forbidden("wrong_room", "the grant is not for the room this resource names");
        }
        let layer = match query.split('&').find_map(|kv| kv.strip_prefix("layer=")) {
            None => 0u8,
            Some(l) => match l.parse::<u8>() {
                Ok(l) => l,
                Err(_) => return problem(StatusCode::BAD_REQUEST, "layer_malformed", "the layer is a decimal integer"),
            },
        };
        let owner_ok = owner.len() == 64 && owner.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b));
        if !owner_ok {
            return problem(StatusCode::NOT_FOUND, "owner_unknown", "the owner is a principal's key in 64 lowercase hexadecimal digits");
        }
        if !Self::content_type_is(&req, SDP) {
            return problem(StatusCode::UNSUPPORTED_MEDIA_TYPE, "not_sdp", "an offer is application/sdp");
        }
        let body = match self.body(req).await {
            Ok(b) => b,
            Err(r) => return r,
        };
        let Ok(offer) = String::from_utf8(body) else { return problem(StatusCode::BAD_REQUEST, "offer_malformed", "the offer is not UTF-8") };
        crate::whep::create(self.clone(), room, ad, crate::room::sha256(&[&token]), owner, layer, offer).await
    }

    async fn on_session(self: &Arc<Self>, req: Request<Incoming>, whep: bool, id: &str) -> Reply {
        let Some(h) = self.session(whep, id) else { return problem(StatusCode::NOT_FOUND, "session_unknown", "no such session") };
        // The grant is public, so the session's own grant presented without its key's
        // proof is refused like any other (live-rooms §10.5).
        let token = match self.authorize(&req, &Scope::Session(id.to_string()), false) {
            Ok((_, _, t)) => t,
            Err(r) => return r,
        };
        if crate::room::sha256(&[&token]) != h.token_hash {
            return forbidden("not_the_session_token", "the session answers the token it was opened with");
        }
        match *req.method() {
            Method::GET | Method::HEAD => Reply::new(StatusCode::NO_CONTENT),
            Method::DELETE => {
                let (tx, rx) = oneshot::channel();
                if h.tx.send(Cmd::Delete { reply: tx }).await.is_ok() {
                    let _ = tokio::time::timeout(Duration::from_secs(30), rx).await;
                }
                Reply::new(StatusCode::OK)
            }
            Method::PATCH => {
                if !Self::content_type_is(&req, SDPFRAG) {
                    return problem(StatusCode::UNSUPPORTED_MEDIA_TYPE, "not_sdpfrag", "a PATCH carries application/trickle-ice-sdpfrag");
                }
                let if_match = req.headers().get(hyper::header::IF_MATCH).and_then(|v| v.to_str().ok()).map(str::to_string);
                let body = match self.body(req).await {
                    Ok(b) => b,
                    Err(r) => return r,
                };
                let Some(frag) = std::str::from_utf8(&body).ok().and_then(sdp::parse_frag) else {
                    return problem(StatusCode::BAD_REQUEST, "sdpfrag_malformed", "the body is not an SDP fragment (RFC 8840)");
                };
                let (tx, rx) = oneshot::channel();
                if h.tx.send(Cmd::Patch { frag, if_match, reply: tx }).await.is_err() {
                    return problem(StatusCode::NOT_FOUND, "session_unknown", "the session has ended");
                }
                match tokio::time::timeout(Duration::from_secs(10), rx).await {
                    Ok(Ok(r)) => r,
                    _ => problem(StatusCode::SERVICE_UNAVAILABLE, "session_busy", "the session did not answer in time"),
                }
            }
            _ => Reply::new(StatusCode::METHOD_NOT_ALLOWED).header(hyper::header::ALLOW, "GET, HEAD, PATCH, DELETE, OPTIONS"),
        }
    }
}

/// An `If-Match` against the session's current entity-tag (RFC 9110 §13.1.1; RFC 9725
/// §4.3.1): `428` when absent, `412` when it names another, `Ok(true)` for `*`.
pub fn check_if_match(if_match: Option<&str>, etag: &str) -> Result<bool, Reply> {
    let Some(v) = if_match else { return Err(problem(StatusCode::PRECONDITION_REQUIRED, "if_match_missing", "a PATCH names the ICE session it is for (If-Match)")) };
    if v.trim() == "*" {
        return Ok(true);
    }
    let quoted = format!("\"{etag}\"");
    if v.split(',').map(str::trim).any(|t| t == quoted) {
        Ok(false)
    } else {
        Err(problem(StatusCode::PRECONDITION_FAILED, "if_match_stale", "the PATCH names an ICE session that is not the current one"))
    }
}

/// The HTTP server, bound and serving.
pub struct GatewayServer {
    pub addr: SocketAddr,
    pub gateway: Arc<Gateway>,
    task: tokio::task::JoinHandle<()>,
}

impl GatewayServer {
    pub async fn start(cfg: GatewayConfig) -> Result<GatewayServer, String> {
        let listener = tokio::net::TcpListener::bind(cfg.http.listen).await.map_err(|e| format!("{}: {e}", cfg.http.listen))?;
        let addr = listener.local_addr().map_err(|e| e.to_string())?;
        let gateway = Arc::new(Gateway::new(cfg)?);
        let gw = gateway.clone();
        let task = tokio::spawn(async move {
            loop {
                let Ok((stream, _)) = listener.accept().await else { continue };
                let gw = gw.clone();
                tokio::spawn(async move {
                    let svc = hyper::service::service_fn(move |req: Request<Incoming>| {
                        let gw = gw.clone();
                        async move { Ok::<_, std::convert::Infallible>(to_response(gw.handle(req).await)) }
                    });
                    let mut b = hyper::server::conn::http1::Builder::new();
                    b.timer(hyper_util::rt::TokioTimer::new()).header_read_timeout(Duration::from_secs(10));
                    let _ = b.serve_connection(hyper_util::rt::TokioIo::new(stream), svc).await;
                });
            }
        });
        Ok(GatewayServer { addr, gateway, task })
    }

    /// End every session, then stop listening.
    pub async fn shutdown(self) {
        let handles: Vec<Handle> = self.gateway.whip.lock().unwrap().values().cloned().chain(self.gateway.whep.lock().unwrap().values().cloned()).collect();
        for h in handles {
            let (tx, rx) = oneshot::channel();
            if h.tx.send(Cmd::Delete { reply: tx }).await.is_ok() {
                let _ = tokio::time::timeout(Duration::from_secs(30), rx).await;
            }
        }
        self.task.abort();
    }
}

fn to_response(r: Reply) -> Response<Full<Bytes>> {
    let mut resp = Response::new(Full::new(Bytes::from(r.body)));
    *resp.status_mut() = r.status;
    for (k, v) in r.headers {
        if let Ok(v) = HeaderValue::from_str(&v) {
            resp.headers_mut().append(k, v);
        }
    }
    resp
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn did_key_decodes_to_its_key() {
        const ALPHABET: &str = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz";
        let key = [0x2e; 32];
        let d = did_key(&key);
        assert!(d.starts_with("did:key:z6Mk"), "{d}");
        // base58btc decode of what follows `z`.
        let mut n: Vec<u8> = Vec::new();
        for c in d["did:key:z".len()..].chars() {
            let mut carry = ALPHABET.find(c).unwrap() as u32;
            for b in n.iter_mut().rev() {
                carry += (*b as u32) * 58;
                *b = (carry & 0xff) as u8;
                carry >>= 8;
            }
            while carry > 0 {
                n.insert(0, (carry & 0xff) as u8);
                carry >>= 8;
            }
        }
        assert_eq!(&n[..2], &[0xed, 0x01]);
        assert_eq!(&n[2..], &key);
    }

    #[test]
    fn if_match_rules() {
        assert_eq!(check_if_match(None, "a").unwrap_err().status, StatusCode::PRECONDITION_REQUIRED);
        assert_eq!(check_if_match(Some("\"b\""), "a").unwrap_err().status, StatusCode::PRECONDITION_FAILED);
        assert!(check_if_match(Some("*"), "a").unwrap());
        assert!(!check_if_match(Some("\"x\", \"a\""), "a").unwrap());
    }
}