diff --git a/quest/m1/README.md b/quest/m1/README.md index f917cce56c..777ed9a007 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -27,7 +27,6 @@ the transport line in m2 assumes a single stack. ## Quests - [One QUIC backend](/quest/m1/quic-one-backend.md) - quinn and quiche are deleted; noq (and iroh on it) is the only QUIC stack, with the qmux fallbacks untouched -- [One auth path](/quest/m1/auth-one-path.md) - Server, Public, and Refuse become tasks on the `Admissions` queue; `admit()` stays send-plus-await; `Mode` is gone - [Announce event](/quest/m1/api-net-announce.md) - publishers announce prefixes on every wire, consumers scoped by a pattern read the covered path already trimmed, with no `as_prefix().expect()` at 89 call sites - [Origin scoping](/quest/m1/api-net-origin.md) - `scope(root, patterns)` is one fallible call, a fresh origin has a random hop, and the handles stop derefing to `Hop` - [Bindings announce match](/quest/m1/api-origin-scopes.md) - every binding takes a pattern scope and reports the announce match with its captures diff --git a/quest/m1/auth-one-path.md b/quest/m1/auth-one-path.md deleted file mode 100644 index 2c1b7929c2..0000000000 --- a/quest/m1/auth-one-path.md +++ /dev/null @@ -1,78 +0,0 @@ -# [S] Every auth mode is a client of Admissions - -## Goal - -`moq_relay::Auth` has one admission path. `admit()` is send-plus-await on -the `Admissions` queue. `--auth-url`, `--auth-public`, and refuse are -decider tasks that answer that queue; embedded is the same queue with no -task, so the embedder answers it. A change to how a session is admitted, -re-checked, or ended is made once. - -`--auth-url` / `--auth-public` / refuse / embedded stay as configuration. -`Mode` is gone. Gateways still do not call `admit`. `admit_fixed` stays the -LAN bypass and is not a fifth mode. Not this quest: the lease clock, in-band -tokens, or a live stats retier. - -## Plan - -Fold Server, Public, and Refuse onto the queue Embedded already uses. - -- `Auth` always holds the `Admission` sender. `admit()` sends and awaits the - oneshot, the way Embedded does today, including the ten-second bound. -- `Config::init` builds that queue and spawns the matching decider, so it - needs a Tokio runtime. `a_public_config_admits_anonymous_and_certificate_alike` - becomes a `#[tokio::test]`. Empty config (no url, no public) still means - embedded: `Relay::load` returns `Admissions` and spawns nothing. -- `--auth-url`: a loop on `Admissions::next` that `tokio::spawn`s one task - per admission, `Client::connect(request)`, then `grant` or `refuse`. A slow - server does not serialize connects. -- `--auth-public`: a loop that answers `grant(lease::Consumer::fixed(grant))`. -- Refuse (`Auth::refuse`, LAN-only with no listener): a loop that answers - `auth::Error::Refused`. A dropped `Admissions` stays an outage (502), not a - policy. `Auth::refuse` also requires a Tokio runtime; cover synchronous - construction with the same runtime-requirement tests as `Config::init`. -- Dropping every `Auth` clone ends `next()` with `None` and the decider - exits. Dropping the decider first makes later `admit()` fail unavailable, - the same as dropping `Admissions` today. -- `Mode` is deleted. The server path is tested through the same grant/refuse - answers the embedded tests already use. - -How a downstream embedder uses the landed API, for reference. moq.pro's edge -serves the contract on a unix socket today and points `--auth-url` at itself; -with `Admissions` its `auth::start` takes `relay.admissions()` after -`Relay::load` and runs one loop: - -```rust -while let Some(admission) = admissions.next().await { - let authorizer = authorizer.clone(); - tokio::spawn(async move { - match authorizer.decide(Facts::from_request(&admission.request), false).await { - Ok(grant) if grant.revalidate.is_none() => admission.grant(lease::Consumer::fixed(grant)), - Ok(grant) => { - let (producer, consumer) = lease::Producer::new(grant.clone()); - admission.grant(consumer); - authorizer.drive(producer, admission.request, grant).await; - } - Err(Error::Refused(why)) => admission.refuse(auth::Error::Refused), - Err(Error::Unavailable(why)) => admission.refuse(auth::Error::Unavailable(why)), - } - }); -} -``` - -`drive` is the edge's re-check loop: sleep the cadence, `decide(.., true)`, -`producer.update` or `producer.revoke`, and `producer.closed()` for the end -event with the totals the session reported through `lease::Consumer::close`. -The unix socket, the axum router, and the JSON round trip go away; the -gateways keep not calling `admit`, and relay/CLI `--listen` keep calling it. - -Public API: breaking on moq-relay's unpublished auth module (`Mode` gone, -`Config::init` and `Auth::refuse` require a runtime). Wire: none. - -## Related - -- [Auth embedder](/quest/m2/auth-embedder.md) - the lease owns the re-check clock, so `drive` above becomes a loop over `producer.due()` instead of a second driver -- [In-band auth](/quest/m2/auth/README.md) - a token presented in band is - another admission on the same lease -- [Stats retier](/quest/m2/stats-retier.md) - what a re-checked grant should - do to a live session, whichever path re-checked it diff --git a/quest/m2/auth-embedder.md b/quest/m2/auth-embedder.md index 36f440d0e0..235cf2f7e3 100644 --- a/quest/m2/auth-embedder.md +++ b/quest/m2/auth-embedder.md @@ -17,8 +17,7 @@ Additive on `moq-auth` and `moq-relay`, so on main after the merge: `producer.due().await -> Due::{Revalidate, Expired}`, `update(grant)` reschedules, `failed()` applies the backoff. `moq_auth::Client` becomes a ten-line loop over it, and an embedder answering `Admissions` writes the - same ten lines instead of a second driver. [One auth path](/quest/m1/auth-one-path.md) - should reference this so its fold does not ship a third. + same ten lines instead of a second driver. - `Cluster::admit(&self, auth: &Auth, request: moq_auth::Request) -> Result` with `Admitted { lease: Lease, publisher: Option, subscriber: Option, stats: stats::Session }` diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 028f2eaf65..b057f77ee3 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -15,9 +15,9 @@ use serde_with::{OneOrMany, serde_as}; use tokio::sync::{mpsc, oneshot}; use url::Url; -/// The longest an embedder may take to answer an admission, the bound +/// The longest a decider may take to answer an admission, the bound /// `moq_auth::Client` puts on a server, so a stalled decider refuses rather -/// than parks the sessions behind it. +/// than parks the session behind it. const ADMIT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); /// Where every session's grant comes from. Exactly one of `url` and the public @@ -101,21 +101,21 @@ impl Config { /// Build the [`Auth`] this configuration describes. `tls` is the client /// identity an `https://` server is dialed with; `node` names this relay in - /// every request. + /// every request. Must be called within a Tokio runtime, which drives the + /// admission decider. pub fn init(&self, node: impl Into, tls: &moq_tokio::tls::Connect) -> anyhow::Result { self.validate()?; - let mode = match (&self.url, self.public_grant()) { + let decider = match (&self.url, self.public_grant()) { (Some(url), _) => { let tls = tls.build()?; - Mode::Server(moq_auth::Client::new(url.clone(), Some(tls))?) + Decider::Server(moq_auth::Client::new(url.clone(), Some(tls))?) } - (None, Some(grant)) => Mode::Public(grant), + (None, Some(grant)) => Decider::Public(grant), (None, None) => unreachable!("validated above"), }; - Ok(Auth { - mode: Arc::new(mode), - node: Arc::from(node.into()), - }) + let (auth, admissions) = Auth::embedded(node); + decider.spawn(admissions); + Ok(auth) } } @@ -140,7 +140,7 @@ pub enum Error { impl From for Error { fn from(err: moq_auth::Error) -> Self { match err { - moq_auth::Error::Refused => Self::Refused, + moq_auth::Error::Refused | moq_auth::Error::UselessGrant => Self::Refused, other => Self::Unavailable(other.to_string()), } } @@ -301,23 +301,38 @@ impl Lease { } } -enum Mode { +enum Decider { Server(moq_auth::Client), Public(Grant), - /// The embedding process decides: every session is queued for whoever holds - /// the [`Admissions`], and admits nothing once they are gone. - Embedded(mpsc::UnboundedSender), - /// Nothing admits an ordinary session; only a locally decided grant (the LAN - /// mesh credential) gets through. What a `--cluster-lan` process with no - /// listener of its own runs. Refuse, } -/// Admits sessions: asks the server, hands out the static public grant, or queues -/// the session for the embedder. +impl Decider { + fn spawn(self, mut admissions: Admissions) { + tokio::spawn(async move { + while let Some(admission) = admissions.next().await { + match &self { + Self::Server(client) => { + let client = client.clone(); + tokio::spawn(async move { + match client.connect(admission.request.clone()).await { + Ok(lease) => admission.grant(lease), + Err(err) => admission.refuse(err.into()), + } + }); + } + Self::Public(grant) => admission.grant(lease::Consumer::fixed(grant.clone())), + Self::Refuse => admission.refuse(Error::Refused), + } + } + }); + } +} + +/// Admits sessions by queueing every request for one admission decider. #[derive(Clone)] pub struct Auth { - mode: Arc, + admissions: mpsc::UnboundedSender, node: Arc, } @@ -328,7 +343,7 @@ impl Auth { pub fn embedded(node: impl Into) -> (Self, Admissions) { let (sender, receiver) = mpsc::unbounded_channel(); let auth = Self { - mode: Arc::new(Mode::Embedded(sender)), + admissions: sender, node: Arc::from(node.into()), }; (auth, Admissions(receiver)) @@ -336,11 +351,11 @@ impl Auth { /// An `Auth` that refuses every session a server or a public grant would have /// decided, admitting only what the relay decides for itself (a LAN peer). + /// Must be called within a Tokio runtime, which drives the refusal decider. pub fn refuse(node: impl Into) -> Self { - Self { - mode: Arc::new(Mode::Refuse), - node: Arc::from(node.into()), - } + let (auth, admissions) = Self::embedded(node); + Decider::Refuse.spawn(admissions); + auth } /// The name this relay puts in every request. @@ -356,27 +371,17 @@ impl Auth { /// Admit a session: the lease it holds, carrying the scope the origin applies. pub async fn admit(&self, request: Request) -> Result { let path = request.path.clone(); - let consumer = match self.mode.as_ref() { - Mode::Server(client) => client.connect(request).await?, - // A certificate is a fact for a server to weigh; with no server it admits - // nothing on its own, so the peer gets what any anonymous session gets. - Mode::Public(grant) => lease::Consumer::fixed(grant.clone()), - Mode::Embedded(admissions) => { - let (reply, answer) = oneshot::channel(); - admissions - .send(Admission { request, reply }) - .map_err(|_| Error::Unavailable("nobody is answering admissions".into()))?; - let consumer = tokio::time::timeout(ADMIT_TIMEOUT, answer) - .await - .map_err(|_| Error::Unavailable("the admission timed out".into()))? - .map_err(|_| Error::Unavailable("the admission went unanswered".into()))??; - // Held to what a server's answer is held to: a grant that admits nothing - // or asks for a re-check without a bound is the decider's bug, not a refusal. - consumer.grant().validate()?; - consumer - } - Mode::Refuse => return Err(Error::Refused), - }; + let (reply, answer) = oneshot::channel(); + self.admissions + .send(Admission { request, reply }) + .map_err(|_| Error::Unavailable("nobody is answering admissions".into()))?; + let consumer = tokio::time::timeout(ADMIT_TIMEOUT, answer) + .await + .map_err(|_| Error::Unavailable("the admission timed out".into()))? + .map_err(|_| Error::Unavailable("the admission went unanswered".into()))??; + // Every decider is held to the same answer contract: a grant that admits + // nothing or asks for a re-check without a bound is a bug, not a refusal. + consumer.grant().validate()?; Ok(Lease::new(&path, consumer)) } @@ -506,18 +511,34 @@ mod tests { assert!(grant.publish.is_empty()); } - #[test] - fn a_public_config_admits_anonymous_and_certificate_alike() { + #[tokio::test] + async fn a_public_config_admits_anonymous_and_certificate_alike() { let auth = config(None, &["anon/**"]) .init("relay-1", &moq_tokio::tls::Connect::default()) .unwrap(); let request = auth.request(moq_auth::Transport::Quic, "/anon/room"); - let lease = futures::executor::block_on(auth.admit(request)).unwrap(); + let lease = auth.admit(request).await.unwrap(); assert_eq!(lease.token().root, Path::new("anon/room").to_owned()); assert_eq!(lease.token().subscribe, patterns(&["anon/**"])); assert_eq!(lease.token().tier, Tier::default()); } + #[test] + fn config_init_requires_a_runtime() { + let result = std::panic::catch_unwind(|| { + let _ = config(None, &["anon/**"]) + .init("relay-1", &moq_tokio::tls::Connect::default()) + .unwrap(); + }); + assert!(result.is_err()); + } + + #[test] + fn refuse_requires_a_runtime() { + let result = std::panic::catch_unwind(|| Auth::refuse("relay-1")); + assert!(result.is_err()); + } + /// The embedder's answer is the session's verdict; an answer that never comes, /// or a decider that is gone, is an outage rather than a refusal. #[tokio::test] @@ -542,9 +563,8 @@ mod tests { let lease = auth.admit(request()).await.expect("granted"); assert_eq!(lease.token().root, Path::new("anon/room").to_owned()); assert!(matches!(auth.admit(request()).await, Err(Error::Refused))); - for _ in 0..2 { - assert!(matches!(auth.admit(request()).await, Err(Error::Unavailable(_)))); - } + assert!(matches!(auth.admit(request()).await, Err(Error::Refused))); + assert!(matches!(auth.admit(request()).await, Err(Error::Unavailable(_)))); }; let (admissions, ()) = tokio::join!(decide, admit); @@ -552,6 +572,27 @@ mod tests { assert!(matches!(auth.admit(request()).await, Err(Error::Unavailable(_)))); } + #[tokio::test] + async fn dropping_the_last_auth_ends_admissions() { + let (auth, mut admissions) = Auth::embedded("relay-1"); + let clone = auth.clone(); + drop(auth); + assert!( + tokio::time::timeout(std::time::Duration::ZERO, admissions.next()) + .await + .is_err() + ); + drop(clone); + assert!(admissions.next().await.is_none()); + } + + #[tokio::test] + async fn refuse_answers_the_admission_queue() { + let auth = Auth::refuse("relay-1"); + let request = auth.request(moq_auth::Transport::Quic, "/anon/room"); + assert!(matches!(auth.admit(request).await, Err(Error::Refused))); + } + #[test] fn token_keeps_every_grant_pattern() { let mut grant = Grant::new(patterns(&["alice/**"]), patterns(&["**"]));