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