From 57c33efb9b92d5d3e9fe26e2660b810d70636a7e Mon Sep 17 00:00:00 2001 From: dawn Date: Fri, 16 Oct 2026 20:43:20 +0300 Subject: [PATCH] knot-xrpc: tell syncers apart from users and check their tokens without a pds Signed-off-by: dawn --- knot2/crates/knot-acl/src/lib.rs | 2 +- knot2/crates/knot-acl/src/repos.rs | 2 + knot2/crates/knot-acl/src/viewer.rs | 63 +++++++++- knot2/crates/knot-atproto/src/lib.rs | 21 ++-- knot2/crates/knot-atproto/src/resolve.rs | 59 ++++++---- knot2/crates/knot-keyfill/src/lib.rs | 2 +- knot2/crates/knot-server/src/main.rs | 2 + knot2/crates/knot-sim/src/harness.rs | 1 + knot2/crates/knot-sim/tests/lfs_roundtrip.rs | 1 + knot2/crates/knot-ssh/tests/ssh_push.rs | 1 + knot2/crates/knot-xrpc/src/lib.rs | 70 ++++++----- knot2/crates/knot-xrpc/src/plc.rs | 11 +- knot2/crates/knot-xrpc/src/repos.rs | 6 +- knot2/crates/knot-xrpc/src/social.rs | 6 +- knot2/crates/knot-xrpc/src/tests.rs | 1 + knot2/crates/knot-xrpc/tests/common/mod.rs | 98 ++++++++++++---- knot2/crates/knot-xrpc/tests/syncers.rs | 116 +++++++++++++++++++ 17 files changed, 365 insertions(+), 97 deletions(-) create mode 100644 knot2/crates/knot-xrpc/tests/syncers.rs diff --git a/knot2/crates/knot-acl/src/lib.rs b/knot2/crates/knot-acl/src/lib.rs index 18409f369..039fe4d2c 100644 --- a/knot2/crates/knot-acl/src/lib.rs +++ b/knot2/crates/knot-acl/src/lib.rs @@ -17,7 +17,7 @@ pub use repos::{ ContributeGrant, PushGrant, ReadGrant, Repos, Social, push_grant, read_grant, read_grant_by_path, }; -pub use viewer::Viewer; +pub use viewer::{Admitted, DeclaredSyncer, Member, SyncerRefused, Viewer}; pub trait Acl { fn is_admin(&self, who: &AccountDid) -> bool; diff --git a/knot2/crates/knot-acl/src/repos.rs b/knot2/crates/knot-acl/src/repos.rs index debfb9de1..852120fef 100644 --- a/knot2/crates/knot-acl/src/repos.rs +++ b/knot2/crates/knot-acl/src/repos.rs @@ -140,6 +140,8 @@ fn reads_private(index: &Index, repo: &RepoDid, viewer: &Viewer) -> bool { let owner = index.owner_of(repo).ready_or_default(); let member = match viewer.who() { Who::Anonymous => return false, + // lease coverage wires in next + Who::Syncer(_) => return false, Who::Account(who, now) => { let owns = owner.is_some_and(|owner| owner.is(who)); let collaborates = matches!( diff --git a/knot2/crates/knot-acl/src/viewer.rs b/knot2/crates/knot-acl/src/viewer.rs index b1d7b4f30..80455d525 100644 --- a/knot2/crates/knot-acl/src/viewer.rs +++ b/knot2/crates/knot-acl/src/viewer.rs @@ -1,5 +1,5 @@ use knot_atproto::Authenticated; -use knot_types::{AccountDid, OfferedKey, UnixSeconds}; +use knot_types::{AccountDid, Nsid, OfferedKey, ServiceDid, Syncers, UnixSeconds}; // only a verified jwt or an ssh login makes one and never a bare did #[derive(Debug, Clone, PartialEq, Eq)] @@ -11,6 +11,8 @@ pub(crate) enum Who { Account(AccountDid, UnixSeconds), // a key only proves possession and anyone can publish it so each repo decides who it means Key(OfferedKey, UnixSeconds), + // a syncer reads public repos the same as anyone and never as a member + Syncer(ServiceDid), } impl Viewer { @@ -27,10 +29,19 @@ impl Viewer { Self(Who::Key(key, now)) } + pub fn member(member: Member) -> Self { + let verified_at = member.0.verified_at(); + Self(Who::Account(member.0.into_did(), verified_at)) + } + + pub fn declared_syncer(syncer: DeclaredSyncer) -> Self { + Self(Who::Syncer(ServiceDid::from(syncer.0.did()))) + } + pub fn account(&self) -> Option<&AccountDid> { match &self.0 { Who::Account(account, _) => Some(account), - Who::Anonymous | Who::Key(..) => None, + Who::Anonymous | Who::Key(..) | Who::Syncer(_) => None, } } @@ -50,9 +61,49 @@ impl Viewer { } } -impl From for Viewer { - fn from(proof: Authenticated) -> Self { - let verified_at = proof.verified_at(); - Self(Who::Account(proof.into_did(), verified_at)) +pub struct Member(Authenticated); + +impl Member { + pub fn admit(proof: Authenticated, syncers: &Syncers) -> Result { + match syncers.get(proof.did()) { + None => Ok(Self(proof)), + Some(_) => Err(SyncerRefused), + } + } + + pub fn did(&self) -> &AccountDid { + self.0.did() + } + + pub fn into_did(self) -> AccountDid { + self.0.into_did() + } + + pub fn verified_at(&self) -> UnixSeconds { + self.0.verified_at() + } +} + +pub struct DeclaredSyncer(Authenticated); + +pub enum Admitted { + Member(Member), + Syncer(DeclaredSyncer), +} + +#[derive(Debug)] +pub struct SyncerRefused; + +impl Admitted { + pub fn of( + proof: Authenticated, + syncers: &Syncers, + method: &Nsid, + ) -> Result { + match syncers.get(proof.did()) { + None => Ok(Self::Member(Member(proof))), + Some(syncer) if syncer.allows(method) => Ok(Self::Syncer(DeclaredSyncer(proof))), + Some(_) => Err(SyncerRefused), + } } } diff --git a/knot2/crates/knot-atproto/src/lib.rs b/knot2/crates/knot-atproto/src/lib.rs index ed58c84b4..14bfb8248 100644 --- a/knot2/crates/knot-atproto/src/lib.rs +++ b/knot2/crates/knot-atproto/src/lib.rs @@ -527,7 +527,7 @@ impl Atproto { did: &AccountDid, ) -> Result, AtprotoError> { let http = &self.http; - let pds = &identity.pds; + let pds = identity.pds()?; let pages = stream::try_unfold(Page::First(PUBKEY_MAX_PAGES), move |state| async move { let (cursor, budget) = match state { Page::Done | Page::First(0) | Page::Next(_, 0) => { @@ -571,7 +571,7 @@ impl Atproto { match read { RecordPresence::Absent if !cached.fresh => { let current = self.fetch_identity(owner).await?; - match current.pds.url() == cached.value.pds.url() { + match current.pds().ok() == cached.value.pds().ok() { true => Ok(RecordPresence::Absent), false => self.read_declared(¤t, owner, rkey).await, } @@ -604,7 +604,7 @@ impl Atproto { collection: &Nsid, rkey: &str, ) -> Result { - let url = get_record_url(&identity.pds, owner, collection, rkey)?; + let url = get_record_url(identity.pds()?, owner, collection, rkey)?; resolve::guard_fetch_url(&url)?; let response = self.http.execute(HttpRequest::get(url)).await?; match response.status { @@ -740,9 +740,10 @@ impl Atproto { ) -> Result { let identity = self.resolve_identity(subject).await?; let method = pointer::put_record_method(); - let url = xrpc_url(&identity.pds, &method)?; + let pds = identity.pds()?; + let url = xrpc_url(pds, &method)?; resolve::guard_fetch_url(&url)?; - let audience = pointer::pds_service_did(&identity.pds)?; + let audience = pointer::pds_service_did(pds)?; let now = UnixSeconds::new((self.clock.now_unix_micros().get() / 1_000_000) as i64); let body = pointer::put_record_body(subject, rkey, record)?; let mut request = json_post(url, bytes::Bytes::from(body)); @@ -1012,7 +1013,10 @@ mod tests { }); let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc()); let identity = atproto.resolve_identity(&did(SQUID)).await.unwrap(); - assert_eq!(identity.pds.url().as_str(), "https://pds.oyster.cafe/"); + assert_eq!( + identity.pds().unwrap().url().as_str(), + "https://pds.oyster.cafe/" + ); assert_eq!(identity.primary_handle().unwrap().as_str(), "nel.pet"); } @@ -1497,7 +1501,10 @@ mod tests { calling the record deleted would revoke on a guess" ); assert_eq!( - (intact.pds.url().as_str(), lookups.load(Ordering::SeqCst)), + ( + intact.pds().unwrap().url().as_str(), + lookups.load(Ordering::SeqCst) + ), ("https://pds.oyster.cafe/", 2), "a confirmation that fails leaves the document every other caller reads alone: a 400 \ from the directory mustn't cost this account its ssh keys or its service auth, and \ diff --git a/knot2/crates/knot-atproto/src/resolve.rs b/knot2/crates/knot-atproto/src/resolve.rs index e9c2c398b..62b32e688 100644 --- a/knot2/crates/knot-atproto/src/resolve.rs +++ b/knot2/crates/knot-atproto/src/resolve.rs @@ -102,7 +102,7 @@ impl ResolveError { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct PdsEndpoint(Url); impl PdsEndpoint { @@ -148,10 +148,15 @@ pub struct Identity { pub did: AccountDid, pub handles: Vec, pub signing_key: CryptoKey<'static>, - pub pds: PdsEndpoint, + // syncers publish a key and no pds and a jwt check only needs the key + pds: Option, } impl Identity { + pub fn pds(&self) -> Result<&PdsEndpoint, ResolveError> { + self.pds.as_ref().ok_or(ResolveError::MissingPds) + } + pub fn primary_handle(&self) -> Option<&Handle> { self.handles.first() } @@ -227,12 +232,16 @@ pub(crate) fn identity_from_document( }); } let signing_key = atproto_signing_key(&document)?; - let pds_endpoint = document.pds_endpoint().ok_or(ResolveError::MissingPds)?; - let pds = Url::parse(pds_endpoint.as_str()) - .map_err(|_| ResolveError::BadPds { - value: pds_endpoint.as_str().to_string(), + let pds = document + .pds_endpoint() + .map(|endpoint| { + Url::parse(endpoint.as_str()) + .map_err(|_| ResolveError::BadPds { + value: endpoint.as_str().to_string(), + }) + .and_then(PdsEndpoint::new) }) - .and_then(PdsEndpoint::new)?; + .transpose()?; let handles = document .handles() .iter() @@ -438,7 +447,10 @@ mod tests { method: MethodKind::Multikey, }); let identity = identity_from_document(&did(SQUID), &body).unwrap(); - assert_eq!(identity.pds.url().as_str(), "https://pds.oyster.cafe/"); + assert_eq!( + identity.pds().unwrap().url().as_str(), + "https://pds.oyster.cafe/" + ); assert_eq!(identity.primary_handle().unwrap().as_str(), "nel.pet"); } @@ -521,22 +533,6 @@ mod tests { } const DOC_CASES: &[DocCase] = &[ - DocCase { - name: "document declares no atproto_pds service", - requested: SQUID, - body: || { - body(serde_json::json!({ - "id": SQUID, - "verificationMethod": [{ - "id": format!("{SQUID}#atproto"), - "type": "Multikey", - "controller": SQUID, - "publicKeyMultibase": sample_key() - }] - })) - }, - expect: |r| matches!(r, Err(ResolveError::MissingPds)), - }, DocCase { name: "document declares no signing key", requested: SQUID, @@ -629,6 +625,21 @@ mod tests { }); } + #[test] + fn a_document_with_a_key_and_no_pds_still_resolves() { + let document = body(serde_json::json!({ + "id": SQUID, + "verificationMethod": [{ + "id": format!("{SQUID}#atproto"), + "type": "Multikey", + "controller": SQUID, + "publicKeyMultibase": sample_key() + }] + })); + let identity = identity_from_document(&did(SQUID), &document).unwrap(); + assert!(matches!(identity.pds(), Err(ResolveError::MissingPds))); + } + #[test] fn only_a_missing_document_reads_as_an_account_that_no_longer_exists() { assert!( diff --git a/knot2/crates/knot-keyfill/src/lib.rs b/knot2/crates/knot-keyfill/src/lib.rs index a95dc48c1..f1bfa741d 100644 --- a/knot2/crates/knot-keyfill/src/lib.rs +++ b/knot2/crates/knot-keyfill/src/lib.rs @@ -491,7 +491,7 @@ async fn published_keys( let _permit = idle_permit(pass.slots, pass.pace).await; atproto.resolve_identity(did).await? }; - wait_for_turn(pass.pacer, identity.pds.url(), atproto.now()).await; + wait_for_turn(pass.pacer, identity.pds()?.url(), atproto.now()).await; let _permit = idle_permit(pass.slots, pass.pace).await; atproto.pubkeys_at(&identity, did).await } diff --git a/knot2/crates/knot-server/src/main.rs b/knot2/crates/knot-server/src/main.rs index b89a76cc3..f83091f6b 100644 --- a/knot2/crates/knot-server/src/main.rs +++ b/knot2/crates/knot-server/src/main.rs @@ -297,6 +297,7 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { .context("atproto.plc_directory isn't a valid PLC base URL")?, )); let admins: BTreeSet<_> = config.server.admins.iter().cloned().collect(); + let syncers = config.syncers()?; let admission = config.acl.admission; let contribution_policy = config.repo.contribution_policy; let write_limits = knot_resource::RateLimit { @@ -668,6 +669,7 @@ async fn run(config: knot_config::Validated) -> anyhow::Result<()> { blobs: Arc::clone(&blob_views), catalog: Arc::clone(&catalog), rkeys: Default::default(), + syncers, }); let ssh_base = ssh_base.with_ref_surface(Arc::new(knot_xrpc::RefProjection { state: Arc::clone(&xrpc_state), diff --git a/knot2/crates/knot-sim/src/harness.rs b/knot2/crates/knot-sim/src/harness.rs index e79e637a7..dc503b123 100644 --- a/knot2/crates/knot-sim/src/harness.rs +++ b/knot2/crates/knot-sim/src/harness.rs @@ -880,6 +880,7 @@ fn assemble_router(parts: StateParts) -> (Router, FakeState) { blobs: Default::default(), catalog: Arc::new(knot_messages::Catalog::defaults()), rkeys: Default::default(), + syncers: Default::default(), }); let resolver = knot_pack::index_resolver(Arc::clone(&index)); ( diff --git a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs index e7f265903..f1d07f60c 100644 --- a/knot2/crates/knot-sim/tests/lfs_roundtrip.rs +++ b/knot2/crates/knot-sim/tests/lfs_roundtrip.rs @@ -433,6 +433,7 @@ async fn spawn(published_line: String, with_h3: bool) -> World { blobs: Default::default(), catalog: Arc::new(knot_messages::Catalog::defaults()), rkeys: Default::default(), + syncers: Default::default(), }); let (_imports_stop, imports_halt) = tokio::sync::watch::channel(false); diff --git a/knot2/crates/knot-ssh/tests/ssh_push.rs b/knot2/crates/knot-ssh/tests/ssh_push.rs index b33d6def9..1f7f07797 100644 --- a/knot2/crates/knot-ssh/tests/ssh_push.rs +++ b/knot2/crates/knot-ssh/tests/ssh_push.rs @@ -484,6 +484,7 @@ async fn spawn_server_core( ), )), rkeys: Default::default(), + syncers: Default::default(), }); let firehose = Arc::clone(&state.firehose); (Some(state), Some(firehose)) diff --git a/knot2/crates/knot-xrpc/src/lib.rs b/knot2/crates/knot-xrpc/src/lib.rs index 579cec42c..c69ff9672 100644 --- a/knot2/crates/knot-xrpc/src/lib.rs +++ b/knot2/crates/knot-xrpc/src/lib.rs @@ -73,7 +73,7 @@ use serde::de::DeserializeOwned; use serde_json::json; use crate::legacy_admin::LegacyAdminSecret; -use knot_acl::{PushGrant, ReadGrant, Repos, Viewer}; +use knot_acl::{Admitted, Member, PushGrant, ReadGrant, Repos, Viewer}; use knot_atproto::{Atproto, AtprotoError, Authenticated, ServiceJwt}; use knot_events::{EventLog, SubscriberGate}; pub use knot_git::ArchiveLimit; @@ -84,7 +84,7 @@ use knot_runtime::{Clock, Entropy, HttpTransport}; use knot_secrets::SealedStore; use knot_types::{ AccountDid, AdmissionPolicy, AppviewEndpoint, CiLogsAddr, ClonePath, ContributionPolicy, - KnotHostname, KnotId, KnotServiceUrl, Nsid, OwnerDid, OwnerRef, RepoDid, UnixSeconds, + KnotHostname, KnotId, KnotServiceUrl, Nsid, OwnerDid, OwnerRef, RepoDid, Syncers, UnixSeconds, }; use base64::Engine; @@ -223,6 +223,7 @@ pub struct XrpcState { pub catalog: Arc, pub firehose: Arc>, pub rkeys: RecordKeys, + pub syncers: Syncers, } pub fn knot_repo(knot_did: &KnotId) -> Result { @@ -238,16 +239,29 @@ impl XrpcState { self.knot_service_url.authority() } + fn viewer_of(&self, proof: Authenticated, method: &Nsid) -> Result { + match Admitted::of(proof, &self.syncers, method).map_err(|_| syncer_refused(method))? { + Admitted::Member(member) => Ok(Viewer::member(member)), + Admitted::Syncer(syncer) => Ok(Viewer::declared_syncer(syncer)), + } + } + + fn refuse_syncer(&self, proof: Authenticated, method: &Nsid) -> Result { + Member::admit(proof, &self.syncers).map_err(|_| syncer_refused(method)) + } + pub(crate) async fn authenticate( &self, headers: &HeaderMap, method: &Method, - ) -> Result { + ) -> Result { let token = bearer(headers)?; - self.atproto + let proof = self + .atproto .verify_service_jwt(&token, method.nsid()) .await - .map_err(map_verify_error) + .map_err(map_verify_error)?; + self.refuse_syncer(proof, method.nsid()) } pub(crate) async fn authenticate_for_repo( @@ -255,12 +269,14 @@ impl XrpcState { headers: &HeaderMap, method: &Method, repo: &RepoDid, - ) -> Result { + ) -> Result { let token = bearer(headers)?; - self.atproto + let proof = self + .atproto .verify_service_jwt_for_repo(&token, method.nsid(), repo) .await - .map_err(map_verify_error) + .map_err(map_verify_error)?; + self.refuse_syncer(proof, method.nsid()) } // a bad token errors instead of reading as anonymous so the caller knows it's broken @@ -277,16 +293,17 @@ impl XrpcState { let audience = knot_atproto::token_audience(&token) .ok() .filter(|aud| aud == repo); - self.verify_reusable(&token, method.nsid(), audience.as_ref()) - .await - .map(Viewer::from) + let proof = self + .verify_reusable(&token, method.nsid(), audience.as_ref()) + .await?; + self.viewer_of(proof, method.nsid()) } pub(crate) async fn authenticate_write( &self, headers: &HeaderMap, method: &Method, - ) -> Result { + ) -> Result { let actor = self.authenticate(headers, method).await?; admit_write(self, actor.did())?; Ok(actor) @@ -297,7 +314,7 @@ impl XrpcState { headers: &HeaderMap, method: &Method, repo: &RepoDid, - ) -> Result { + ) -> Result { let actor = self.authenticate_for_repo(headers, method, repo).await?; admit_write(self, actor.did())?; Ok(actor) @@ -334,20 +351,11 @@ impl XrpcState { .map_err(map_verify_error) } - pub(crate) async fn authenticate_push( - &self, - headers: &HeaderMap, - ) -> Result { + pub(crate) async fn authenticate_push(&self, headers: &HeaderMap) -> Result { let token = git_credential(headers)?; let method = Nsid::new_owned(PUSH_NSID).expect("Push NSID is always a valid NSID"); - self.atproto - .verify_service_jwt_guarded( - &token, - &method, - knot_atproto::ReplayGuard::ReusableUntilExpiry, - ) - .await - .map_err(map_verify_error) + let proof = self.verify_reusable(&token, &method, None).await?; + self.refuse_syncer(proof, &method) } // owner/path urls take any audience, pinning one would let a prober map paths to dids @@ -379,7 +387,7 @@ impl XrpcState { if proof.is_ok() { guard.refund(); } - proof.map(Viewer::from) + proof.and_then(|proof| self.viewer_of(proof, &method)) } async fn verify_reusable( @@ -405,6 +413,12 @@ impl XrpcState { } } +fn syncer_refused(method: &Nsid) -> XrpcError { + XrpcError::auth_required(format!( + "method binding mismatch: token isn't allowed to call {method}" + )) +} + fn map_verify_error(error: AtprotoError) -> XrpcError { if error.is_transient() { XrpcError::upstream_unavailable(error.to_string()) @@ -780,8 +794,8 @@ pub(crate) async fn authenticate_pusher( Ok(actor_and_viewer(proof)) } -pub(crate) fn actor_and_viewer(proof: Authenticated) -> (AccountDid, Viewer) { - (proof.did().clone(), Viewer::from(proof)) +pub(crate) fn actor_and_viewer(member: Member) -> (AccountDid, Viewer) { + (member.did().clone(), Viewer::member(member)) } pub(crate) async fn run_blocking(task: F) -> Result diff --git a/knot2/crates/knot-xrpc/src/plc.rs b/knot2/crates/knot-xrpc/src/plc.rs index b782861d5..8a363eb92 100644 --- a/knot2/crates/knot-xrpc/src/plc.rs +++ b/knot2/crates/knot-xrpc/src/plc.rs @@ -38,7 +38,10 @@ struct DeletedAccount { deleted_at_micros: u64, } -pub(crate) fn load Deserialize<'de>>(repo: &Repo, name: &str) -> Result, XrpcError> { +pub(crate) fn load Deserialize<'de>>( + repo: &Repo, + name: &str, +) -> Result, XrpcError> { let name = RefName::new(name).map_err(|error| XrpcError::internal(error.to_string()))?; let Some(oid) = repo.find_ref(&name).map_err(XrpcError::from)? else { return Ok(None); @@ -52,7 +55,11 @@ pub(crate) fn load Deserialize<'de>>(repo: &Repo, name: &str) -> Res .map_err(|error| XrpcError::internal(error.to_string())) } -pub(crate) fn store(repo: &Repo, name: &str, value: Option<&T>) -> Result<(), XrpcError> { +pub(crate) fn store( + repo: &Repo, + name: &str, + value: Option<&T>, +) -> Result<(), XrpcError> { let name = RefName::new(name).map_err(|error| XrpcError::internal(error.to_string()))?; let new = value .map(|value| { diff --git a/knot2/crates/knot-xrpc/src/repos.rs b/knot2/crates/knot-xrpc/src/repos.rs index d11f3d40a..8d6c28818 100644 --- a/knot2/crates/knot-xrpc/src/repos.rs +++ b/knot2/crates/knot-xrpc/src/repos.rs @@ -455,7 +455,11 @@ async fn source_of( .atproto .resolve_identity(&AccountDid::from(did.clone())) .await?; - let mut url = identity.pds.url().clone(); + let mut url = identity + .pds() + .map_err(knot_atproto::AtprotoError::from)? + .url() + .clone(); url.path_segments_mut() .map_err(|()| XrpcError::invalid_request("source repo's knot url can't be a base"))? .pop_if_empty() diff --git a/knot2/crates/knot-xrpc/src/social.rs b/knot2/crates/knot-xrpc/src/social.rs index a7e6a29ec..970243fe6 100644 --- a/knot2/crates/knot-xrpc/src/social.rs +++ b/knot2/crates/knot-xrpc/src/social.rs @@ -8,7 +8,7 @@ use axum::response::{IntoResponse, Response}; use http::HeaderMap; use jacquard_common::types::tid::Ticker; use jacquard_repo::storage::BlockStore; -use knot_acl::{ContributeGrant, KnotAcl, ReadGrant, RepoAccess, Viewer}; +use knot_acl::{ContributeGrant, KnotAcl, Member, ReadGrant, RepoAccess, Viewer}; use knot_cob::{ Approved, ChangePayload, CobError, CobHome, CobStore, Materialized, Record, WriteContext, }; @@ -47,8 +47,6 @@ use knot_types::{ use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; -use knot_atproto::Authenticated; - use crate::error::XrpcError; use crate::query::ValidatedQuery; use crate::{XrpcState, admit_social, decode, project_collaborators, run_blocking}; @@ -2054,7 +2052,7 @@ async fn knot_grant_actor( state .authenticate_write(headers, method) .await - .map(Authenticated::into_did) + .map(Member::into_did) } const NOT_LABEL_AUTHORITY: &str = "label defs belong to the repository's owner alone"; diff --git a/knot2/crates/knot-xrpc/src/tests.rs b/knot2/crates/knot-xrpc/src/tests.rs index 56c85313a..0c287e739 100644 --- a/knot2/crates/knot-xrpc/src/tests.rs +++ b/knot2/crates/knot-xrpc/src/tests.rs @@ -403,6 +403,7 @@ fn state_from( ), )), rkeys: Default::default(), + syncers: Default::default(), }); (state, clock) } diff --git a/knot2/crates/knot-xrpc/tests/common/mod.rs b/knot2/crates/knot-xrpc/tests/common/mod.rs index 6903a93b8..d07f51868 100644 --- a/knot2/crates/knot-xrpc/tests/common/mod.rs +++ b/knot2/crates/knot-xrpc/tests/common/mod.rs @@ -45,7 +45,7 @@ pub async fn serve_local_plc(world: &World) -> std::net::SocketAddr { let endpoint = endpoint.clone(); async move { let did = Did::new_owned(did).expect("the plc request path is a DID"); - let doc = did_doc(&did, &multikey, &endpoint); + let doc = did_doc(&did, &multikey, Some(&endpoint)); let mut headers = axum::http::HeaderMap::new(); headers.insert( axum::http::header::CONTENT_TYPE, @@ -231,7 +231,7 @@ use knot_secrets::{MasterKey, SealedStore}; use knot_types::{ AccountDid, ActorId, AtUri, AuthorName, Did, DidRkey, Email, KnotHostname, KnotId, KnotServiceUrl, Nsid, ObjectFormat, Oid, OwnerDid, RecordAddress, RecordCid, RecordCollection, - RecordRkey, RepoDid, RepoName, RepoRkey, ServiceDid, UnixSeconds, + RecordRkey, RepoDid, RepoName, RepoRkey, ServiceDid, Syncers, UnixSeconds, }; use knot_xrpc::{ ArchiveLimit, Budgets, ByteLimits, CobLocks, GlobalQuota, LimitConfig, PerActorQuota, @@ -264,6 +264,11 @@ pub fn nsid(name: &str) -> Nsid { Nsid::new_owned(name).expect("the xrpc method is an NSID") } +pub fn declared(did: &str, lxm: &[&str]) -> knot_types::Syncer { + let lxm = lxm.iter().map(|method| nsid(method)).collect(); + knot_types::Syncer::new(knot_types::ServiceDid::new(did).unwrap(), lxm).unwrap() +} + pub fn fixture_collection() -> RecordCollection { RecordCollection::new(FIXTURE_COLLECTION).unwrap() } @@ -440,6 +445,22 @@ pub fn fixture_record(created_at: UnixSeconds) -> knot_types::RecordBody { pub type Responder = Box Result + Send + Sync>; +struct Terms { + admins: BTreeSet, + policy: knot_types::ContributionPolicy, + syncers: Syncers, +} + +impl Default for Terms { + fn default() -> Self { + Self { + admins: BTreeSet::new(), + policy: knot_types::ContributionPolicy::Anyone, + syncers: Syncers::default(), + } + } +} + pub struct World { _dir: tempfile::TempDir, // keeps the import runner's shutdown sender alive for as long as the world is @@ -510,9 +531,11 @@ impl World { ByteLimits::default(), ObjectFormat::SHA1, LimitConfig::default(), - BTreeSet::from([AccountDid::new(OWNER).unwrap()]), true, - knot_types::ContributionPolicy::Anyone, + Terms { + admins: BTreeSet::from([AccountDid::new(OWNER).unwrap()]), + ..Terms::default() + }, ) } @@ -522,9 +545,25 @@ impl World { ByteLimits::default(), ObjectFormat::SHA1, LimitConfig::default(), - BTreeSet::new(), true, - policy, + Terms { + policy, + ..Terms::default() + }, + ) + } + + pub fn declaring(syncers: Syncers) -> Self { + Self::assemble( + true, + ByteLimits::default(), + ObjectFormat::SHA1, + LimitConfig::default(), + true, + Terms { + syncers, + ..Terms::default() + }, ) } @@ -538,9 +577,8 @@ impl World { ByteLimits::default(), ObjectFormat::SHA1, LimitConfig::default(), - BTreeSet::new(), false, - knot_types::ContributionPolicy::Anyone, + Terms::default(), ) } @@ -555,9 +593,8 @@ impl World { byte_limits, object_format, limits, - BTreeSet::new(), true, - knot_types::ContributionPolicy::Anyone, + Terms::default(), ) } @@ -566,10 +603,14 @@ impl World { byte_limits: ByteLimits, object_format: ObjectFormat, limits: LimitConfig, - admins: BTreeSet, lfs_enabled: bool, - policy: knot_types::ContributionPolicy, + terms: Terms, ) -> Self { + let Terms { + admins, + policy, + syncers, + } = terms; let dir = tempfile::tempdir().unwrap(); let scan_path = dir.path().join("repos"); std::fs::create_dir_all(&scan_path).unwrap(); @@ -601,12 +642,14 @@ impl World { let body = match did { Some(did) => did_doc_at( &did, - &pds_of + // did:web dids here are services and services have no pds + pds_of .lock() .unwrap() .get(&did) .cloned() - .unwrap_or_else(default_pds), + .or_else(|| (!did.as_str().starts_with("did:web:")).then(default_pds)) + .as_ref(), ), None => Bytes::new(), }; @@ -714,6 +757,7 @@ impl World { ), )), rkeys: Default::default(), + syncers, }); // the runner lives like it does in the server, so wakes from tests are processed tokio::spawn(knot_xrpc::imports::run( @@ -936,7 +980,19 @@ fn actor_key() -> SigningKey { SigningKey::from_bytes(&[9u8; 32].into()).unwrap() } -fn did_doc(did: &Did, multikey: &ActorId, endpoint: &KnotServiceUrl) -> serde_json::Value { +fn pds_service(pds: Option<&KnotServiceUrl>) -> Vec { + pds.map(|pds| { + serde_json::json!({ + "id": "#atproto_pds", + "type": "AtprotoPersonalDataServer", + "serviceEndpoint": pds.as_str() + }) + }) + .into_iter() + .collect() +} + +fn did_doc(did: &Did, multikey: &ActorId, pds: Option<&KnotServiceUrl>) -> serde_json::Value { serde_json::json!({ "id": did.as_str(), "alsoKnownAs": [], @@ -946,22 +1002,18 @@ fn did_doc(did: &Did, multikey: &ActorId, endpoint: &KnotServiceUrl) -> serde_js "controller": did.as_str(), "publicKeyMultibase": multikey.as_str() }], - "service": [{ - "id": "#atproto_pds", - "type": "AtprotoPersonalDataServer", - "serviceEndpoint": endpoint.as_str() - }] + "service": pds_service(pds) }) } -fn did_doc_at(did: &Did, endpoint: &KnotServiceUrl) -> Bytes { +fn did_doc_at(did: &Did, pds: Option<&KnotServiceUrl>) -> Bytes { let sec1 = actor_key() .verifying_key() .to_encoded_point(true) .as_bytes() .to_vec(); let multikey = ActorId::new(knot_types::crypto::multikey(0xe7, &sec1)).unwrap(); - let doc = did_doc(did, &multikey, endpoint); + let doc = did_doc(did, &multikey, pds); Bytes::from(serde_json::to_vec(&doc).unwrap()) } @@ -1047,7 +1099,7 @@ pub fn repo_did_document(world: &World, did: &RepoDid) -> serde_json::Value { did_doc( &Did::new_owned(did.as_str()).unwrap(), &knot_multikey(world), - &world.state.knot_service_url, + Some(&world.state.knot_service_url), ) } diff --git a/knot2/crates/knot-xrpc/tests/syncers.rs b/knot2/crates/knot-xrpc/tests/syncers.rs new file mode 100644 index 000000000..56f9d3117 --- /dev/null +++ b/knot2/crates/knot-xrpc/tests/syncers.rs @@ -0,0 +1,116 @@ +mod common; + +use http::{HeaderMap, HeaderValue, StatusCode, header}; +use knot_types::{AccountDid, RepoDid, RepoName, RepoRkey, Syncers, Visibility}; + +use common::{World, declared, get_with_headers, nsid, owner, post_as, service_jwt}; + +const DESCRIBE: &str = "com.atproto.repo.describeRepo"; +const CREATE: &str = "com.atproto.repo.createRecord"; + +fn bobbin() -> AccountDid { + AccountDid::new("did:web:bobbin.example").unwrap() +} + +fn world_declaring(lxm: &[&str]) -> World { + let bobbin = declared(bobbin().as_str(), lxm); + World::declaring(Syncers::new(vec![bobbin]).unwrap()) +} + +fn repo(world: &World, name: &str, visibility: Visibility) -> RepoDid { + let did = RepoDid::new(format!("did:plc:{name}fixture")).unwrap(); + world.layout.create(&did).unwrap(); + world.register_as( + &did, + &RepoName::new(name).unwrap(), + &RepoRkey::new(name).unwrap(), + visibility, + ); + did +} + +async fn describe_as(world: &World, did: &RepoDid, actor: &AccountDid) -> (StatusCode, String) { + let mut headers = HeaderMap::new(); + let token = service_jwt(&nsid(DESCRIBE), actor); + headers.insert( + header::AUTHORIZATION, + HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), + ); + let path = format!("/xrpc/{DESCRIBE}?repo={did}"); + let (status, _, body) = get_with_headers(world, &path, headers).await; + let body = String::from_utf8_lossy(&body).replace(did.as_str(), ""); + (status, body) +} + +#[tokio::test] +async fn a_declared_syncer_reads_a_public_repo_like_anyone() { + let world = world_declaring(&[DESCRIBE]); + let open = repo(&world, "open", Visibility::Public); + let (status, body) = describe_as(&world, &open, &bobbin()).await; + assert_eq!(status, StatusCode::OK, "{body}"); +} + +#[tokio::test] +async fn a_declared_syncer_gets_a_private_repo_as_a_missing_one() { + let world = world_declaring(&[DESCRIBE]); + let hidden = repo(&world, "hidden", Visibility::Private); + let ghost = RepoDid::new("did:plc:ghostfixture").unwrap(); + + let (member, _) = describe_as(&world, &hidden, &owner()).await; + assert_eq!(member, StatusCode::OK, "the owner reads it, so it is there"); + + let private = describe_as(&world, &hidden, &bobbin()).await; + let missing = describe_as(&world, &ghost, &bobbin()).await; + assert_ne!(private.0, StatusCode::OK); + assert_eq!(private, missing); +} + +#[tokio::test] +async fn a_declared_syncer_is_refused_a_method_outside_its_list() { + let world = world_declaring(&["com.atproto.sync.getRepo"]); + let open = repo(&world, "open", Visibility::Public); + let (status, body) = describe_as(&world, &open, &bobbin()).await; + assert_eq!(status, StatusCode::UNAUTHORIZED, "{body}"); +} + +#[tokio::test] +async fn an_undeclared_service_did_is_a_stranger() { + let world = world_declaring(&[DESCRIBE]); + let open = repo(&world, "open", Visibility::Public); + let hidden = repo(&world, "hidden", Visibility::Private); + let ghost = RepoDid::new("did:plc:ghostfixture").unwrap(); + let deliberi = AccountDid::new("did:web:deliberi.example").unwrap(); + + let (status, body) = describe_as(&world, &open, &deliberi).await; + assert_eq!(status, StatusCode::OK, "{body}"); + let private = describe_as(&world, &hidden, &deliberi).await; + assert_eq!(private, describe_as(&world, &ghost, &deliberi).await); + assert_ne!(private.0, StatusCode::OK); +} + +#[tokio::test] +async fn a_declared_syncer_cant_write_even_when_its_list_names_the_write() { + let world = world_declaring(&[DESCRIBE, CREATE]); + let open = repo(&world, "open", Visibility::Public); + let collection = common::ticket_collection(); + let (status, answer) = post_as( + &world, + &nsid(CREATE), + &bobbin(), + &knot_types::ServiceDid::new(open.as_str()).unwrap(), + serde_json::json!({ + "repo": open, + "collection": &collection, + "record": { + "$type": &collection, + "repo": open, + "title": "let me in", + "body": "", + "state": "open", + "createdAt": "2026-09-11T00:00:00Z" + } + }), + ) + .await; + assert_eq!(status, StatusCode::UNAUTHORIZED, "{answer}"); +} -- 2.54.0