@@ -40,13 +40,78 @@ pub fn uri(did: &str, name: &str) -> String { format!("at://{did}/space/{SPACE_TYPE}/{name}")}/// A space, and what lets requests read its owner's repo in it./// A space, and what lets requests read its owner's repo in it. Its clones/// share what lets them in, renewed for all of them as it runs out: a/// credential lasts minutes, and a runner reads a space for days.#[derive(Clone)]pub struct Space { pub uri: String, pub auth: Auth, /// Whose repo in the space its requests read: the audience a /// credential's name. audience: String, access: Arc<Access>,}/// What lets a space's requests in, and how to get it again.struct Access { held: tokio::sync::Mutex<Held>, renew: Renew,}#[derive(Clone)]enum Held { /// The owner's own session, which reads only their repo there. Session(Box<Session>), /// A credential and its key, to read any member's repo in the space. credential: Option<(String, Arc<P256Key>)>, Credential(String, Arc<P256Key>),}impl Held { /// Whether it ends within a minute, as a session or credential kept on /// disk is renewed. fn ending(&self) -> bool { match self { Held::Session(session) => session.expiring(), Held::Credential(credential, _) => !lasts(credential), } }}/// What gets a space's reader in again: their session, kept under `ctx`,/// and for a credential the PDS of the space's authority, `host`.struct Renew { ctx: Ctx, host: Xrpc, /// Whom the space is read as. did: String,}impl Renew { fn of(ctx: &Ctx, host: &Xrpc, did: &str) -> Renew { Renew { ctx: ctx.clone(), host: host.clone(), did: did.to_string(), } } /// Another of what `held` is: the reader's session as it is now, or a /// credential, kept on disk if another read renewed it already, else /// exchanged afresh. async fn renewed(&self, uri: &str, held: &Held) -> Result<Held> { let session = matches!(held, Held::Session(_)); if !session && let Some(kept) = kept_credential(&self.ctx, uri, &self.did) { return Ok(kept); } let id = identity::resolve(&self.ctx, &self.did).await?; let me = session::ensure(&self.ctx, &id).await?; if session { return Ok(Held::Session(Box::new(me))); } let key = P256Key::generate(); let credential = credential(&self.ctx, &self.host, uri, &me, &key).await?; Ok(store_credential(&self.ctx, uri, &self.did, credential, key)) }}#[derive(Serialize, Deserialize)]@@ -105,48 +170,65 @@ pub fn scope_missing(err: &anyhow::Error) -> bool { }impl Space { /// The owner reads their own repo in the space with their session. pub fn owner(uri: &str, session: &Session) -> Space { Space { uri: uri.to_string(), auth: session.auth(), credential: None, } /// The owner reads their own repo in the space with their session, /// renewed under `ctx` as it runs out. pub fn owner(ctx: &Ctx, uri: &str, session: &Session) -> Space { let held = Held::Session(Box::new(session.clone())); let host = Xrpc::new(ctx.http.clone(), &session.service); let renew = Renew::of(ctx, &host, &session.did); Space::with(uri, &session.did, held, renew) } /// A credential holder reads `audience`'s repo in the space. pub fn member(uri: &str, audience: &str, credential: &str, key: P256Key) -> Space { fn with(uri: &str, audience: &str, held: Held, renew: Renew) -> Space { Space { uri: uri.to_string(), auth: Auth::Headers(signed( &key, &format!("Atproto-Space-Credential {credential}"), audience, )), credential: Some((credential.to_string(), Arc::new(key))), audience: audience.to_string(), access: Arc::new(Access { held: tokio::sync::Mutex::new(held), renew, }), } } /// The space as read from `did`'s repo in it. pub fn at(&self, did: &str) -> Space { Space { auth: self.for_repo(did), audience: did.to_string(), ..self.clone() } } /// What lets a request read the repo this space is read from. pub async fn auth(&self) -> Result<Auth> { self.for_repo(&self.audience).await } /// What lets a request read `did`'s repo in the space, or ask the /// authority (`did` the space's) which repos it holds. Only a /// credential reaches past its holder's own repo. pub fn for_repo(&self, did: &str) -> Auth { match &self.credential { Some((credential, key)) => Auth::Headers(signed( key, pub async fn for_repo(&self, did: &str) -> Result<Auth> { Ok(match self.held().await? { Held::Credential(credential, key) => Auth::Headers(signed( &key, &format!("Atproto-Space-Credential {credential}"), did, )), None => self.auth.clone(), Held::Session(session) => session.auth(), }) } /// What lets this space's requests in, renewed first if it ends within /// a minute: one renewal for every clone reading at once. async fn held(&self) -> Result<Held> { let mut held = self.access.held.lock().await; if held.ending() { // Boxed: a renewal logs in and exchanges, a future far larger // than the read it is part of, and every read of a space has // this one in it. let renewed = Box::pin(self.access.renew.renewed(&self.uri, &held)); *held = renewed.await?; } Ok(held.clone()) } /// Where the space's writes are now, if any member's repo there moved@@ -184,7 +266,7 @@ impl Space { if let Some(c) = &cursor { params.push(("cursor", c)); } let auth = self.for_repo(authority); let auth = self.for_repo(authority).await?; let page: Repos = host .query("com.atproto.space.listRepos", ¶ms, Some(&auth)) .await?;@@ -200,9 +282,6 @@ impl Space { } } /// Joins `uri` as `me`, a member reading `authority`'s repo there: /// a cached credential while it lasts, else a fresh one. `host` is the /// authority's PDS. /// Asks the space's authority to tell `service`, a DID and the entry in /// its document to reach, of each write in the space from now on, as /// `notifyWrite`: until the time it answers, unless asked again. Only@@ -219,13 +298,16 @@ impl Space { expires_at: String, } let body = json!({ "space": self.uri, "service": service }); let auth = self.for_repo(authority); let auth = self.for_repo(authority).await?; let registered: Registered = host .procedure("com.atproto.space.registerNotify", Some(&body), Some(&auth)) .await?; Ok(registered.expires_at) } /// Joins `uri` as `me`, a member reading `authority`'s repo there: /// a cached credential while it lasts, else a fresh one, renewed as it /// runs out. `host` is the authority's PDS. pub async fn join( ctx: &Ctx, host: &Xrpc,@@ -233,39 +315,36 @@ impl Space { authority: &str, me: &Session, ) -> Result<Space> { if let Some(kept) = Space::kept(ctx, uri, authority, &me.did) { if let Some(kept) = Space::kept(ctx, host, uri, authority, &me.did) { return Ok(kept); } let key = P256Key::generate(); let credential = credential(ctx, host, uri, me, &key).await?; Ok(Space::keep(ctx, uri, authority, &me.did, credential, key)) Ok(Space::keep( ctx, host, uri, authority, &me.did, credential, key, )) } /// The credential kept under `ctx` for `did` to read the space `uri`, /// while it has a minute left in it. pub fn kept(ctx: &Ctx, uri: &str, authority: &str, did: &str) -> Option<Space> { let cached = store::read::<Cached>(&kept_at(ctx, uri, did))?; let key = P256Key::from_bytes(&STANDARD.decode(&cached.key).ok()?).ok()?; lasts(&cached.credential).then(|| Space::member(uri, authority, &cached.credential, key)) pub fn kept(ctx: &Ctx, host: &Xrpc, uri: &str, authority: &str, did: &str) -> Option<Space> { let held = kept_credential(ctx, uri, did)?; Some(Space::with(uri, authority, held, Renew::of(ctx, host, did))) } /// Keeps `credential` and its key under `ctx` for `did` to read the /// space `uri` with until it ends, and reads it with them. fn keep( ctx: &Ctx, host: &Xrpc, uri: &str, authority: &str, did: &str, credential: String, key: P256Key, ) -> Space { let cached = Cached { credential: credential.clone(), key: STANDARD.encode(key.to_bytes()), }; // A credential that cannot be cached only costs the next exchange. let _ = store::write(&kept_at(ctx, uri, did), &cached, true); Space::member(uri, authority, &credential, key) let held = store_credential(ctx, uri, did, credential, key); Space::with(uri, authority, held, Renew::of(ctx, host, did)) } /// Joins the space `uri` as `did`, with `token`, a delegation token@@ -283,10 +362,30 @@ impl Space { ) -> Result<Space> { let key = P256Key::generate(); let credential = exchange(host, uri, token, &key).await?; Ok(Space::keep(ctx, uri, authority, did, credential, key)) Ok(Space::keep(ctx, host, uri, authority, did, credential, key)) }}/// The credential kept under `ctx` for `did` to read the space `uri`, while/// it has a minute left in it.fn kept_credential(ctx: &Ctx, uri: &str, did: &str) -> Option<Held> { let cached = store::read::<Cached>(&kept_at(ctx, uri, did))?; let key = P256Key::from_bytes(&STANDARD.decode(&cached.key).ok()?).ok()?; lasts(&cached.credential).then(|| Held::Credential(cached.credential, Arc::new(key)))}/// Keeps `credential` and its key under `ctx` for `did` to read the space/// `uri` with until it ends.fn store_credential(ctx: &Ctx, uri: &str, did: &str, credential: String, key: P256Key) -> Held { let cached = Cached { credential: credential.clone(), key: STANDARD.encode(key.to_bytes()), }; // A credential that cannot be cached only costs the next exchange. let _ = store::write(&kept_at(ctx, uri, did), &cached, true); Held::Credential(credential, Arc::new(key))}/// Where the credential for `did` to read the space `uri` is kept.fn kept_at(ctx: &Ctx, uri: &str, did: &str) -> std::path::PathBuf { let name = format!("{}.{}.json", did.replace(':', "_"), keys::ref_hash(uri));@@ -596,28 +695,71 @@ impl Admin<'_> { mod tests { use super::*; #[test] fn signature_bases() { /// A context with nothing kept, whose accounts are all on `pds`. fn ctx_on(dir: &std::path::Path, pds: &str) -> Ctx { let at = |sub: &str| Some(dir.join(sub).to_string_lossy().into_owned()); Ctx::new(crate::env::Env::from_vars(|k| match k { "KNOTLESS_CONFIG_DIR" => at("config"), "KNOTLESS_CACHE_DIR" => at("cache"), "KNOTLESS_SERVICE" => Some(pds.to_string()), _ => None, })) } /// A JWT that ends `secs` from now. fn ends_in(secs: i64) -> String { let now = (keys::now_micros() / 1_000_000) as i64; P256Key::generate().sign_jwt(&json!({}), &json!({"exp": now + secs})) } fn session(did: &str, service: &str, access_jwt: String) -> Session { Session { did: did.into(), handle: "me.test".into(), service: service.into(), access_jwt, refresh_jwt: "r".into(), oauth: None, } } #[tokio::test] async fn signature_bases() { let key = P256Key::generate(); let headers = |auth: &Auth| auth.headers("GET", "http://pds.test/xrpc/x"); let own = Space { uri: "at://x/space/t/k".into(), auth: Auth::bearer("t"), credential: None, }; assert_eq!(headers(&own.for_repo("did:plc:m")), headers(&own.auth)); let space = Space::member("at://x/space/t/k", "did:plc:o", "cred", key); let space_headers = headers(&space.auth); let dir = tempfile::tempdir().unwrap(); let ctx = ctx_on(dir.path(), "http://pds.test"); let me = session("did:plc:o", "http://pds.test", ends_in(600)); let own = Space::owner(&ctx, "at://x/space/t/k", &me); let own_auth = own.auth().await.unwrap(); assert_eq!( headers(&own.for_repo("did:plc:m").await.unwrap()), headers(&own_auth) ); assert_eq!(headers(&own_auth), headers(&me.auth())); let (cred, key) = (ends_in(600), Arc::new(key)); let held = Held::Credential(cred.clone(), key.clone()); let host = Xrpc::new(ctx.http.clone(), "http://pds.test"); let space = Space::with( "at://x/space/t/k", "did:plc:o", held, Renew::of(&ctx, &host, "m"), ); let space_headers = headers(&space.auth().await.unwrap()); assert_eq!(space_headers.len(), 4); assert_eq!(headers(&space.for_repo("did:plc:o"))[1], space_headers[1]); assert_eq!(headers(&space.for_repo("did:plc:m"))[1].1, "did:plc:m"); let key = P256Key::from_bytes(&space.credential.as_ref().unwrap().1.to_bytes()).unwrap(); let to_o = space.for_repo("did:plc:o").await.unwrap(); assert_eq!(headers(&to_o)[1], space_headers[1]); let to_m = space.for_repo("did:plc:m").await.unwrap(); assert_eq!(headers(&to_m)[1].1, "did:plc:m"); let at = space.at("did:plc:m").auth().await.unwrap(); assert_eq!(headers(&at)[1], headers(&to_m)[1]); assert_eq!( space_headers[0], ( "authorization".into(), "Atproto-Space-Credential cred".into() format!("Atproto-Space-Credential {cred}") ) ); assert_eq!(@@ -636,7 +778,7 @@ mod tests { .strip_suffix(':') .unwrap(); let base = format!( "\"authorization\": Atproto-Space-Credential cred\n\"atproto-space-audience\": did:plc:o\n\"@signature-params\": {params}" "\"authorization\": Atproto-Space-Credential {cred}\n\"atproto-space-audience\": did:plc:o\n\"@signature-params\": {params}" ); key.public() .verify(base.as_bytes(), &STANDARD.decode(sig).unwrap())@@ -665,6 +807,110 @@ mod tests { assert!(!lasts("junk")); } /// A PDS that is both the reader's and the space's authority: it gives /// a credential for each exchange, counting them, and lists the space's /// repos to a credential or session that has not ended, as the alpha /// and pds.js refuse one that has, `JwtExpired`. async fn authority() -> (String, Arc<std::sync::atomic::AtomicUsize>) { use axum::http::{HeaderMap, StatusCode}; use axum::routing::{get, post}; use std::sync::atomic::{AtomicUsize, Ordering}; let exchanged = Arc::new(AtomicUsize::new(0)); let counted = exchanged.clone(); let list = |headers: HeaderMap| async move { let authorization = headers["authorization"].to_str().unwrap_or_default(); let jwt = authorization.split(' ').nth(1).unwrap_or_default(); if lasts(jwt) { return (StatusCode::OK, r#"{"repos":[]}"#); } let expired = r#"{"error":"JwtExpired","message":"token expired"}"#; (StatusCode::UNAUTHORIZED, expired) }; let router = axum::Router::new() .route( "/xrpc/com.atproto.space.getDelegationToken", get(|| async { axum::Json(json!({"token": "t"})) }), ) .route( "/xrpc/com.atproto.space.getSpaceCredential", post(move || async move { counted.fetch_add(1, Ordering::SeqCst); axum::Json(json!({"credential": ends_in(600)})) }), ) .route("/xrpc/com.atproto.space.listRepos", get(list)); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); (format!("http://{addr}"), exchanged) } #[tokio::test] async fn a_credential_held_past_its_end_is_renewed() { use std::sync::atomic::Ordering; let (pds, exchanged) = authority().await; let dir = tempfile::tempdir().unwrap(); let ctx = ctx_on(dir.path(), &pds); let host = Xrpc::new(ctx.http.clone(), &pds); let me = session("did:plc:me", &pds, ends_in(3600)); session::save(&ctx, &me).unwrap(); let uri = "at://did:plc:o/space/t/k"; // Joined a while ago: what it was let in with has ended since, and // the authority refuses it. let (credential, key) = (ends_in(-1), P256Key::generate()); let auth = Auth::Headers(signed( &key, &format!("Atproto-Space-Credential {credential}"), "did:plc:o", )); let params = [("space", uri)]; let refused = host.query::<Value>("com.atproto.space.listRepos", ¶ms, Some(&auth)); let refused = refused.await.unwrap_err(); assert_eq!(XrpcError::of(&refused).unwrap().error, "JwtExpired"); let ended = Held::Credential(credential, Arc::new(key)); let space = Space::with(uri, "did:plc:o", ended, Renew::of(&ctx, &host, &me.did)); let first = space.writers_since(&host, "did:plc:o", None).await.unwrap(); assert_eq!(first, (Vec::new(), None)); assert_eq!(exchanged.load(Ordering::SeqCst), 1); // Its clones, and the space read from a member's repo, read with // the one renewal, and so does another reader as the same account, // from the credential the renewal kept. space.at("did:plc:m").auth().await.unwrap(); space .clone() .moved_since(&host, "did:plc:o", None) .await .unwrap(); let ended = Held::Credential(ends_in(-1), Arc::new(P256Key::generate())); let other = Space::with(uri, "did:plc:o", ended, Renew::of(&ctx, &host, &me.did)); other.moved_since(&host, "did:plc:o", None).await.unwrap(); assert_eq!(exchanged.load(Ordering::SeqCst), 1); assert!(Space::kept(&ctx, &host, uri, "did:plc:o", &me.did).is_some()); // With no session to ask for another, the reader is told so. session::remove(&ctx, &me.did); forget(&ctx); let ended = Held::Credential(ends_in(-1), Arc::new(P256Key::generate())); let lost = Space::with(uri, "did:plc:o", ended, Renew::of(&ctx, &host, &me.did)); let err = lost.auth().await.err().unwrap(); assert!(err.to_string().contains("not logged in"), "{err}"); } #[tokio::test] async fn an_owner_session_held_past_its_end_is_renewed() { let (pds, exchanged) = authority().await; let dir = tempfile::tempdir().unwrap(); let ctx = ctx_on(dir.path(), &pds); let host = Xrpc::new(ctx.http.clone(), &pds); let ended = session("did:plc:o", &pds, ends_in(-1)); let space = Space::owner(&ctx, "at://did:plc:o/space/t/k", &ended); // Its session as it is now, as another read refreshed it. session::save(&ctx, &session("did:plc:o", &pds, ends_in(3600))).unwrap(); space.moved_since(&host, "did:plc:o", None).await.unwrap(); assert_eq!(exchanged.load(std::sync::atomic::Ordering::SeqCst), 0); } #[test] fn denied_is_an_authority_saying_no() { let said = |error: &str| {
chadtmiller.com/knotless