alice.pds.demo.boltless.dev
This repository has no description
78 kB
2102 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod auth;
2mod identity;
3mod jwt;
4mod pointer;
5mod pubkeys;
6mod resolve;
7#[cfg(test)]
8mod test_support;
9
10pub use auth::{OauthAuthorizer, PointerAuth, PointerAuthorizer, ServiceAuth};
11pub use identity::{
12 IdentityError, MintNonce, PreparedRepoDid, knot_did_document, prepare_repo_did,
13};
14pub use jwt::{JwtError, JwtNonce, ServiceJwt};
15pub use pointer::PointerReceipt;
16pub use pubkeys::{KeyParseError, parse_authorized_key};
17pub use resolve::{Identity, PdsEndpoint, PlcDirectory, ResolveError};
18
19#[doc(hidden)]
20pub mod fuzz {
21 pub fn pubkey(data: &[u8]) {
22 let _ = crate::pubkeys::offered_page(data, 100);
23 let _ = crate::parse_authorized_key(&String::from_utf8_lossy(data));
24 }
25
26 pub fn did_document(data: &[u8]) {
27 let did = knot_types::AccountDid::new("did:plc:nel").expect("constant test did is valid");
28 let _ = crate::resolve::identity_from_document(&did, data);
29 let repo_did = knot_types::RepoDid::new("did:plc:nel").expect("constant test did is valid");
30 let _ = crate::resolve::document_publishes_key(
31 &repo_did,
32 data,
33 &knot_runtime::PublicKeyBytes::from_bytes(data.to_vec()),
34 );
35 }
36}
37
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub enum RecordPresence {
40 Present,
41 Absent,
42}
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
45pub enum ReplayGuard {
46 SingleUse,
47 ReusableUntilExpiry,
48}
49
50use std::sync::Arc;
51use std::time::Duration;
52
53use futures::stream::{self, TryStreamExt};
54use http::StatusCode;
55use knot_cache::{
56 Admitted, AsyncCache, EntryCount, Expiring, GroupQuota, MokaFuture, Quotas, Rejected,
57 TotalQuota,
58};
59use knot_runtime::{
60 Clock, DnsTxtResolver, HttpRequest, HttpTransport, NetworkError, PublicKeyBytes, SystemDns,
61 UnixMicros,
62};
63use knot_types::{
64 AccountDid, Collection, Handle, HttpStatus, KnotId, Nsid, OfferedKey, RepoDid, RepoRkey, Rkey,
65 UnixSeconds,
66};
67use pubkeys::Cursor;
68use serde::{Deserialize, Serialize};
69use url::Url;
70
71const DEFAULT_TTL: Duration = Duration::from_secs(300);
72const NEGATIVE_TTL: Duration = Duration::from_secs(30);
73const STALE_TTL: Duration = Duration::from_secs(30);
74const PUBKEY_PAGE_LIMIT: u16 = 100;
75fn repo_collection() -> Nsid {
76 Nsid::new_static("sh.tangled.repo").expect("literal nsid parses")
77}
78const PUBKEY_MAX_PAGES: usize = 8;
79const MAX_IDENTITY_CACHE: usize = 4096;
80const MAX_SEEN_JTI: usize = 8192;
81
82const MAX_JTI_PER_ISSUER: usize = MAX_SEEN_JTI / 16;
83
84#[derive(Debug, thiserror::Error)]
85pub enum AtprotoError {
86 #[error(transparent)]
87 Resolve(#[from] ResolveError),
88 #[error(transparent)]
89 Jwt(#[from] JwtError),
90 #[error("network failure: {0}")]
91 Network(#[from] NetworkError),
92 #[error("listRecords for {did} returned HTTP {status}")]
93 ListRecords { did: AccountDid, status: HttpStatus },
94 #[error("listRecords response wasn't valid JSON: {0}")]
95 MalformedRecords(String),
96 #[error("PDS endpoint {pds:?} isn't a usable base URL")]
97 BadPdsEndpoint { pds: String },
98 #[error("service token {jti:?} has already been presented")]
99 Replay { jti: JwtNonce },
100 #[error("replay-protection store is full and cannot accept another nonce")]
101 ReplayStoreSaturated,
102 #[error("issuer {issuer} has too many live replay nonces")]
103 ReplayShareExhausted { issuer: AccountDid },
104 #[error(transparent)]
105 Identity(#[from] IdentityError),
106 #[error("plc submission for {did} returned HTTP {status}")]
107 PlcSubmit { did: RepoDid, status: HttpStatus },
108 #[error("putRecord for {subject} returned HTTP {status}")]
109 PutRecord {
110 subject: AccountDid,
111 status: HttpStatus,
112 },
113 #[error("getRecord for {owner} returned HTTP {status}")]
114 GetRecord {
115 owner: AccountDid,
116 status: HttpStatus,
117 },
118 #[error("pointer record couldn't be encoded: {0}")]
119 PointerEncode(String),
120 #[error("putRecord response isn't a valid receipt: {0}")]
121 MalformedReceipt(String),
122}
123
124impl AtprotoError {
125 pub fn is_transient(&self) -> bool {
126 match self {
127 AtprotoError::Network(_)
128 | AtprotoError::ReplayStoreSaturated
129 | AtprotoError::ReplayShareExhausted { .. } => true,
130 AtprotoError::Resolve(error) => error.is_transient(),
131 AtprotoError::ListRecords { status, .. }
132 | AtprotoError::PlcSubmit { status, .. }
133 | AtprotoError::PutRecord { status, .. }
134 | AtprotoError::GetRecord { status, .. } => status.is_transient(),
135 _ => false,
136 }
137 }
138}
139
140#[derive(Clone)]
141enum Resolution {
142 Found(Identity),
143 Failed(ResolveError),
144 Transient(ResolveError),
145}
146
147#[derive(Clone)]
148enum HandleResolution {
149 Bound(AccountDid),
150 Unbound(ResolveError),
151 Transient(ResolveError),
152}
153
154#[derive(Clone)]
155struct Cached<R> {
156 resolution: R,
157 expires_at: UnixMicros,
158}
159
160pub struct Atproto<H, C> {
161 http: H,
162 clock: C,
163 knot_did: KnotId,
164 plc_directory: PlcDirectory,
165 dns: Arc<dyn DnsTxtResolver>,
166 identities: MokaFuture<AccountDid, Cached<Resolution>>,
167 handles: MokaFuture<Handle, Cached<HandleResolution>>,
168 seen_jti: Expiring<(AccountDid, JwtNonce), AccountDid, ()>,
169}
170
171impl<H: HttpTransport, C: Clock> Atproto<H, C> {
172 pub fn new(http: H, clock: C, knot_did: KnotId, plc_directory: PlcDirectory) -> Self {
173 Self {
174 http,
175 clock,
176 knot_did,
177 plc_directory,
178 dns: Arc::new(SystemDns::new()),
179 identities: MokaFuture::by_count(EntryCount::new(MAX_IDENTITY_CACHE as u64)),
180 handles: MokaFuture::by_count(EntryCount::new(MAX_IDENTITY_CACHE as u64)),
181 seen_jti: Expiring::new(Quotas {
182 per_group: GroupQuota::new(MAX_JTI_PER_ISSUER),
183 total: TotalQuota::new(MAX_SEEN_JTI),
184 }),
185 }
186 }
187
188 pub fn with_dns(mut self, dns: Arc<dyn DnsTxtResolver>) -> Self {
189 self.dns = dns;
190 self
191 }
192
193 pub fn now(&self) -> UnixMicros {
194 self.clock.now_unix_micros()
195 }
196
197 pub async fn resolve_identity(&self, did: &AccountDid) -> Result<Identity, AtprotoError> {
198 self.resolve_identity_inner(did).await.map_err(Into::into)
199 }
200
201 async fn resolve_identity_inner(&self, did: &AccountDid) -> Result<Identity, ResolveError> {
202 let now = self.clock.now_unix_micros();
203 let filled = self
204 .identities
205 .get_or_fill_if(
206 did.clone(),
207 |cached: &Cached<Resolution>| cached.expires_at <= now,
208 self.fill_identity(did, now),
209 )
210 .await;
211 let fresh = filled.fresh;
212 match filled.value.resolution {
213 Resolution::Found(identity) => Ok(identity),
214 Resolution::Transient(error) => Err(error),
215 Resolution::Failed(error) if fresh => Err(error),
216 Resolution::Failed(_) => Err(ResolveError::RecentlyFailed { did: did.clone() }),
217 }
218 }
219
220 pub async fn resolve_handle_to_did(&self, handle: &Handle) -> Result<AccountDid, AtprotoError> {
221 let now = self.clock.now_unix_micros();
222 let filled = self
223 .handles
224 .get_or_fill_if(
225 handle.clone(),
226 |cached: &Cached<HandleResolution>| cached.expires_at <= now,
227 self.fill_handle(handle, now),
228 )
229 .await;
230 let fresh = filled.fresh;
231 match filled.value.resolution {
232 HandleResolution::Bound(did) => Ok(did),
233 HandleResolution::Transient(error) => Err(error.into()),
234 HandleResolution::Unbound(error) if fresh => Err(error.into()),
235 HandleResolution::Unbound(_) => Err(ResolveError::HandleRecentlyFailed {
236 handle: handle.clone(),
237 }
238 .into()),
239 }
240 }
241
242 async fn stale_handle_did(&self, handle: &Handle) -> Option<AccountDid> {
243 // `fill_handle` calls this from inside its own
244 // `or_insert_with_if` init for this key.
245 // Moka will keep the prior entry readable until init returns,
246 // so this `get` will yield the last resolved DID to re-serve,
247 // when there's a temporary outage.
248 match self.handles.get(handle).await?.resolution {
249 HandleResolution::Bound(did) => Some(did),
250 _ => None,
251 }
252 }
253
254 async fn fill_handle(&self, handle: &Handle, now: UnixMicros) -> Cached<HandleResolution> {
255 match self.verify_handle(handle).await {
256 Ok(did) => Cached {
257 resolution: HandleResolution::Bound(did),
258 expires_at: expires(now, DEFAULT_TTL),
259 },
260 Err(error) if error.is_transient() => match self.stale_handle_did(handle).await {
261 Some(did) => Cached {
262 resolution: HandleResolution::Bound(did),
263 expires_at: expires(now, STALE_TTL),
264 },
265 None => Cached {
266 resolution: HandleResolution::Transient(error),
267 expires_at: now,
268 },
269 },
270 Err(error) => Cached {
271 resolution: HandleResolution::Unbound(error),
272 expires_at: expires(now, NEGATIVE_TTL),
273 },
274 }
275 }
276
277 async fn verify_handle(&self, handle: &Handle) -> Result<AccountDid, ResolveError> {
278 let candidate = match self.dns_txt_did(handle).await {
279 Ok(Some(did)) => did,
280 Ok(None) => self.wellknown_did(handle).await?,
281 Err(dns_error) if dns_error.is_transient() => match self.wellknown_did(handle).await {
282 Ok(did) => did,
283 Err(_) => return Err(dns_error),
284 },
285 Err(dns_error) => return Err(dns_error),
286 };
287 let identity = self.resolve_identity_inner(&candidate).await?;
288 if identity.claims_handle(handle) {
289 Ok(candidate)
290 } else {
291 Err(ResolveError::HandleMismatch {
292 handle: handle.clone(),
293 resolved: candidate,
294 claimed: identity.primary_handle().cloned(),
295 })
296 }
297 }
298
299 async fn dns_txt_did(&self, handle: &Handle) -> Result<Option<AccountDid>, ResolveError> {
300 let records = self
301 .dns
302 .lookup_txt(format!("_atproto.{}", handle.as_str()))
303 .await?;
304 let dids = records
305 .iter()
306 .filter_map(|record| record.trim().strip_prefix("did=").map(str::trim))
307 .map(|value| {
308 AccountDid::new(value).map_err(|_| ResolveError::HandleForwardMalformed {
309 handle: handle.clone(),
310 value: value.to_string(),
311 })
312 })
313 .collect::<Result<Vec<AccountDid>, ResolveError>>()?;
314 let distinct = dids
315 .iter()
316 .map(AccountDid::as_str)
317 .collect::<std::collections::BTreeSet<&str>>()
318 .len();
319 match distinct {
320 0 => Ok(None),
321 1 => Ok(dids.into_iter().next()),
322 _ => Err(ResolveError::HandleAmbiguous {
323 handle: handle.clone(),
324 }),
325 }
326 }
327
328 async fn wellknown_did(&self, handle: &Handle) -> Result<AccountDid, ResolveError> {
329 let url = Url::parse(&format!(
330 "https://{}/.well-known/atproto-did",
331 handle.as_str()
332 ))
333 .map_err(|_| ResolveError::HandleUnresolvable {
334 handle: handle.clone(),
335 })?;
336 resolve::guard_fetch_url(&url)?;
337 let response = self
338 .http
339 .execute(HttpRequest::get(url))
340 .await
341 .map_err(ResolveError::from)?;
342 if !response.status.is_success() {
343 let status = HttpStatus::from(response.status);
344 if status.is_transient() {
345 return Err(ResolveError::Status { status });
346 }
347 return Err(ResolveError::HandleUnresolvable {
348 handle: handle.clone(),
349 });
350 }
351 let value = std::str::from_utf8(&response.body)
352 .map_err(|_| ResolveError::HandleUnresolvable {
353 handle: handle.clone(),
354 })?
355 .trim();
356 AccountDid::new(value).map_err(|_| ResolveError::HandleForwardMalformed {
357 handle: handle.clone(),
358 value: value.to_string(),
359 })
360 }
361
362 async fn fill_identity(&self, did: &AccountDid, now: UnixMicros) -> Cached<Resolution> {
363 match self.fetch_identity(did).await {
364 Ok(identity) => Cached {
365 resolution: Resolution::Found(identity),
366 expires_at: expires(now, DEFAULT_TTL),
367 },
368 Err(error) if warrants_negative_cache(&error) => Cached {
369 resolution: Resolution::Failed(error),
370 expires_at: expires(now, NEGATIVE_TTL),
371 },
372 Err(error) => Cached {
373 resolution: Resolution::Transient(error),
374 expires_at: now,
375 },
376 }
377 }
378
379 async fn fetch_identity(&self, did: &AccountDid) -> Result<Identity, ResolveError> {
380 let url = resolve::document_url(did, &self.plc_directory)?;
381 resolve::guard_fetch_url(&url)?;
382 let response = self
383 .http
384 .execute(HttpRequest::get(url))
385 .await
386 .map_err(ResolveError::from)?;
387 if !response.status.is_success() {
388 return Err(ResolveError::Status {
389 status: HttpStatus::from(response.status),
390 });
391 }
392 resolve::identity_from_document(did, &response.body)
393 }
394
395 pub async fn resolve_pubkeys(&self, did: &AccountDid) -> Result<Vec<OfferedKey>, AtprotoError> {
396 let identity = self.resolve_identity(did).await?;
397 let http = &self.http;
398 let pds = &identity.pds;
399 let pages = stream::try_unfold(Page::First(PUBKEY_MAX_PAGES), move |state| async move {
400 let (cursor, budget) = match state {
401 Page::Done | Page::First(0) | Page::Next(_, 0) => {
402 return Ok::<_, AtprotoError>(None);
403 }
404 Page::First(budget) => (None, budget),
405 Page::Next(cursor, budget) => (Some(cursor), budget),
406 };
407 let url = list_records_url(pds, did, cursor.as_ref())?;
408 resolve::guard_fetch_url(&url)?;
409 let response = http.execute(HttpRequest::get(url)).await?;
410 if !response.status.is_success() {
411 return Err(AtprotoError::ListRecords {
412 did: did.clone(),
413 status: HttpStatus::from(response.status),
414 });
415 }
416 let page = pubkeys::offered_page(&response.body, PUBKEY_PAGE_LIMIT as usize)
417 .map_err(|error| AtprotoError::MalformedRecords(error.to_string()))?;
418 let next = match page.cursor {
419 Some(cursor) => Page::Next(cursor, budget - 1),
420 None => Page::Done,
421 };
422 Ok(Some((page.keys, next)))
423 });
424 pages
425 .try_fold(Vec::new(), |mut acc, keys| async move {
426 acc.extend(keys);
427 Ok(acc)
428 })
429 .await
430 }
431
432 pub async fn repo_record_present(
433 &self,
434 owner: &AccountDid,
435 rkey: &RepoRkey,
436 ) -> Result<RecordPresence, AtprotoError> {
437 let identity = self.resolve_identity(owner).await?;
438 let url = get_record_url(&identity.pds, owner, &repo_collection(), rkey)?;
439 resolve::guard_fetch_url(&url)?;
440 let response = self.http.execute(HttpRequest::get(url)).await?;
441 match response.status {
442 status if status.is_success() => Ok(RecordPresence::Present),
443 StatusCode::NOT_FOUND => Ok(RecordPresence::Absent),
444 StatusCode::BAD_REQUEST if record_not_found(response.body.as_ref()) => {
445 Ok(RecordPresence::Absent)
446 }
447 status => Err(AtprotoError::GetRecord {
448 owner: owner.clone(),
449 status: HttpStatus::from(status),
450 }),
451 }
452 }
453
454 pub async fn verify_service_jwt(
455 &self,
456 token: &ServiceJwt,
457 method: &Nsid,
458 ) -> Result<AccountDid, AtprotoError> {
459 self.verify_service_jwt_guarded(token, method, ReplayGuard::SingleUse)
460 .await
461 }
462
463 pub async fn verify_service_jwt_guarded(
464 &self,
465 token: &ServiceJwt,
466 method: &Nsid,
467 replay: ReplayGuard,
468 ) -> Result<AccountDid, AtprotoError> {
469 let parsed = jwt::parse(token)?;
470 let issuer = jwt::issuer(&parsed)?;
471 let now_micros = self.clock.now_unix_micros();
472 let now = UnixSeconds::new((now_micros.get() / 1_000_000) as i64);
473
474 jwt::check_claims(&parsed, &self.knot_did, method, now)?;
475 let jti = jwt::nonce(&parsed)?;
476
477 let identity = self.resolve_identity(&issuer).await?;
478 jwt::verify_signature(&parsed, &identity.signing_key)?;
479
480 match replay {
481 ReplayGuard::SingleUse => {
482 let exp = UnixSeconds::new(parsed.claims().exp);
483 self.record_jti(&issuer, jti, exp, now_micros)?;
484 }
485 ReplayGuard::ReusableUntilExpiry => drop(jti),
486 }
487 Ok(issuer)
488 }
489
490 pub async fn submit_plc_operation(
491 &self,
492 prepared: &PreparedRepoDid,
493 ) -> Result<(), AtprotoError> {
494 let account = AccountDid::from(prepared.did.clone());
495 let url = resolve::document_url(&account, &self.plc_directory)?;
496 resolve::guard_fetch_url(&url)?;
497 let request = json_post(
498 url,
499 bytes::Bytes::copy_from_slice(prepared.operation_json()),
500 );
501 let response = self.http.execute(request).await?;
502 if response.status.is_success() {
503 Ok(())
504 } else {
505 Err(AtprotoError::PlcSubmit {
506 did: prepared.did.clone(),
507 status: HttpStatus::from(response.status),
508 })
509 }
510 }
511
512 pub async fn publish_pointer<R: Collection + Serialize>(
513 &self,
514 authorizer: &dyn PointerAuthorizer,
515 subject: &AccountDid,
516 rkey: &Rkey,
517 record: &R,
518 ) -> Result<PointerReceipt, AtprotoError> {
519 let identity = self.resolve_identity(subject).await?;
520 let method = pointer::put_record_method();
521 let url = xrpc_url(&identity.pds, &method)?;
522 resolve::guard_fetch_url(&url)?;
523 let audience = pointer::pds_service_did(&identity.pds)?;
524 let now = UnixSeconds::new((self.clock.now_unix_micros().get() / 1_000_000) as i64);
525 let body = pointer::put_record_body(subject, rkey, record)?;
526 let mut request = json_post(url, bytes::Bytes::from(body));
527 authorizer.authorize(
528 &mut request,
529 &PointerAuth {
530 issuer: &self.knot_did,
531 audience: &audience,
532 lxm: &method,
533 now_unix: now,
534 },
535 )?;
536 let response = self.http.execute(request).await?;
537 if !response.status.is_success() {
538 return Err(AtprotoError::PutRecord {
539 subject: subject.clone(),
540 status: HttpStatus::from(response.status),
541 });
542 }
543 pointer::receipt_from_response(&response.body)
544 }
545
546 pub async fn verify_did_web_publishes_key(
547 &self,
548 did: &RepoDid,
549 expected: &PublicKeyBytes,
550 ) -> Result<(), AtprotoError> {
551 let url = resolve::web_document_url_for(did)?;
552 resolve::guard_fetch_url(&url)?;
553 let response = self
554 .http
555 .execute(HttpRequest::get(url))
556 .await
557 .map_err(ResolveError::from)?;
558 if !response.status.is_success() {
559 return Err(ResolveError::Status {
560 status: HttpStatus::from(response.status),
561 }
562 .into());
563 }
564 resolve::document_publishes_key(did, &response.body, expected).map_err(Into::into)
565 }
566
567 fn record_jti(
568 &self,
569 issuer: &AccountDid,
570 jti: JwtNonce,
571 exp: UnixSeconds,
572 now: UnixMicros,
573 ) -> Result<(), AtprotoError> {
574 let horizon = exp.saturating_add_secs(jwt::CLOCK_SKEW_SECS).get().max(0) as u64;
575 let expires_at = UnixMicros::new(horizon.saturating_mul(1_000_000));
576 match self.seen_jti.admit(
577 (issuer.clone(), jti.clone()),
578 issuer.clone(),
579 (),
580 expires_at,
581 now,
582 ) {
583 Ok(Admitted::Inserted) => Ok(()),
584 Ok(Admitted::Occupied(())) => Err(AtprotoError::Replay { jti }),
585 Err(Rejected::Total) => Err(AtprotoError::ReplayStoreSaturated),
586 Err(Rejected::Group) => Err(AtprotoError::ReplayShareExhausted {
587 issuer: issuer.clone(),
588 }),
589 }
590 }
591}
592
593fn expires(now: UnixMicros, ttl: Duration) -> UnixMicros {
594 let micros = u64::try_from(ttl.as_micros()).unwrap_or(u64::MAX);
595 UnixMicros::new(now.get().saturating_add(micros))
596}
597
598fn warrants_negative_cache(error: &ResolveError) -> bool {
599 match error {
600 ResolveError::Status { status } => {
601 (400..500).contains(&status.get()) && status.get() != 429
602 }
603 ResolveError::Malformed(_)
604 | ResolveError::IdMismatch { .. }
605 | ResolveError::BadSigningKey(_)
606 | ResolveError::BadPds { .. } => true,
607 _ => false,
608 }
609}
610
611enum Page {
612 First(usize),
613 Next(Cursor, usize),
614 Done,
615}
616
617fn json_post(url: Url, body: bytes::Bytes) -> HttpRequest {
618 let mut request = HttpRequest::post(url, body);
619 request.headers.insert(
620 http::header::CONTENT_TYPE,
621 http::HeaderValue::from_static("application/json"),
622 );
623 request
624}
625
626fn xrpc_url(pds: &PdsEndpoint, method: &Nsid) -> Result<Url, AtprotoError> {
627 let mut url = pds.url().clone();
628 url.path_segments_mut()
629 .map_err(|_| AtprotoError::BadPdsEndpoint {
630 pds: pds.url().as_str().to_string(),
631 })?
632 .pop_if_empty()
633 .extend(["xrpc", method.as_str()]);
634 Ok(url)
635}
636
637fn list_records_url(
638 pds: &PdsEndpoint,
639 did: &AccountDid,
640 cursor: Option<&Cursor>,
641) -> Result<Url, AtprotoError> {
642 let method: Nsid =
643 Nsid::new_static("com.atproto.repo.listRecords").expect("literal nsid parses");
644 let collection: Nsid = Nsid::new_static("sh.tangled.publicKey").expect("literal nsid parses");
645 let mut url = xrpc_url(pds, &method)?;
646 url.query_pairs_mut()
647 .append_pair("repo", did.as_str())
648 .append_pair("collection", collection.as_str())
649 .append_pair("limit", &PUBKEY_PAGE_LIMIT.to_string());
650 if let Some(cursor) = cursor {
651 url.query_pairs_mut().append_pair("cursor", cursor.as_str());
652 }
653 Ok(url)
654}
655
656#[derive(Deserialize)]
657struct XrpcErrorBody {
658 error: String,
659}
660
661fn record_not_found(body: &[u8]) -> bool {
662 serde_json::from_slice::<XrpcErrorBody>(body)
663 .is_ok_and(|parsed| parsed.error == "RecordNotFound")
664}
665
666fn get_record_url(
667 pds: &PdsEndpoint,
668 owner: &AccountDid,
669 collection: &Nsid,
670 rkey: &RepoRkey,
671) -> Result<Url, AtprotoError> {
672 let method: Nsid = Nsid::new_static("com.atproto.repo.getRecord").expect("literal nsid parses");
673 let mut url = xrpc_url(pds, &method)?;
674 url.query_pairs_mut()
675 .append_pair("repo", owner.as_str())
676 .append_pair("collection", collection.as_str())
677 .append_pair("rkey", rkey.as_str());
678 Ok(url)
679}
680
681#[cfg(test)]
682mod tests {
683 use super::*;
684 use crate::test_support::*;
685 use bytes::Bytes;
686 use futures::StreamExt;
687 use http::StatusCode;
688 use knot_runtime::{DnsTxtResolver, FakeDns, FakeHttp, ManualClock, NetworkError};
689 use std::sync::Arc;
690 use std::sync::atomic::{AtomicUsize, Ordering};
691
692 const POINTER_RKEY: &str = "3jzfcijpj2z2a";
693 const POINTER_CID: &str = "bafyreidfayvfuwqa7qlnopdjiqrxzs6blmoeu4rujcjtnci5beludirz2a";
694
695 fn squid_doc(signing: &k256::ecdsa::SigningKey) -> Bytes {
696 did_doc(DocSpec {
697 id: SQUID,
698 signing,
699 handle: "nel.pet",
700 pds: "https://pds.oyster.cafe",
701 method: MethodKind::Multikey,
702 })
703 }
704
705 fn resolver<T: HttpTransport>(dns: impl DnsTxtResolver, http: T) -> Atproto<T, ManualClock> {
706 Atproto::new(http, clock(), knot_did(KNOT), plc()).with_dns(Arc::new(dns))
707 }
708
709 #[tokio::test]
710 async fn an_identity_is_resolved_from_a_did_document() {
711 let signing = signer(9);
712 let http = FakeHttp::new(move |request| {
713 assert_eq!(request.url.as_str(), "https://plc.directory/did:plc:squid");
714 Ok(ok(squid_doc(&signing)))
715 });
716 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
717 let identity = atproto.resolve_identity(&did(SQUID)).await.unwrap();
718 assert_eq!(identity.pds.url().as_str(), "https://pds.oyster.cafe/");
719 assert_eq!(identity.primary_handle().unwrap().as_str(), "nel.pet");
720 }
721
722 #[tokio::test]
723 async fn a_second_resolution_is_served_from_cache() {
724 let signing = signer(9);
725 let hits = Arc::new(AtomicUsize::new(0));
726 let counter = hits.clone();
727 let http = FakeHttp::new(move |_| {
728 counter.fetch_add(1, Ordering::SeqCst);
729 Ok(ok(squid_doc(&signing)))
730 });
731 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
732 atproto.resolve_identity(&did(SQUID)).await.unwrap();
733 atproto.resolve_identity(&did(SQUID)).await.unwrap();
734 assert_eq!(hits.load(Ordering::SeqCst), 1);
735 }
736
737 #[tokio::test]
738 async fn an_expired_cache_entry_is_refetched() {
739 let signing = signer(9);
740 let hits = Arc::new(AtomicUsize::new(0));
741 let counter = hits.clone();
742 let http = FakeHttp::new(move |_| {
743 counter.fetch_add(1, Ordering::SeqCst);
744 Ok(ok(squid_doc(&signing)))
745 });
746 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
747 atproto.resolve_identity(&did(SQUID)).await.unwrap();
748 atproto
749 .clock
750 .advance(DEFAULT_TTL + Duration::from_micros(1));
751 atproto.resolve_identity(&did(SQUID)).await.unwrap();
752 assert_eq!(hits.load(Ordering::SeqCst), 2);
753 }
754
755 #[tokio::test]
756 async fn a_handle_resolves_bidirectionally_via_dns_and_is_cached() {
757 let signing = signer(9);
758 let dns_hits = Arc::new(AtomicUsize::new(0));
759 let counter = dns_hits.clone();
760 let dns = FakeDns::new(move |name: &str| {
761 assert_eq!(name, "_atproto.nel.pet");
762 counter.fetch_add(1, Ordering::SeqCst);
763 Ok(vec!["did=did:plc:squid".to_string()])
764 });
765 let http = FakeHttp::new(move |request| {
766 assert_eq!(request.url.as_str(), "https://plc.directory/did:plc:squid");
767 let mut doc: serde_json::Value = serde_json::from_slice(&squid_doc(&signing)).unwrap();
768 doc["alsoKnownAs"] = serde_json::json!(["at://olaren.dev", "at://nel.pet"]);
769 Ok(ok(Bytes::from(serde_json::to_vec(&doc).unwrap())))
770 });
771 let atproto = resolver(dns, http);
772 let owner = handle("nel.pet");
773 assert_eq!(
774 atproto
775 .resolve_handle_to_did(&owner)
776 .await
777 .unwrap()
778 .as_str(),
779 "did:plc:squid",
780 "a handle listed anywhere in alsoKnownAs must resolve, even when it isn't first"
781 );
782 assert_eq!(
783 atproto
784 .resolve_handle_to_did(&owner)
785 .await
786 .unwrap()
787 .as_str(),
788 "did:plc:squid"
789 );
790 assert_eq!(
791 dns_hits.load(Ordering::SeqCst),
792 1,
793 "a resolved handle is served from cache"
794 );
795 }
796
797 #[tokio::test]
798 async fn a_handle_falls_back_to_well_known_when_dns_is_empty_or_transient() {
799 let well_known = || {
800 let signing = signer(9);
801 FakeHttp::new(move |request| match request.url.as_str() {
802 "https://nel.pet/.well-known/atproto-did" => {
803 Ok(ok(Bytes::from_static(b"did:plc:squid\n")))
804 }
805 "https://plc.directory/did:plc:squid" => Ok(ok(squid_doc(&signing))),
806 other => panic!("unexpected url {other}"),
807 })
808 };
809 let empty = resolver(FakeDns::new(|_| Ok(Vec::new())), well_known());
810 let transient = resolver(
811 FakeDns::new(|_| Err(NetworkError::Request("dns unreachable".to_string()))),
812 well_known(),
813 );
814 for atproto in [empty, transient] {
815 assert_eq!(
816 atproto
817 .resolve_handle_to_did(&handle("nel.pet"))
818 .await
819 .unwrap()
820 .as_str(),
821 "did:plc:squid"
822 );
823 }
824 }
825
826 #[tokio::test]
827 async fn a_handle_is_rejected_when_ambiguous_or_disowned() {
828 let ambiguous = resolver(
829 FakeDns::new(|_| {
830 Ok(vec![
831 "did=did:plc:squid".to_string(),
832 "did=did:plc:limpet".to_string(),
833 ])
834 }),
835 FakeHttp::new(|_| panic!("resolution must stop before any fetch")),
836 );
837 assert!(matches!(
838 ambiguous
839 .resolve_handle_to_did(&handle("nel.pet"))
840 .await
841 .unwrap_err(),
842 AtprotoError::Resolve(ResolveError::HandleAmbiguous { .. })
843 ));
844
845 let signing = signer(9);
846 let disowned = resolver(
847 FakeDns::new(|_| Ok(vec!["did=did:plc:squid".to_string()])),
848 FakeHttp::new(move |_| Ok(ok(squid_doc(&signing)))),
849 );
850 assert!(matches!(
851 disowned
852 .resolve_handle_to_did(&handle("olaren.dev"))
853 .await
854 .unwrap_err(),
855 AtprotoError::Resolve(ResolveError::HandleMismatch { .. })
856 ));
857 }
858
859 #[tokio::test]
860 async fn handle_failures_cache_by_class() {
861 let owner = handle("nel.pet");
862
863 let hits = Arc::new(AtomicUsize::new(0));
864 let counter = hits.clone();
865 let unresolvable = resolver(
866 FakeDns::new(move |_| {
867 counter.fetch_add(1, Ordering::SeqCst);
868 Ok(Vec::new())
869 }),
870 FakeHttp::new(|_| Ok(status(StatusCode::NOT_FOUND, Bytes::new()))),
871 );
872 assert!(matches!(
873 unresolvable
874 .resolve_handle_to_did(&owner)
875 .await
876 .unwrap_err(),
877 AtprotoError::Resolve(ResolveError::HandleUnresolvable { .. })
878 ));
879 assert!(matches!(
880 unresolvable
881 .resolve_handle_to_did(&owner)
882 .await
883 .unwrap_err(),
884 AtprotoError::Resolve(ResolveError::HandleRecentlyFailed { .. })
885 ));
886 assert_eq!(
887 hits.load(Ordering::SeqCst),
888 1,
889 "an unresolvable handle is resolved once then served from the negative cache"
890 );
891
892 let hits = Arc::new(AtomicUsize::new(0));
893 let counter = hits.clone();
894 let transient = resolver(
895 FakeDns::new(move |_| {
896 counter.fetch_add(1, Ordering::SeqCst);
897 Ok(Vec::new())
898 }),
899 FakeHttp::new(|_| Ok(status(StatusCode::SERVICE_UNAVAILABLE, Bytes::new()))),
900 );
901 transient.resolve_handle_to_did(&owner).await.unwrap_err();
902 transient.resolve_handle_to_did(&owner).await.unwrap_err();
903 assert_eq!(
904 hits.load(Ordering::SeqCst),
905 2,
906 "a transient well-known status mustn't be negatively cached"
907 );
908 }
909
910 #[tokio::test]
911 async fn a_transient_outage_serves_the_last_good_did() {
912 let signing = signer(9);
913 let outage = Arc::new(AtomicUsize::new(0));
914 let switch = outage.clone();
915 let dns = FakeDns::new(move |_| match switch.load(Ordering::SeqCst) {
916 0 => Ok(vec!["did=did:plc:squid".to_string()]),
917 _ => Err(NetworkError::Request("dns unreachable".to_string())),
918 });
919 let atproto = resolver(dns, FakeHttp::new(move |_| Ok(ok(squid_doc(&signing)))));
920 let owner = handle("nel.pet");
921 assert_eq!(
922 atproto
923 .resolve_handle_to_did(&owner)
924 .await
925 .unwrap()
926 .as_str(),
927 "did:plc:squid"
928 );
929
930 atproto.clock.advance(DEFAULT_TTL + Duration::from_secs(1));
931 outage.store(1, Ordering::SeqCst);
932 assert_eq!(
933 atproto
934 .resolve_handle_to_did(&owner)
935 .await
936 .unwrap()
937 .as_str(),
938 "did:plc:squid",
939 "a transient outage must serve the last resolved DID"
940 );
941
942 atproto.clock.advance(STALE_TTL + Duration::from_secs(1));
943 outage.store(0, Ordering::SeqCst);
944 assert_eq!(
945 atproto
946 .resolve_handle_to_did(&owner)
947 .await
948 .unwrap()
949 .as_str(),
950 "did:plc:squid"
951 );
952 }
953
954 #[tokio::test]
955 async fn a_404_identity_is_negatively_cached() {
956 let hits = Arc::new(AtomicUsize::new(0));
957 let counter = hits.clone();
958 let http = FakeHttp::new(move |_| {
959 counter.fetch_add(1, Ordering::SeqCst);
960 Ok(status(StatusCode::NOT_FOUND, Bytes::new()))
961 });
962 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
963 let first = atproto.resolve_identity(&did(SQUID)).await.unwrap_err();
964 assert!(
965 matches!(first, AtprotoError::Resolve(ResolveError::Status { status }) if status.get() == 404),
966 "got {first:?}"
967 );
968 let second = atproto.resolve_identity(&did(SQUID)).await.unwrap_err();
969 assert!(
970 matches!(
971 second,
972 AtprotoError::Resolve(ResolveError::RecentlyFailed { .. })
973 ),
974 "got {second:?}"
975 );
976 assert_eq!(
977 hits.load(Ordering::SeqCst),
978 1,
979 "404 is served from the negative cache"
980 );
981 }
982
983 #[tokio::test]
984 async fn a_transient_5xx_identity_is_not_cached() {
985 let hits = Arc::new(AtomicUsize::new(0));
986 let counter = hits.clone();
987 let http = FakeHttp::new(move |_| {
988 counter.fetch_add(1, Ordering::SeqCst);
989 Ok(status(StatusCode::SERVICE_UNAVAILABLE, Bytes::new()))
990 });
991 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
992 let _ = atproto.resolve_identity(&did(SQUID)).await.unwrap_err();
993 let _ = atproto.resolve_identity(&did(SQUID)).await.unwrap_err();
994 assert_eq!(
995 hits.load(Ordering::SeqCst),
996 2,
997 "a transient 503 is re-fetched instead of negatively cached"
998 );
999 }
1000
1001 #[tokio::test]
1002 async fn a_document_describing_another_did_is_rejected() {
1003 let doc_key = signer(7);
1004 let http = FakeHttp::new(move |_| {
1005 Ok(ok(did_doc(DocSpec {
1006 id: LIMPET,
1007 signing: &doc_key,
1008 handle: "nel.pet",
1009 pds: "https://pds.oyster.cafe",
1010 method: MethodKind::Multikey,
1011 })))
1012 });
1013 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1014 let error = atproto.resolve_identity(&did(SQUID)).await.unwrap_err();
1015 assert!(
1016 matches!(
1017 error,
1018 AtprotoError::Resolve(ResolveError::IdMismatch { .. })
1019 ),
1020 "got {error:?}"
1021 );
1022 }
1023
1024 #[tokio::test]
1025 async fn a_claimed_handle_is_returned_unverified() {
1026 let signing = signer(9);
1027 let http = FakeHttp::new(move |_| {
1028 Ok(ok(did_doc(DocSpec {
1029 id: SQUID,
1030 signing: &signing,
1031 handle: "olaren.dev",
1032 pds: "https://pds.oyster.cafe",
1033 method: MethodKind::Multikey,
1034 })))
1035 });
1036 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1037 let identity = atproto.resolve_identity(&did(SQUID)).await.unwrap();
1038 assert_eq!(
1039 identity.primary_handle().unwrap().as_str(),
1040 "olaren.dev",
1041 "alsoKnownAs handle is taken at face value with no bidirectional verification"
1042 );
1043 }
1044
1045 #[tokio::test]
1046 async fn an_internal_ip_with_a_port_is_refused() {
1047 let (sink, urls) = recorder();
1048 let http = FakeHttp::new(move |request| {
1049 sink.lock().unwrap().push(request.url.clone());
1050 Ok(ok(Bytes::new()))
1051 });
1052 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1053 let error = atproto
1054 .resolve_identity(&did("did:web:169.254.169.254%3A6379"))
1055 .await
1056 .unwrap_err();
1057 assert!(
1058 matches!(
1059 error,
1060 AtprotoError::Resolve(ResolveError::BlockedHost { .. })
1061 ),
1062 "got {error:?}"
1063 );
1064 assert!(urls.lock().unwrap().is_empty());
1065 }
1066
1067 #[tokio::test]
1068 async fn repo_record_presence_maps_pds_status_to_a_verdict() {
1069 let signing = signer(9);
1070 let http = FakeHttp::new(move |request| {
1071 if request.url.path().ends_with("did.json")
1072 || request.url.host_str() == Some("plc.directory")
1073 {
1074 return Ok(ok(squid_doc(&signing)));
1075 }
1076 assert!(request.url.path().ends_with("com.atproto.repo.getRecord"));
1077 assert!(request.url.query().unwrap().contains("sh.tangled.repo"));
1078 let rkey = request
1079 .url
1080 .query_pairs()
1081 .find(|(key, _)| key == "rkey")
1082 .map(|(_, value)| value.into_owned())
1083 .unwrap_or_default();
1084 let (st, body) = match rkey.as_str() {
1085 "present" => (StatusCode::OK, Bytes::new()),
1086 "missing" => (
1087 StatusCode::BAD_REQUEST,
1088 Bytes::from_static(b"{\"error\":\"RecordNotFound\"}"),
1089 ),
1090 _ => (StatusCode::INTERNAL_SERVER_ERROR, Bytes::new()),
1091 };
1092 Ok(status(st, body))
1093 });
1094 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1095 let owner = did(SQUID);
1096 assert_eq!(
1097 atproto
1098 .repo_record_present(&owner, &RepoRkey::new("present").unwrap())
1099 .await
1100 .unwrap(),
1101 RecordPresence::Present
1102 );
1103 assert_eq!(
1104 atproto
1105 .repo_record_present(&owner, &RepoRkey::new("missing").unwrap())
1106 .await
1107 .unwrap(),
1108 RecordPresence::Absent
1109 );
1110 assert!(
1111 atproto
1112 .repo_record_present(&owner, &RepoRkey::new("boom").unwrap())
1113 .await
1114 .is_err(),
1115 "5xx from the PDS surfaces as an error so the caller can fall back to best-effort"
1116 );
1117 }
1118
1119 #[tokio::test]
1120 async fn pubkeys_are_fetched_from_the_resolved_pds() {
1121 let signing = signer(9);
1122 let line = ssh_line("ssh-ed25519", &[4u8; 32], "nel@oyster.cafe");
1123 let expected = parse_authorized_key(&line).unwrap();
1124 let body = list_body(&[line], None);
1125 let http = FakeHttp::new(move |request| {
1126 if request.url.path().ends_with("did.json")
1127 || request.url.host_str() == Some("plc.directory")
1128 {
1129 Ok(ok(squid_doc(&signing)))
1130 } else {
1131 assert_eq!(request.url.host_str(), Some("pds.oyster.cafe"));
1132 assert!(request.url.path().ends_with("com.atproto.repo.listRecords"));
1133 assert!(
1134 request
1135 .url
1136 .query()
1137 .unwrap()
1138 .contains("sh.tangled.publicKey")
1139 );
1140 Ok(ok(body.clone()))
1141 }
1142 });
1143 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1144 let keys = atproto.resolve_pubkeys(&did(SQUID)).await.unwrap();
1145 assert_eq!(keys, vec![expected]);
1146 }
1147
1148 #[tokio::test]
1149 async fn pubkey_resolution_follows_the_cursor() {
1150 let doc_key = signer(9);
1151 let list_hits = Arc::new(AtomicUsize::new(0));
1152 let counter = list_hits.clone();
1153 let page_one = list_body(
1154 &[ssh_line("ssh-ed25519", &[1u8; 32], "one")],
1155 Some("page-2"),
1156 );
1157 let page_two = list_body(&[ssh_line("ssh-ed25519", &[2u8; 32], "two")], None);
1158 let http = FakeHttp::new(move |request| {
1159 if request.url.host_str() == Some("plc.directory") {
1160 return Ok(ok(squid_doc(&doc_key)));
1161 }
1162 counter.fetch_add(1, Ordering::SeqCst);
1163 let on_second = request.url.query().unwrap().contains("cursor=page-2");
1164 Ok(ok(if on_second {
1165 page_two.clone()
1166 } else {
1167 page_one.clone()
1168 }))
1169 });
1170 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1171 let keys = atproto.resolve_pubkeys(&did(SQUID)).await.unwrap();
1172 assert_eq!(keys.len(), 2);
1173 assert_eq!(list_hits.load(Ordering::SeqCst), 2);
1174 }
1175
1176 #[tokio::test]
1177 async fn a_plain_http_pds_is_refused() {
1178 let doc_key = signer(9);
1179 let (sink, urls) = recorder();
1180 let http = FakeHttp::new(move |request| {
1181 sink.lock().unwrap().push(request.url.clone());
1182 if request.url.host_str() == Some("plc.directory") {
1183 Ok(ok(did_doc(DocSpec {
1184 id: SQUID,
1185 signing: &doc_key,
1186 handle: "nel.pet",
1187 pds: "http://127.0.0.1:6379",
1188 method: MethodKind::Multikey,
1189 })))
1190 } else {
1191 Ok(ok(list_body(&[], None)))
1192 }
1193 });
1194 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1195 let error = atproto.resolve_pubkeys(&did(SQUID)).await.unwrap_err();
1196 assert!(
1197 matches!(
1198 error,
1199 AtprotoError::Resolve(ResolveError::InsecureScheme { .. })
1200 ),
1201 "got {error:?}"
1202 );
1203 assert_eq!(urls.lock().unwrap().len(), 1);
1204 }
1205
1206 #[tokio::test]
1207 async fn a_single_page_is_bounded_by_the_pubkey_page_limit() {
1208 let signing = signer(9);
1209 let lines: Vec<String> = (0..3_000u32)
1210 .map(|seed| {
1211 let mut material = [0u8; 32];
1212 material[..4].copy_from_slice(&seed.to_be_bytes());
1213 ssh_line("ssh-ed25519", &material, "k")
1214 })
1215 .collect();
1216 let page = list_body(&lines, None);
1217 let http = FakeHttp::new(move |request| {
1218 if request.url.host_str() == Some("plc.directory") {
1219 Ok(ok(squid_doc(&signing)))
1220 } else {
1221 Ok(ok(page.clone()))
1222 }
1223 });
1224 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1225 let keys = atproto.resolve_pubkeys(&did(SQUID)).await.unwrap();
1226 assert_eq!(
1227 keys.len(),
1228 PUBKEY_PAGE_LIMIT as usize,
1229 "single page yields at most the page limit even when the PDS floods it"
1230 );
1231 }
1232
1233 #[tokio::test]
1234 async fn a_service_jwt_authenticates_against_the_resolved_issuer_key() {
1235 let signing = signer(9);
1236 let doc = signing.clone();
1237 let http = FakeHttp::new(move |_| Ok(ok(squid_doc(&doc))));
1238 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1239 let method = member_method();
1240 let claims = serde_json::json!({
1241 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1242 "jti": "nonce-authenticates", "lxm": METHOD,
1243 });
1244 let token = mint(&signing, &claims);
1245 let authed = atproto.verify_service_jwt(&token, &method).await.unwrap();
1246 assert_eq!(authed, did(SQUID));
1247 }
1248
1249 #[tokio::test]
1250 async fn a_service_jwt_signed_by_an_impostor_is_rejected() {
1251 let real = signer(9);
1252 let impostor = signer(3);
1253 let http = FakeHttp::new(move |_| Ok(ok(squid_doc(&real))));
1254 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1255 let method = member_method();
1256 let claims = serde_json::json!({
1257 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1258 "jti": "nonce-impostor", "lxm": METHOD,
1259 });
1260 let token = mint(&impostor, &claims);
1261 let error = atproto
1262 .verify_service_jwt(&token, &method)
1263 .await
1264 .unwrap_err();
1265 assert!(matches!(
1266 error,
1267 AtprotoError::Jwt(JwtError::InvalidSignature)
1268 ));
1269 }
1270
1271 #[tokio::test]
1272 async fn a_replayed_token_is_rejected() {
1273 let key = signer(9);
1274 let http = FakeHttp::new(move |_| Ok(ok(squid_doc(&key))));
1275 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1276 let method = member_method();
1277 let claims = serde_json::json!({
1278 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1279 "jti": "single-use-nonce", "lxm": METHOD,
1280 });
1281 let token = mint(&signer(9), &claims);
1282 assert_eq!(
1283 atproto.verify_service_jwt(&token, &method).await.unwrap(),
1284 did(SQUID)
1285 );
1286 let replay = atproto
1287 .verify_service_jwt(&token, &method)
1288 .await
1289 .unwrap_err();
1290 assert!(
1291 matches!(replay, AtprotoError::Replay { .. }),
1292 "got {replay:?}"
1293 );
1294 }
1295
1296 #[tokio::test]
1297 async fn two_issuers_may_share_a_nonce() {
1298 let squid_key = signer(1);
1299 let limpet_key = signer(2);
1300 let sq = squid_key.clone();
1301 let li = limpet_key.clone();
1302 let http = FakeHttp::new(move |request| {
1303 if request.url.as_str().ends_with(LIMPET) {
1304 Ok(ok(did_doc(DocSpec {
1305 id: LIMPET,
1306 signing: &li,
1307 handle: "nel.pet",
1308 pds: "https://pds.oyster.cafe",
1309 method: MethodKind::Multikey,
1310 })))
1311 } else {
1312 Ok(ok(squid_doc(&sq)))
1313 }
1314 });
1315 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1316 let method = member_method();
1317 let squid = serde_json::json!({
1318 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1319 "jti": "shared-nonce", "lxm": METHOD,
1320 });
1321 let limpet = serde_json::json!({
1322 "iss": LIMPET, "aud": KNOT, "exp": 1_001, "iat": 999,
1323 "jti": "shared-nonce", "lxm": METHOD,
1324 });
1325 assert_eq!(
1326 atproto
1327 .verify_service_jwt(&mint(&squid_key, &squid), &method)
1328 .await
1329 .unwrap(),
1330 did(SQUID)
1331 );
1332 assert_eq!(
1333 atproto
1334 .verify_service_jwt(&mint(&limpet_key, &limpet), &method)
1335 .await
1336 .unwrap(),
1337 did(LIMPET)
1338 );
1339 }
1340
1341 #[test]
1342 fn knot_id_lowercases_its_host() {
1343 assert_eq!(knot_did("did:web:NEL.PET").as_str(), "did:web:nel.pet");
1344 }
1345
1346 #[test]
1347 fn list_records_url_is_well_formed() {
1348 let url = list_records_url(
1349 &PdsEndpoint::new(Url::parse("https://pds.oyster.cafe").unwrap()).unwrap(),
1350 &did(SQUID),
1351 None,
1352 )
1353 .unwrap();
1354 assert_eq!(url.host_str(), Some("pds.oyster.cafe"));
1355 assert_eq!(url.path(), "/xrpc/com.atproto.repo.listRecords");
1356 let query = url.query().unwrap();
1357 assert!(query.contains("repo=did%3Aplc%3Asquid"));
1358 assert!(query.contains("collection=sh.tangled.publicKey"));
1359 }
1360
1361 #[test]
1362 fn list_records_url_preserves_a_pds_base_path() {
1363 let url = list_records_url(
1364 &PdsEndpoint::new(Url::parse("https://shared.host/account-pds").unwrap()).unwrap(),
1365 &did(SQUID),
1366 Some(&Cursor::new("page2")),
1367 )
1368 .unwrap();
1369 assert_eq!(url.path(), "/account-pds/xrpc/com.atproto.repo.listRecords");
1370 assert!(url.query().unwrap().contains("cursor=page2"));
1371 }
1372
1373 struct JwtCase {
1374 name: &'static str,
1375 header: &'static [u8],
1376 mutate: fn(&mut serde_json::Value),
1377 expect: fn(&Result<AccountDid, AtprotoError>) -> bool,
1378 zero_network: bool,
1379 }
1380
1381 const JWT_HEADER: &[u8] = br#"{"alg":"ES256K","typ":"JWT"}"#;
1382
1383 const JWT_CASES: &[JwtCase] = &[
1384 JwtCase {
1385 name: "internal-ip issuer is blocked before any fetch",
1386 header: JWT_HEADER,
1387 mutate: |c| c["iss"] = serde_json::json!("did:web:169.254.169.254"),
1388 expect: |r| {
1389 matches!(
1390 r,
1391 Err(AtprotoError::Resolve(ResolveError::BlockedHost { .. }))
1392 )
1393 },
1394 zero_network: true,
1395 },
1396 JwtCase {
1397 name: "decade-long token exceeds the lifetime limit",
1398 header: JWT_HEADER,
1399 mutate: |c| {
1400 c["exp"] = serde_json::json!(1000 + 315_360_000i64);
1401 c["iat"] = serde_json::json!(1000);
1402 },
1403 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::LifetimeTooLong { .. }))),
1404 zero_network: true,
1405 },
1406 JwtCase {
1407 name: "audience in a different case is accepted",
1408 header: JWT_HEADER,
1409 mutate: |c| c["aud"] = serde_json::json!("did:web:NEL.PET"),
1410 expect: |r| r.is_ok(),
1411 zero_network: false,
1412 },
1413 JwtCase {
1414 name: "legacy-typed issuer key verifies",
1415 header: JWT_HEADER,
1416 mutate: |c| c["iss"] = serde_json::json!(LIMPET),
1417 expect: |r| matches!(r, Ok(did) if did.as_str() == LIMPET),
1418 zero_network: false,
1419 },
1420 JwtCase {
1421 name: "token addressed to another knot is refused",
1422 header: JWT_HEADER,
1423 mutate: |c| c["aud"] = serde_json::json!("did:web:somewhere.else"),
1424 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::AudienceMismatch { .. }))),
1425 zero_network: true,
1426 },
1427 JwtCase {
1428 name: "stale token is expired before resolution",
1429 header: JWT_HEADER,
1430 mutate: |c| {
1431 c["exp"] = serde_json::json!(1);
1432 c["iat"] = serde_json::json!(0);
1433 },
1434 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::Expired { .. }))),
1435 zero_network: true,
1436 },
1437 JwtCase {
1438 name: "nonceless token is refused before resolution",
1439 header: JWT_HEADER,
1440 mutate: |c| {
1441 c.as_object_mut().unwrap().remove("jti");
1442 },
1443 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::MissingNonce))),
1444 zero_network: true,
1445 },
1446 JwtCase {
1447 name: "alg none is refused",
1448 header: br#"{"alg":"none","typ":"JWT"}"#,
1449 mutate: |_| {},
1450 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::Parse(_)))),
1451 zero_network: false,
1452 },
1453 JwtCase {
1454 name: "es256 header against a k256 doc key is refused",
1455 header: br#"{"alg":"ES256","typ":"JWT"}"#,
1456 mutate: |_| {},
1457 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::Parse(_)))),
1458 zero_network: false,
1459 },
1460 JwtCase {
1461 name: "token type that isn't JWT is refused",
1462 header: br#"{"alg":"ES256K","typ":"secevent+jwt"}"#,
1463 mutate: |_| {},
1464 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::UnexpectedType { .. }))),
1465 zero_network: true,
1466 },
1467 JwtCase {
1468 name: "method mismatch is refused before resolution",
1469 header: JWT_HEADER,
1470 mutate: |c| c["lxm"] = serde_json::json!("sh.tangled.repo.delete"),
1471 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::MethodMismatch { .. }))),
1472 zero_network: true,
1473 },
1474 JwtCase {
1475 name: "oversized nonce is refused before resolution",
1476 header: JWT_HEADER,
1477 mutate: |c| c["jti"] = serde_json::json!("n".repeat(100_000)),
1478 expect: |r| matches!(r, Err(AtprotoError::Jwt(JwtError::OversizedNonce { .. }))),
1479 zero_network: true,
1480 },
1481 ];
1482
1483 #[tokio::test]
1484 async fn verify_service_jwt_rejects_every_malformed_or_adversarial_token() {
1485 let method = member_method();
1486 stream::iter(JWT_CASES)
1487 .for_each(|case| {
1488 let method = &method;
1489 async move {
1490 let doc_key = signer(1);
1491 let calls = Arc::new(AtomicUsize::new(0));
1492 let counter = calls.clone();
1493 let http = FakeHttp::new(move |request| {
1494 counter.fetch_add(1, Ordering::SeqCst);
1495 if request.url.as_str().ends_with(LIMPET) {
1496 Ok(ok(did_doc(DocSpec {
1497 id: LIMPET,
1498 signing: &doc_key,
1499 handle: "nel.pet",
1500 pds: "https://pds.oyster.cafe",
1501 method: MethodKind::LegacyK256,
1502 })))
1503 } else {
1504 Ok(ok(squid_doc(&doc_key)))
1505 }
1506 });
1507 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1508 let mut claims = serde_json::json!({
1509 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1510 "jti": "nonce-3a", "lxm": METHOD,
1511 });
1512 (case.mutate)(&mut claims);
1513 let token = mint_with_header(&signer(1), case.header, &claims);
1514 let result = atproto.verify_service_jwt(&token, method).await;
1515 assert!(
1516 (case.expect)(&result),
1517 "case {:?} got {result:?}",
1518 case.name
1519 );
1520 if case.zero_network {
1521 assert_eq!(
1522 calls.load(Ordering::SeqCst),
1523 0,
1524 "case {:?} must be refused before any network resolution",
1525 case.name
1526 );
1527 }
1528 }
1529 })
1530 .await;
1531 }
1532
1533 #[tokio::test]
1534 async fn the_jti_replay_store_is_bounded_and_fails_closed_when_saturated() {
1535 let signing = signer(9);
1536 let doc = signing.clone();
1537 let http = FakeHttp::new(move |request| {
1538 let host = request.url.host_str().unwrap().to_string();
1539 let id = format!("did:web:{host}");
1540 Ok(ok(did_doc(DocSpec {
1541 id: &id,
1542 signing: &doc,
1543 handle: "nel.pet",
1544 pds: "https://pds.oyster.cafe",
1545 method: MethodKind::Multikey,
1546 })))
1547 });
1548 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1549 let method = member_method();
1550 let issuers = MAX_SEEN_JTI / MAX_JTI_PER_ISSUER;
1551 stream::iter(
1552 (0..issuers).flat_map(|issuer| (0..MAX_JTI_PER_ISSUER).map(move |i| (issuer, i))),
1553 )
1554 .for_each(|(issuer, i)| {
1555 let atproto = &atproto;
1556 let method = &method;
1557 let signing = &signing;
1558 async move {
1559 let claims = serde_json::json!({
1560 "iss": format!("did:web:i{issuer}.oyster.cafe"), "aud": KNOT,
1561 "exp": 1_001, "iat": 999,
1562 "jti": format!("nonce-{issuer}-{i}"), "lxm": METHOD,
1563 });
1564 atproto
1565 .verify_service_jwt(&mint(signing, &claims), method)
1566 .await
1567 .unwrap();
1568 }
1569 })
1570 .await;
1571 assert_eq!(atproto.seen_jti.len(), MAX_SEEN_JTI);
1572 let overflow = serde_json::json!({
1573 "iss": "did:web:fresh.oyster.cafe", "aud": KNOT, "exp": 1_001, "iat": 999,
1574 "jti": "nonce-overflow", "lxm": METHOD,
1575 });
1576 let error = atproto
1577 .verify_service_jwt(&mint(&signing, &overflow), &method)
1578 .await
1579 .unwrap_err();
1580 assert!(
1581 matches!(error, AtprotoError::ReplayStoreSaturated),
1582 "every verify past the global limit fails closed, got {error:?}"
1583 );
1584 assert!(
1585 atproto.seen_jti.len() <= MAX_SEEN_JTI,
1586 "replay store must stay bounded, held {}",
1587 atproto.seen_jti.len()
1588 );
1589 }
1590
1591 #[tokio::test]
1592 async fn one_issuer_cannot_hog_the_replay_store() {
1593 let signing = signer(9);
1594 let doc = signing.clone();
1595 let http = FakeHttp::new(move |request| {
1596 let host = request.url.host_str().unwrap().to_string();
1597 if host == "plc.directory" {
1598 Ok(ok(squid_doc(&doc)))
1599 } else {
1600 let id = format!("did:web:{host}");
1601 Ok(ok(did_doc(DocSpec {
1602 id: &id,
1603 signing: &doc,
1604 handle: "nel.pet",
1605 pds: "https://pds.oyster.cafe",
1606 method: MethodKind::Multikey,
1607 })))
1608 }
1609 });
1610 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1611 let method = member_method();
1612 stream::iter(0..MAX_JTI_PER_ISSUER)
1613 .for_each(|i| {
1614 let atproto = &atproto;
1615 let method = &method;
1616 let signing = &signing;
1617 async move {
1618 let claims = serde_json::json!({
1619 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1620 "jti": format!("nonce-{i}"), "lxm": METHOD,
1621 });
1622 atproto
1623 .verify_service_jwt(&mint(signing, &claims), method)
1624 .await
1625 .unwrap();
1626 }
1627 })
1628 .await;
1629
1630 let hogged = serde_json::json!({
1631 "iss": SQUID, "aud": KNOT, "exp": 1_001, "iat": 999,
1632 "jti": "nonce-over-budget", "lxm": METHOD,
1633 });
1634 let error = atproto
1635 .verify_service_jwt(&mint(&signing, &hogged), &method)
1636 .await
1637 .unwrap_err();
1638 assert!(
1639 matches!(error, AtprotoError::ReplayShareExhausted { .. }),
1640 "issuer past its share fails closed, got {error:?}"
1641 );
1642
1643 let bystander = serde_json::json!({
1644 "iss": "did:web:bystander.oyster.cafe", "aud": KNOT, "exp": 1_001, "iat": 999,
1645 "jti": "nonce-bystander", "lxm": METHOD,
1646 });
1647 atproto
1648 .verify_service_jwt(&mint(&signing, &bystander), &method)
1649 .await
1650 .expect("unrelated issuer is unaffected by the hog");
1651
1652 atproto.clock.advance(Duration::from_secs(62));
1653 let after_expiry = serde_json::json!({
1654 "iss": SQUID, "aud": KNOT, "exp": 1_100, "iat": 1_050,
1655 "jti": "nonce-after-expiry", "lxm": METHOD,
1656 });
1657 atproto
1658 .verify_service_jwt(&mint(&signing, &after_expiry), &method)
1659 .await
1660 .expect("hog recovers once its nonces expire from the store");
1661 }
1662
1663 #[tokio::test]
1664 async fn the_identity_cache_stays_bounded_and_retains_hot_entries() {
1665 let signing = signer(9);
1666 let hot_hits = Arc::new(AtomicUsize::new(0));
1667 let sentinel_hits = Arc::new(AtomicUsize::new(0));
1668 let hot = hot_hits.clone();
1669 let sentinel = sentinel_hits.clone();
1670 let http = FakeHttp::new(move |request| {
1671 let host = request.url.host_str().unwrap().to_string();
1672 if host == "hot.oyster.cafe" {
1673 hot.fetch_add(1, Ordering::SeqCst);
1674 }
1675 // yeah so what
1676 if host == "sentinel.oyster.cafe" {
1677 sentinel.fetch_add(1, Ordering::SeqCst);
1678 }
1679 let id = format!("did:web:{host}");
1680 Ok(ok(did_doc(DocSpec {
1681 id: &id,
1682 signing: &signing,
1683 handle: "nel.pet",
1684 pds: "https://pds.oyster.cafe",
1685 method: MethodKind::Multikey,
1686 })))
1687 });
1688 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1689 let hot_did = did("did:web:hot.oyster.cafe");
1690 stream::iter(0..64)
1691 .for_each(|_| {
1692 let atproto = &atproto;
1693 let hot_did = hot_did.clone();
1694 async move {
1695 atproto.resolve_identity(&hot_did).await.unwrap();
1696 }
1697 })
1698 .await;
1699 stream::iter(0..MAX_IDENTITY_CACHE * 2)
1700 .for_each(|i| {
1701 let atproto = &atproto;
1702 let hot_did = hot_did.clone();
1703 async move {
1704 let filler = AccountDid::new(format!("did:web:c{i}.oyster.cafe")).unwrap();
1705 atproto.resolve_identity(&filler).await.unwrap();
1706 if i.is_multiple_of(8) {
1707 atproto.resolve_identity(&hot_did).await.unwrap();
1708 }
1709 }
1710 })
1711 .await;
1712 atproto.identities.run_pending_tasks().await;
1713 assert!(
1714 atproto.identities.entry_count().get() <= MAX_IDENTITY_CACHE as u64,
1715 "cache stays bounded under a cold-DID flood"
1716 );
1717 let before = hot_hits.load(Ordering::SeqCst);
1718 atproto.resolve_identity(&hot_did).await.unwrap();
1719 assert_eq!(
1720 hot_hits.load(Ordering::SeqCst),
1721 before,
1722 "frequently resolved identity is retained through the flood"
1723 );
1724 let novel = did("did:web:sentinel.oyster.cafe");
1725 atproto.resolve_identity(&novel).await.unwrap();
1726 atproto.resolve_identity(&novel).await.unwrap();
1727 assert_eq!(
1728 sentinel_hits.load(Ordering::SeqCst),
1729 1,
1730 "saturated cache still admits a new entry and reuses it without a re-fetch"
1731 );
1732 atproto.identities.run_pending_tasks().await;
1733 assert!(
1734 atproto.identities.entry_count().get() <= MAX_IDENTITY_CACHE as u64,
1735 "cache stays bounded after eviction"
1736 );
1737 }
1738
1739 struct GatedHttp {
1740 status: StatusCode,
1741 body: Bytes,
1742 calls: Arc<AtomicUsize>,
1743 gate: Arc<tokio::sync::Notify>,
1744 }
1745
1746 impl HttpTransport for GatedHttp {
1747 fn execute(&self, _request: HttpRequest) -> knot_runtime::HttpFuture {
1748 self.calls.fetch_add(1, Ordering::SeqCst);
1749 let gate = self.gate.clone();
1750 let response = status(self.status, self.body.clone());
1751 Box::pin(async move {
1752 gate.notified().await;
1753 Ok(response)
1754 })
1755 }
1756 }
1757
1758 async fn gated_wave(
1759 st: StatusCode,
1760 body: Bytes,
1761 ) -> (Vec<Result<Identity, AtprotoError>>, usize) {
1762 let calls = Arc::new(AtomicUsize::new(0));
1763 let gate = Arc::new(tokio::sync::Notify::new());
1764 let http = GatedHttp {
1765 status: st,
1766 body,
1767 calls: calls.clone(),
1768 gate: gate.clone(),
1769 };
1770 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1771 let subject = did(SQUID);
1772 let resolves =
1773 futures::future::join_all((0..2000).map(|_| atproto.resolve_identity(&subject)));
1774 let releaser = async {
1775 stream::iter(0..16)
1776 .for_each(|_| async {
1777 tokio::task::yield_now().await;
1778 })
1779 .await;
1780 gate.notify_one();
1781 };
1782 let (results, ()) = futures::join!(resolves, releaser);
1783 (results, calls.load(Ordering::SeqCst))
1784 }
1785
1786 #[tokio::test]
1787 async fn concurrent_cold_resolves_coalesce_into_one_fetch() {
1788 let (results, calls) = gated_wave(StatusCode::OK, squid_doc(&signer(9))).await;
1789 assert_eq!(results.len(), 2000);
1790 assert!(results.iter().all(|outcome| outcome.is_ok()));
1791 assert_eq!(
1792 calls, 1,
1793 "2000 concurrent resolves of one cold DID issue exactly one outbound fetch"
1794 );
1795 }
1796
1797 #[tokio::test]
1798 async fn a_concurrent_429_wave_gets_the_real_error_and_never_poisons_the_cache() {
1799 let (results, calls) = gated_wave(StatusCode::TOO_MANY_REQUESTS, Bytes::new()).await;
1800 assert_eq!(calls, 1, "failing wave coalesces into one outbound fetch");
1801 assert!(
1802 results.iter().all(|outcome| matches!(
1803 outcome,
1804 Err(AtprotoError::Resolve(ResolveError::Status { status })) if status.get() == 429
1805 )),
1806 "every caller in the wave receives a real 429, never a poisoned RecentlyFailed"
1807 );
1808 }
1809
1810 #[tokio::test]
1811 async fn a_coalesced_404_wave_gives_one_caller_the_real_error_and_masks_the_rest() {
1812 let (results, calls) = gated_wave(StatusCode::NOT_FOUND, Bytes::new()).await;
1813 assert_eq!(calls, 1, "404 wave coalesces into one outbound fetch");
1814 let real = results
1815 .iter()
1816 .filter(|outcome| {
1817 matches!(
1818 outcome,
1819 Err(AtprotoError::Resolve(ResolveError::Status { status })) if status.get() == 404
1820 )
1821 })
1822 .count();
1823 let masked = results
1824 .iter()
1825 .filter(|outcome| {
1826 matches!(
1827 outcome,
1828 Err(AtprotoError::Resolve(ResolveError::RecentlyFailed { .. }))
1829 )
1830 })
1831 .count();
1832 assert_eq!(real, 1, "exactly one caller observes the real 404");
1833 assert_eq!(
1834 masked, 1999,
1835 "the rest of the coalesced wave is masked as RecentlyFailed"
1836 );
1837 }
1838
1839 #[tokio::test]
1840 async fn a_prepared_plc_operation_is_posted_and_a_rejection_is_typed() {
1841 let prepared = prepare_repo_did(
1842 &runtime_signer(11),
1843 &knot_types::KnotServiceUrl::new("https://knot.nel.pet").unwrap(),
1844 &repo_nonce(111),
1845 )
1846 .unwrap();
1847 let expected_path = format!("/{}", prepared.did.as_str());
1848 let http = FakeHttp::new(move |request| {
1849 assert_eq!(request.method, http::Method::POST);
1850 assert_eq!(request.url.path(), expected_path);
1851 assert!(request.body.as_ref().is_some_and(|body| !body.is_empty()));
1852 Ok(ok(Bytes::new()))
1853 });
1854 Atproto::new(http, clock(), knot_did(KNOT), plc())
1855 .submit_plc_operation(&prepared)
1856 .await
1857 .unwrap();
1858
1859 let rejected = prepare_repo_did(
1860 &runtime_signer(12),
1861 &knot_types::KnotServiceUrl::new("https://knot.nel.pet").unwrap(),
1862 &repo_nonce(112),
1863 )
1864 .unwrap();
1865 let http = FakeHttp::new(move |_| Ok(status(StatusCode::BAD_REQUEST, Bytes::new())));
1866 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1867 assert!(matches!(
1868 atproto.submit_plc_operation(&rejected).await,
1869 Err(AtprotoError::PlcSubmit { status, .. }) if status.get() == 400
1870 ));
1871 }
1872
1873 #[tokio::test]
1874 async fn a_did_web_document_verification_covers_present_absent_and_missing() {
1875 let signing = signer(9);
1876 let doc = signing.clone();
1877 let http = FakeHttp::new(move |request| {
1878 assert_eq!(
1879 request.url.as_str(),
1880 "https://limpet.olaren.dev/.well-known/did.json"
1881 );
1882 Ok(ok(did_doc(DocSpec {
1883 id: "did:web:limpet.olaren.dev",
1884 signing: &doc,
1885 handle: "nel.pet",
1886 pds: "https://pds.oyster.cafe",
1887 method: MethodKind::Multikey,
1888 })))
1889 });
1890 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1891 atproto
1892 .verify_did_web_publishes_key(
1893 &repo_did("did:web:limpet.olaren.dev"),
1894 &PublicKeyBytes::from_bytes(sec1(&signing)),
1895 )
1896 .await
1897 .unwrap();
1898
1899 let published = signer(9);
1900 let http = FakeHttp::new(move |_| {
1901 Ok(ok(did_doc(DocSpec {
1902 id: "did:web:limpet.olaren.dev",
1903 signing: &published,
1904 handle: "nel.pet",
1905 pds: "https://pds.oyster.cafe",
1906 method: MethodKind::Multikey,
1907 })))
1908 });
1909 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1910 let error = atproto
1911 .verify_did_web_publishes_key(
1912 &repo_did("did:web:limpet.olaren.dev"),
1913 &PublicKeyBytes::from_bytes(sec1(&signer(3))),
1914 )
1915 .await
1916 .unwrap_err();
1917 assert!(matches!(
1918 error,
1919 AtprotoError::Resolve(ResolveError::ExpectedKeyAbsent { .. })
1920 ));
1921
1922 let http = FakeHttp::new(|_| Ok(status(StatusCode::NOT_FOUND, Bytes::new())));
1923 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1924 let error = atproto
1925 .verify_did_web_publishes_key(
1926 &repo_did("did:web:limpet.olaren.dev"),
1927 &PublicKeyBytes::from_bytes(vec![1, 2, 3]),
1928 )
1929 .await
1930 .unwrap_err();
1931 assert!(matches!(
1932 error,
1933 AtprotoError::Resolve(ResolveError::Status { status }) if status.get() == 404
1934 ));
1935 }
1936
1937 #[tokio::test]
1938 async fn a_pointer_record_is_published_to_the_subjects_pds_over_service_auth() {
1939 let doc_signer = signer(9);
1940 let knot_key = runtime_signer(21);
1941 let knot_public = knot_runtime::Signer::public_key(&knot_key);
1942 let http = FakeHttp::new(move |request| {
1943 if request.url.host_str() == Some("plc.directory") {
1944 return Ok(ok(squid_doc(&doc_signer)));
1945 }
1946 assert_eq!(request.method, http::Method::POST);
1947 assert_eq!(
1948 request.url.as_str(),
1949 "https://pds.oyster.cafe/xrpc/com.atproto.repo.putRecord"
1950 );
1951 let bearer = request
1952 .headers
1953 .get(http::header::AUTHORIZATION)
1954 .and_then(|value| value.to_str().ok())
1955 .and_then(|value| value.strip_prefix("Bearer "))
1956 .expect("request includes a bearer service token");
1957 let parsed = knot_types::service_auth::parse_jwt(bearer).unwrap();
1958 assert_eq!(parsed.claims().iss.as_str(), KNOT);
1959 assert_eq!(parsed.claims().aud.as_str(), "did:web:pds.oyster.cafe");
1960 assert_eq!(
1961 parsed.claims().lxm.as_ref().unwrap().as_str(),
1962 "com.atproto.repo.putRecord"
1963 );
1964 assert!(parsed.claims().jti.is_some());
1965 let key = knot_types::service_auth::PublicKey::from_k256_bytes(knot_public.as_bytes())
1966 .unwrap();
1967 knot_types::service_auth::verify_signature(&parsed, &key)
1968 .expect("token is signed by the knot key");
1969 let body: serde_json::Value =
1970 serde_json::from_slice(request.body.as_ref().unwrap()).unwrap();
1971 assert_eq!(body["repo"], SQUID);
1972 assert_eq!(body["collection"], "sh.tangled.knot.member");
1973 assert_eq!(body["rkey"], POINTER_RKEY);
1974 assert_eq!(body["record"]["$type"], "sh.tangled.knot.member");
1975 assert_eq!(body["record"]["subject"], "did:plc:lyna");
1976 assert_eq!(body["record"]["domain"], "knot.nel.pet");
1977 let receipt = serde_json::json!({
1978 "uri": format!("at://{SQUID}/sh.tangled.knot.member/{POINTER_RKEY}"),
1979 "cid": POINTER_CID,
1980 });
1981 Ok(ok(Bytes::from(serde_json::to_vec(&receipt).unwrap())))
1982 });
1983 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
1984 let rkey = Rkey::new_owned(POINTER_RKEY).unwrap();
1985 let receipt = atproto
1986 .publish_pointer(
1987 &ServiceAuth::new(&knot_key, &entropy(31)),
1988 &did(SQUID),
1989 &rkey,
1990 &member_pointer(),
1991 )
1992 .await
1993 .unwrap();
1994 assert_eq!(
1995 receipt.uri.as_str(),
1996 format!("at://{SQUID}/sh.tangled.knot.member/{POINTER_RKEY}")
1997 );
1998 assert_eq!(receipt.cid.as_str(), POINTER_CID);
1999 }
2000
2001 async fn rejected_put_record(st: StatusCode) -> AtprotoError {
2002 let doc_signer = signer(9);
2003 let http = FakeHttp::new(move |request| {
2004 if request.url.host_str() == Some("plc.directory") {
2005 return Ok(ok(squid_doc(&doc_signer)));
2006 }
2007 Ok(status(st, Bytes::new()))
2008 });
2009 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
2010 let rkey = Rkey::new_owned(POINTER_RKEY).unwrap();
2011 atproto
2012 .publish_pointer(
2013 &ServiceAuth::new(&runtime_signer(22), &entropy(31)),
2014 &did(SQUID),
2015 &rkey,
2016 &member_pointer(),
2017 )
2018 .await
2019 .unwrap_err()
2020 }
2021
2022 #[tokio::test]
2023 async fn a_rejected_put_record_is_a_typed_error_and_transient_only_on_server_failure() {
2024 let server_failure = rejected_put_record(StatusCode::BAD_GATEWAY).await;
2025 assert!(matches!(
2026 server_failure,
2027 AtprotoError::PutRecord { status, .. } if status.get() == 502
2028 ));
2029 assert!(server_failure.is_transient());
2030
2031 let rate_limited = rejected_put_record(StatusCode::TOO_MANY_REQUESTS).await;
2032 assert!(matches!(
2033 rate_limited,
2034 AtprotoError::PutRecord { status, .. } if status.get() == 429
2035 ));
2036 assert!(rate_limited.is_transient());
2037
2038 let client_failure = rejected_put_record(StatusCode::BAD_REQUEST).await;
2039 assert!(matches!(
2040 client_failure,
2041 AtprotoError::PutRecord { status, .. } if status.get() == 400
2042 ));
2043 assert!(!client_failure.is_transient());
2044 }
2045
2046 #[tokio::test]
2047 async fn a_malformed_put_record_receipt_is_a_typed_error() {
2048 let doc_signer = signer(9);
2049 let http = FakeHttp::new(move |request| {
2050 if request.url.host_str() == Some("plc.directory") {
2051 return Ok(ok(squid_doc(&doc_signer)));
2052 }
2053 Ok(ok(Bytes::from_static(b"not json")))
2054 });
2055 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
2056 let rkey = Rkey::new_owned(POINTER_RKEY).unwrap();
2057 let error = atproto
2058 .publish_pointer(
2059 &ServiceAuth::new(&runtime_signer(23), &entropy(31)),
2060 &did(SQUID),
2061 &rkey,
2062 &member_pointer(),
2063 )
2064 .await
2065 .unwrap_err();
2066 assert!(matches!(error, AtprotoError::MalformedReceipt(_)));
2067 }
2068
2069 #[tokio::test]
2070 async fn a_pointer_to_an_insecure_pds_endpoint_is_refused() {
2071 let doc_signer = signer(9);
2072 let http = FakeHttp::new(move |request| {
2073 assert_eq!(
2074 request.url.host_str(),
2075 Some("plc.directory"),
2076 "no request may reach the insecure PDS"
2077 );
2078 Ok(ok(did_doc(DocSpec {
2079 id: SQUID,
2080 signing: &doc_signer,
2081 handle: "nel.pet",
2082 pds: "http://pds.oyster.cafe",
2083 method: MethodKind::Multikey,
2084 })))
2085 });
2086 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc());
2087 let rkey = Rkey::new_owned(POINTER_RKEY).unwrap();
2088 let error = atproto
2089 .publish_pointer(
2090 &ServiceAuth::new(&runtime_signer(24), &entropy(31)),
2091 &did(SQUID),
2092 &rkey,
2093 &member_pointer(),
2094 )
2095 .await
2096 .unwrap_err();
2097 assert!(matches!(
2098 error,
2099 AtprotoError::Resolve(ResolveError::InsecureScheme { .. })
2100 ));
2101 }
2102}