Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197use std::ops::Not;use std::sync::Arc;use std::sync::atomic::{AtomicUsize, Ordering};use std::time::{Duration, Instant};
use futures::future::BoxFuture;use hickory_resolver::TokioResolver;use hickory_resolver::proto::rr::{RData, rdata::TXT};use jacquard_common::Data;use jacquard_common::IntoStatic;use jacquard_common::types::crypto::PublicKey;use jacquard_common::types::did_doc::DidDocument;use jacquard_common::types::ident::AtIdentifier;use jacquard_common::types::string::{Did, Handle};use miette::{Diagnostic, IntoDiagnostic};use reqwest::StatusCode;use scc::HashCache;use serde::Deserialize;use smol_str::SmolStr;use thiserror::Error;use url::Url;
use crate::net::{ HANDLE_REQUEST_TIMEOUT, PublicHttpClient, PublicHttpError, handle_http, public_http,};
// as per spec: https://web.plc.directory/spec/v0.1/did-plc// "As an anti-abuse mechanism, operations have a maximum size when encoded as// DAG-CBOR. The current limit is 7500 bytes."const MAX_DID_DOC_BYTES: usize = 16384; // this is generous i think but we deal with json so..const MAX_HANDLE_BODY_BYTES: usize = 8192;const PUBLIC_HANDLE_RESOLVER: &str = "https://public.api.bsky.app/";
/// the total budget for every source used to resolve one handle.const HANDLE_VERIFY_TIMEOUT: Duration = HANDLE_REQUEST_TIMEOUT;
/// how long a successful handle verdict is reused; failures are never stored.const HANDLE_VERIFY_TTL: Duration = Duration::from_secs(30);
trait HandleDns: Send + Sync { fn resolve(&self, handle: Handle) -> BoxFuture<'static, Option<Did>>;}
struct SystemHandleDns(TokioResolver);
impl SystemHandleDns { fn new() -> Self { let mut builder = TokioResolver::builder_tokio().expect("system DNS config must load"); builder.options_mut().timeout = HANDLE_REQUEST_TIMEOUT; Self(builder.build().expect("system DNS resolver must build")) }}
impl HandleDns for SystemHandleDns { fn resolve(&self, handle: Handle) -> BoxFuture<'static, Option<Did>> { let resolver = self.0.clone(); Box::pin(async move { let records = resolver .txt_lookup(format!("_atproto.{}.", handle.as_str())) .await .ok()?; records .answers() .iter() .find_map(|record| match &record.data { RData::TXT(txt) => did_from_txt(txt), _ => None, }) }) }}
fn did_from_txt(txt: &TXT) -> Option<Did> { let value = txt .txt_data .iter() .flat_map(|part| part.iter().copied()) .collect::<Vec<_>>(); std::str::from_utf8(&value) .ok()? .strip_prefix("did=") .and_then(|did| Did::new_owned(did).ok())}
#[cfg(test)]struct NoHandleDns;
#[cfg(test)]impl HandleDns for NoHandleDns { fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option<Did>> { Box::pin(async { None }) }}
#[derive(Debug, Diagnostic, Error)]pub enum ResolverError { #[error("{0}")] Generic(miette::Report), #[error("too many requests")] Ratelimited, #[error("handle resolution exhausted: {0}")] HandleResolutionExhausted(miette::Report), #[error("transport error: {0}")] Transport(SmolStr), #[error("no PDS service found in DID Doc for {0}")] MissingPds(Did),}
impl From<miette::Report> for ResolverError { fn from(report: miette::Report) -> Self { ResolverError::Generic(report) }}
#[derive(Clone, Debug)]pub struct MiniDoc { pub pds: Url, pub handle: Option<Handle>, pub key: Option<PublicKey<'static>>,}
#[derive(Clone)]enum HandleSource { HttpsWellKnown, Xrpc(Url),}
#[derive(Deserialize)]struct ResolveHandleOutput { did: Did,}
struct ResolverInner { plc_urls: Vec<Url>, handle_sources: Vec<HandleSource>, documents: PublicHttpClient, handles: PublicHttpClient, handle_dns: Arc<dyn HandleDns>, next_idx: AtomicUsize, cache: HashCache<Did, MiniDoc>, /// when a `(did, handle)` verdict, always positive, stops being reused. verified_handles: HashCache<(Did, Handle), Instant>,}
#[derive(Clone)]pub struct Resolver { inner: Arc<ResolverInner>,}
fn production_handle_sources() -> Vec<HandleSource> { vec![ HandleSource::HttpsWellKnown, HandleSource::Xrpc( Url::parse(PUBLIC_HANDLE_RESOLVER).expect("public resolver URL is valid"), ), ]}
impl Resolver { pub fn new(plc_urls: Vec<Url>, identity_cache_size: u64) -> Self { Self::with_clients( plc_urls, production_handle_sources(), identity_cache_size, public_http(), handle_http(), Arc::new(SystemHandleDns::new()), ) }
/// one stub serves both the DID document and `resolveHandle`. #[cfg(all(test, feature = "indexer_stream"))] pub(crate) fn stubbed_for_test( base: Url, identity_cache_size: u64, public: PublicHttpClient, ) -> Self { Self::with_clients( vec![base.clone()], vec![HandleSource::Xrpc(base)], identity_cache_size, public.clone(), public, Arc::new(NoHandleDns), ) }
#[cfg(test)] pub(crate) fn with_plc_urls_and_client( plc_urls: Vec<Url>, identity_cache_size: u64, public: PublicHttpClient, ) -> Self { Self::with_clients( plc_urls, production_handle_sources(), identity_cache_size, public.clone(), public, Arc::new(NoHandleDns), ) }
fn with_clients( plc_urls: Vec<Url>, handle_sources: Vec<HandleSource>, identity_cache_size: u64, documents: PublicHttpClient, handles: PublicHttpClient, handle_dns: Arc<dyn HandleDns>, ) -> Self { assert!( !plc_urls.is_empty(), "at least one PLC source must be provided" ); let capacity = std::cmp::min(1000, (identity_cache_size / 100) as usize);
Self { inner: Arc::new(ResolverInner { plc_urls, handle_sources, documents, handles, handle_dns, next_idx: AtomicUsize::new(0), cache: HashCache::with_capacity(capacity, identity_cache_size as usize), verified_handles: HashCache::with_capacity(capacity, identity_cache_size as usize), }), } }
pub async fn invalidate(&self, did: &Did) { self.inner.cache.remove_async(did).await; }
pub fn invalidate_sync(&self, did: &Did) { self.inner.cache.remove_sync(did); }
pub async fn resolve_did(&self, identifier: &AtIdentifier) -> Result<Did, ResolverError> { match identifier { AtIdentifier::Did(did) => Ok(did.clone()), AtIdentifier::Handle(handle) => { self.resolve_handle_with_timeout(handle, HANDLE_VERIFY_TIMEOUT) .await } } }
pub async fn resolve_doc(&self, did: &Did) -> Result<MiniDoc, ResolverError> { let did_static = did.clone(); if let Some(entry) = self.inner.cache.get_async(&did_static).await { return Ok(entry.get().clone()); }
let mini = self.resolve_doc_fresh(did).await?; let _ = self.inner.cache.put_async(did_static, mini.clone()).await; Ok(mini) }
/// resolves the current DID document without reading or replacing the /// cached mini document. callers that discard a raced result leave the /// cache untouched. pub async fn resolve_doc_fresh(&self, did: &Did) -> Result<MiniDoc, ResolverError> { Self::decode_mini_doc(did, &self.fetch_document(did).await?) }
pub(crate) async fn cache_doc(&self, did: Did, doc: MiniDoc) { let _ = self.inner.cache.entry_async(did).await.put_entry(doc); }
pub async fn resolve_identity_info( &self, did: &Did, ) -> Result<(Url, Option<Handle>), ResolverError> { let mini = self.resolve_doc(did).await?; Ok((mini.pds, mini.handle)) }
pub async fn resolve_signing_key( &self, did: &Did, ) -> Result<PublicKey<'static>, ResolverError> { let did = did.clone(); let mini = self.resolve_doc(&did).await?; Ok(mini.key.ok_or(NoSigningKeyError(did)).into_diagnostic()?) }
/// resolves the full DID document as raw [`Data`] without caching, and /// extracts the first `alsoKnownAs` handle if present. a document without a /// PDS service still answers here: only [`Self::resolve_doc`] needs one. pub async fn resolve_raw_doc( &self, did: &Did, ) -> Result<(Data, Option<Handle>), ResolverError> { let body = self.fetch_document(did).await?; let handle = first_handle(&parse_document(did, &body)?); let raw = serde_json::from_slice(&body).map_err(|error| { ResolverError::Generic(miette::miette!("failed to parse DID document: {error}")) })?; Ok((raw, handle)) }
async fn fetch_document(&self, did: &Did) -> Result<Vec<u8>, ResolverError> { if did.starts_with("did:web:") { let url = did_web_url(did)?; return self .get(&self.inner.documents, &url, MAX_DID_DOC_BYTES) .await; } if !did.starts_with("did:plc:") { return Err(ResolverError::Generic(miette::miette!( "unsupported DID method: {did}" ))); }
let sources = &self.inner.plc_urls; let start = self.inner.next_idx.fetch_add(1, Ordering::Relaxed) % sources.len(); for offset in 0..sources.len() { let url = plc_url(&sources[(start + offset) % sources.len()], did); let result = self .get(&self.inner.documents, &url, MAX_DID_DOC_BYTES) .await; let retriable = matches!( result, Err(ResolverError::Ratelimited | ResolverError::Transport(_)) ); if !retriable || offset + 1 == sources.len() { return result; } } unreachable!("PLC source list is non-empty") }
fn decode_mini_doc(did: &Did, body: &[u8]) -> Result<MiniDoc, ResolverError> { let doc = parse_document(did, body)?; let pds = doc .pds_endpoint() .ok_or_else(|| ResolverError::MissingPds(did.clone()))?; Ok(MiniDoc { pds: Url::parse(pds.as_str()).map_err(|error| { ResolverError::Generic(miette::miette!("invalid PDS endpoint for {did}: {error}")) })?, handle: first_handle(&doc), // an undecodable key is a missing key: the PDS endpoint still // works, and `resolve_signing_key` reports the absence itself. key: doc.atproto_public_key().ok().flatten(), }) }
async fn get( &self, client: &PublicHttpClient, url: &Url, max_body_bytes: usize, ) -> Result<Vec<u8>, ResolverError> { let request = http::Request::get(url.as_str()) .body(Vec::new()) .map_err(|error| { ResolverError::Generic(miette::miette!("invalid request URL {url}: {error}")) })?; let response = client .send_http_bounded(request, max_body_bytes) .await .map_err(|error| match error { // an oversized body is the source's answer, not an outage. PublicHttpError::BodyTooLarge { limit } => ResolverError::Generic(miette::miette!( "identity source {url} response exceeds {limit} bytes" )), error => ResolverError::Transport(error.to_string().into()), })?; match response.status() { status if status.is_success() => Ok(response.into_body()), StatusCode::TOO_MANY_REQUESTS => Err(ResolverError::Ratelimited), // an outage at one mirror is not an answer about the identity, so // it stays retriable for the next source. status if status.is_server_error() => Err(ResolverError::Transport( format!("identity source {url} returned {status}").into(), )), status => Err(ResolverError::Generic(miette::miette!( "identity source {url} returned {status}" ))), } }
async fn resolve_handle_with_timeout( &self, handle: &Handle, timeout: Duration, ) -> Result<Did, ResolverError> { tokio::time::timeout(timeout, self.resolve_handle_inner(handle)) .await .map_err(|_| ResolverError::Transport("handle resolution timed out".into()))? }
async fn resolve_handle_inner(&self, handle: &Handle) -> Result<Did, ResolverError> { if let Some(did) = self.inner.handle_dns.resolve(handle.clone()).await { return Ok(did); } let mut ratelimited = false; for source in &self.inner.handle_sources { match self.resolve_handle_source(source, handle).await { Ok(did) => return Ok(did), // a rate limit is our own fault, so it must never be reported // as an absent handle mapping. Err(ResolverError::Ratelimited) => ratelimited = true, Err(_) => {} } } if ratelimited { return Err(ResolverError::Ratelimited); } Err(ResolverError::HandleResolutionExhausted(miette::miette!( "failed to resolve handle {handle}" ))) }
async fn resolve_handle_source( &self, source: &HandleSource, handle: &Handle, ) -> Result<Did, ResolverError> { match source { HandleSource::HttpsWellKnown => { let url = Url::parse(&format!( "https://{}/.well-known/atproto-did", handle.as_str() )) .map_err(|error| { ResolverError::Generic(miette::miette!("invalid handle host {handle}: {error}")) })?; let body = self .get(&self.inner.handles, &url, MAX_HANDLE_BODY_BYTES) .await?; let text = std::str::from_utf8(&body).map_err(|error| { ResolverError::Generic(miette::miette!( "invalid well-known body for {handle}: {error}" )) })?; let did = text .lines() .find(|line| !line.trim().is_empty()) .ok_or_else(|| { ResolverError::Generic(miette::miette!( "empty well-known body for {handle}" )) })?; Did::new_owned(did.trim()).map_err(|error| { ResolverError::Generic(miette::miette!( "invalid well-known DID for {handle}: {error}" )) }) } HandleSource::Xrpc(base) => { let url = xrpc_url( base, "com.atproto.identity.resolveHandle", "handle", handle.as_str(), ); let body = self .get(&self.inner.handles, &url, MAX_HANDLE_BODY_BYTES) .await?; serde_json::from_slice::<ResolveHandleOutput>(&body) .map(|output| output.did) .map_err(|error| { ResolverError::Generic(miette::miette!( "invalid resolveHandle response from {url}: {error}" )) }) } } }
/// returns `true` if `handle` resolves back to `did`. /// /// this is the reverse half of handle validity: the caller supplies the /// forward half by taking `handle` from that DID's own document. /// /// an exhausted, failed, or timed-out network lookup answers `false`, so /// callers can use `handle.invalid` (atproto.com/specs/handle#invalid-handles). /// only positive verdicts are stored, for [`HANDLE_VERIFY_TTL`]: a failure /// cannot distinguish a missing mapping from an outage, and a stored /// verdict belongs to one pair, so a renamed account ages out rather than /// being invalidated. pub async fn verify_handle(&self, did: &Did, handle: &Handle) -> Result<bool, ResolverError> { let key = (did.clone(), handle.clone()); let fresh = self .inner .verified_handles .read_async(&key, |_, expires_at| *expires_at > Instant::now()) .await; if fresh == Some(true) { return Ok(true); }
let verdict = self .verify_handle_with_timeout(did, handle, HANDLE_VERIFY_TIMEOUT) .await?; if verdict { // concurrent verdicts of one pair can land in either order; both // were true, so the loser only re-dates the entry. let _ = self .inner .verified_handles .entry_async(key) .await .put_entry(Instant::now() + HANDLE_VERIFY_TTL); } Ok(verdict) }
async fn verify_handle_with_timeout( &self, did: &Did, handle: &Handle, timeout: Duration, ) -> Result<bool, ResolverError> { Self::handle_verdict(did, self.resolve_handle_with_timeout(handle, timeout).await) }
fn handle_verdict( did: &Did, result: Result<Did, ResolverError>, ) -> Result<bool, ResolverError> { match result { Ok(resolved_did) => Ok(resolved_did.as_str() == did.as_str()), Err(ResolverError::HandleResolutionExhausted(_) | ResolverError::Transport(_)) => { Ok(false) } Err(error) => Err(error), } }}
/// parses a DID document and rejects one that answers for another DID.fn parse_document(did: &Did, body: &[u8]) -> Result<DidDocument, ResolverError> { let doc: DidDocument = serde_json::from_slice(body).map_err(|error| { ResolverError::Generic(miette::miette!("failed to parse DID document: {error}")) })?; if doc.id.as_str() != did.as_str() { return Err(ResolverError::Generic(miette::miette!( "DID document id {} does not match requested {did}", doc.id ))); } Ok(doc)}
fn first_handle(doc: &DidDocument) -> Option<Handle> { let mut handles = doc.handles(); handles .is_empty() .not() .then(|| handles.remove(0).into_static())}
fn plc_url(base: &Url, did: &Did) -> Url { let mut url = base.clone(); url.set_path(&format!( "{}/{}", base.path().trim_end_matches('/'), did.as_str() )); url.set_query(None); url.set_fragment(None); url}
fn xrpc_url(base: &Url, nsid: &str, key: &str, value: &str) -> Url { let mut url = base.clone(); let path = format!("{}/xrpc/{nsid}", base.path().trim_end_matches('/')); url.set_path(&path); url.set_query(None); url.set_fragment(None); url.query_pairs_mut().append_pair(key, value); url}
fn did_web_url(did: &Did) -> Result<Url, ResolverError> { let rest = did .as_str() .strip_prefix("did:web:") .ok_or_else(|| ResolverError::Generic(miette::miette!("unsupported DID method: {did}")))?; let mut parts = rest.split(':'); let host = parts .next() .filter(|host| !host.is_empty()) .ok_or_else(|| ResolverError::Generic(miette::miette!("invalid did:web: {did}")))? // a port is percent-encoded in the method-specific id // (w3c-ccg.github.io/did-method-web/#did-method-operations). .replace("%3A", ":") .replace("%3a", ":"); let segments = parts.collect::<Vec<_>>(); let mut url = Url::parse(&format!("https://{host}/")).map_err(|error| { ResolverError::Generic(miette::miette!("invalid did:web host in {did}: {error}")) })?; if segments.is_empty() { url.set_path("/.well-known/did.json"); } else { url.set_path(&format!("/{}/did.json", segments.join("/"))); } Ok(url)}
#[derive(Debug, Diagnostic, Error)]#[error("no atproto signing key in DID doc for {0}")]pub struct NoSigningKeyError(Did);
#[cfg(test)]mod tests { use super::*; use axum::{Router, http::header, response::IntoResponse}; use std::sync::Arc; use tokio::sync::RwLock;
struct HandleStub { addr: std::net::SocketAddr, resolve_calls: Arc<std::sync::atomic::AtomicUsize>, }
struct FixedHandleDns { did: Option<Did>, calls: Arc<std::sync::atomic::AtomicUsize>, }
impl HandleDns for FixedHandleDns { fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option<Did>> { self.calls.fetch_add(1, Ordering::SeqCst); let did = self.did.clone(); Box::pin(async move { did }) } }
struct PendingHandleDns;
impl HandleDns for PendingHandleDns { fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option<Did>> { Box::pin(std::future::pending()) } }
/// serves the XRPC `resolveHandle` endpoint over http. async fn spawn_handle_stub(did: Did) -> HandleStub { use axum::extract::State; use std::sync::atomic::AtomicUsize;
#[derive(Clone)] struct Counters { did: Did, resolve_calls: Arc<AtomicUsize>, }
async fn resolve_handle(State(counters): State<Counters>) -> axum::Json<serde_json::Value> { counters .resolve_calls .fetch_add(1, std::sync::atomic::Ordering::SeqCst); axum::Json(serde_json::json!({ "did": counters.did.as_str() })) }
async fn not_found() -> (axum::http::StatusCode, &'static str) { (axum::http::StatusCode::NOT_FOUND, "no well-known here") }
let counters = Counters { did, resolve_calls: Arc::new(AtomicUsize::new(0)), }; let app = Router::new() .route( "/xrpc/com.atproto.identity.resolveHandle", axum::routing::get(resolve_handle), ) .fallback(not_found) .with_state(counters.clone()); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); HandleStub { addr, resolve_calls: counters.resolve_calls, } }
/// accepts connections and never answers them: a dropped connect is not /// reproducible locally, and both end at the same transport budget. async fn spawn_silent_listener() -> (std::net::SocketAddr, Arc<std::sync::atomic::AtomicUsize>) { use std::sync::atomic::AtomicUsize;
let accepted = Arc::new(AtomicUsize::new(0)); let counter = accepted.clone(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { let mut held = Vec::new(); while let Ok((stream, _)) = listener.accept().await { counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst); held.push(stream); } }); (addr, accepted) }
/// the well-known step is always https, so only the http resolveHandle step /// can be served by a local stub. fn stub_resolver(public: PublicHttpClient, handle_dns: Arc<dyn HandleDns>) -> Resolver { let base = Url::parse(HANDLE_BASE).unwrap(); Resolver::with_clients( vec![base.clone()], vec![HandleSource::HttpsWellKnown, HandleSource::Xrpc(base)], 10, public.clone(), public, handle_dns, ) }
/// stub hosts resolve locally, with the handle transport's budget shape. fn stub_client( base: std::net::SocketAddr, hosts: &[(&str, std::net::SocketAddr)], ) -> PublicHttpClient { let mut builder = reqwest::Client::builder() .no_proxy() .resolve("public.example", base) .timeout(Duration::from_millis(500)) .connect_timeout(Duration::from_millis(500)); for &(host, addr) in hosts { builder = builder.resolve(host, addr); } PublicHttpClient::test(builder.build().unwrap()) }
const HANDLE_BASE: &str = "http://public.example/";
#[test] fn split_dns_txt_data_is_joined_before_parsing() { let txt = TXT::from_bytes(vec![b"did=did:plc:", b"ewvi7nxzyoun6zhxrhs64oiz"]); assert_eq!( did_from_txt(&txt).unwrap().as_str(), "did:plc:ewvi7nxzyoun6zhxrhs64oiz" ); assert!(did_from_txt(&TXT::new(vec!["not-a-did".into()])).is_none()); }
#[tokio::test] async fn dns_txt_answer_precedes_network_fallbacks() { use std::sync::atomic::Ordering;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let stub = spawn_handle_stub(did.clone()).await; let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let public = stub_client(stub.addr, &[("fast.test", stub.addr)]); let resolver = stub_resolver( public, Arc::new(FixedHandleDns { did: Some(did.clone()), calls: calls.clone(), }), );
assert!(resolver.verify_handle(&did, &handle).await.unwrap()); assert_eq!(calls.load(Ordering::SeqCst), 1); assert_eq!( stub.resolve_calls.load(Ordering::SeqCst), 0, "a DNS answer must prevent all network fallbacks" ); }
#[tokio::test] async fn the_whole_verification_has_one_budget() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("slow.test").unwrap(); let stub = spawn_handle_stub(did.clone()).await; let public = stub_client(stub.addr, &[]); let resolver = stub_resolver(public, Arc::new(PendingHandleDns));
let verdict = tokio::time::timeout( Duration::from_secs(1), resolver.verify_handle_with_timeout(&did, &handle, Duration::from_millis(20)), ) .await .expect("the total verification budget must terminate a stalled source") .unwrap();
assert!(!verdict); }
#[tokio::test] async fn a_handle_host_that_never_answers_is_bounded() { use std::sync::atomic::Ordering;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("slow.test").unwrap(); let stub = spawn_handle_stub(did.clone()).await; let (silent, silent_accepts) = spawn_silent_listener().await; let resolver = stub_resolver( stub_client(stub.addr, &[("slow.test", silent)]), Arc::new(NoHandleDns), );
let started = Instant::now(); let verdict = tokio::time::timeout( Duration::from_secs(5), resolver.verify_handle(&did, &handle), ) .await .expect("a handle host that never answers must not hold the verdict") .unwrap();
assert!(verdict); assert_eq!( silent_accepts.load(Ordering::SeqCst), 1, "the account's host is still asked, in spec order" ); assert_eq!( stub.resolve_calls.load(Ordering::SeqCst), 1, "the bounded attempt must fall through to the next step" ); assert!( started.elapsed() < Duration::from_secs(2), "the attempt must be bounded, took {:?}", started.elapsed() ); }
#[tokio::test] async fn only_answered_verdicts_are_reused() { use std::sync::atomic::Ordering;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let other = Did::new_static("did:plc:j5hmlfdrwkvtxm7cjmu7j2is").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let stub = spawn_handle_stub(did.clone()).await; // fast.test's https step fails the handshake, so resolveHandle answers. let resolver = stub_resolver( stub_client(stub.addr, &[("fast.test", stub.addr)]), Arc::new(NoHandleDns), );
assert!(resolver.verify_handle(&did, &handle).await.unwrap()); assert_eq!(stub.resolve_calls.load(Ordering::SeqCst), 1);
assert!(resolver.verify_handle(&did, &handle).await.unwrap()); assert_eq!( stub.resolve_calls.load(Ordering::SeqCst), 1, "a successful verdict is reused" );
assert!(!resolver.verify_handle(&other, &handle).await.unwrap()); assert_eq!( stub.resolve_calls.load(Ordering::SeqCst), 2, "a verdict belongs to one did and handle pair" ); assert!(!resolver.verify_handle(&other, &handle).await.unwrap()); assert_eq!( stub.resolve_calls.load(Ordering::SeqCst), 3, "a negative verdict is never remembered" ); }
async fn spawn_status_server(status: StatusCode) -> std::net::SocketAddr { let app = Router::new().fallback(move || async move { status }); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); addr }
/// resolves handles through one http XRPC source only. fn xrpc_handle_resolver(address: std::net::SocketAddr) -> Resolver { let base = Url::parse(HANDLE_BASE).unwrap(); let public = test_public_client(address); Resolver::with_clients( vec![base.clone()], vec![HandleSource::Xrpc(base)], 10, public.clone(), public, Arc::new(NoHandleDns), ) }
#[tokio::test] async fn a_rate_limited_document_source_stays_rate_limited() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let resolver = plc_resolver(spawn_status_server(StatusCode::TOO_MANY_REQUESTS).await);
assert!(matches!( resolver.resolve_doc(&did).await.unwrap_err(), ResolverError::Ratelimited )); }
#[tokio::test] async fn a_rate_limited_handle_source_is_not_an_invalid_handle() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let resolver = xrpc_handle_resolver(spawn_status_server(StatusCode::TOO_MANY_REQUESTS).await);
assert!(matches!( resolver.verify_handle(&did, &handle).await.unwrap_err(), ResolverError::Ratelimited )); }
#[tokio::test] async fn a_handle_no_source_knows_is_unverified() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let resolver = xrpc_handle_resolver(spawn_status_server(StatusCode::NOT_FOUND).await);
assert!( !resolver.verify_handle(&did, &handle).await.unwrap(), "an answer of 'no such handle' is an unverified handle, not an error" ); }
#[tokio::test] async fn the_xrpc_source_answers_with_the_did_it_encodes() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let stub = spawn_handle_stub(did.clone()).await; let resolver = xrpc_handle_resolver(stub.addr);
assert_eq!( resolver .resolve_did(&AtIdentifier::Handle(handle)) .await .unwrap(), did ); }
async fn spawn_mutable_doc_server(body: String) -> (std::net::SocketAddr, Arc<RwLock<String>>) { let current = Arc::new(RwLock::new(body)); let served = current.clone(); let app = Router::new().fallback(move || { let served = served.clone(); async move { let body = served.read().await.clone(); ([(header::CONTENT_TYPE, "application/json")], body).into_response() } }); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); (addr, current) }
async fn spawn_doc_server(body: String) -> std::net::SocketAddr { spawn_mutable_doc_server(body).await.0 }
fn test_public_client(address: std::net::SocketAddr) -> PublicHttpClient { let client = reqwest::Client::builder() .no_proxy() .resolve("public.example", address) .build() .unwrap(); PublicHttpClient::test(client) }
fn plc_resolver(address: std::net::SocketAddr) -> Resolver { Resolver::with_plc_urls_and_client( vec![Url::parse("http://public.example/").unwrap()], 10, test_public_client(address), ) }
fn did_doc_json(did: &Did, extra: serde_json::Value) -> String { let mut doc = serde_json::json!({ "@context": ["https://www.w3.org/ns/did/v1"], "id": did.as_str(), "alsoKnownAs": ["at://alice.test"], }); let object = doc.as_object_mut().unwrap(); for (key, value) in extra.as_object().unwrap() { object.insert(key.clone(), value.clone()); } doc.to_string() }
fn pds_service(endpoint: &str) -> serde_json::Value { serde_json::json!({ "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": endpoint }] }) }
#[test] fn did_web_urls_follow_the_method_spec() { let url = |did: &str| { did_web_url(&Did::new_owned(did).unwrap()) .unwrap() .to_string() };
assert_eq!( url("did:web:example.com"), "https://example.com/.well-known/did.json" ); assert_eq!( url("did:web:example.com:user:alice"), "https://example.com/user/alice/did.json" ); assert_eq!( url("did:web:example.com%3A3000"), "https://example.com:3000/.well-known/did.json", "a port is percent-encoded in the method-specific id" ); }
#[tokio::test] async fn a_document_answering_for_another_did_is_rejected() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let other = Did::new_static("did:plc:j5hmlfdrwkvtxm7cjmu7j2is").unwrap(); let body = did_doc_json(&other, pds_service("https://pds.example")); let resolver = plc_resolver(spawn_doc_server(body).await);
let error = resolver.resolve_doc(&did).await.unwrap_err(); assert!( error.to_string().contains("does not match requested"), "unexpected error: {error}" ); }
#[tokio::test] async fn a_document_without_a_pds_still_yields_its_raw_form() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let resolver = plc_resolver(spawn_doc_server(did_doc_json(&did, serde_json::json!({}))).await);
assert!(matches!( resolver.resolve_doc(&did).await.unwrap_err(), ResolverError::MissingPds(_) ));
let (raw, handle) = resolver .resolve_raw_doc(&did) .await .expect("a deactivated account still has a document"); assert_eq!(serde_json::to_value(raw).unwrap()["id"], did.as_str()); assert_eq!(handle.unwrap().as_str(), "alice.test"); }
#[tokio::test] async fn an_undecodable_signing_key_is_a_missing_key() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let mut extra = pds_service("https://pds.example"); extra.as_object_mut().unwrap().insert( "verificationMethod".into(), serde_json::json!([{ "id": format!("{}#atproto", did.as_str()), "type": "Multikey", "controller": did.as_str(), "publicKeyMultibase": "zNotAKey" }]), ); let resolver = plc_resolver(spawn_doc_server(did_doc_json(&did, extra)).await);
let mini = resolver .resolve_doc(&did) .await .expect("a broken key must not hide the PDS endpoint"); assert_eq!(mini.pds.as_str(), "https://pds.example/"); assert!(mini.key.is_none()); assert!( resolver .resolve_signing_key(&did) .await .unwrap_err() .to_string() .contains("no atproto signing key") ); }
#[tokio::test] async fn a_failing_mirror_rolls_to_the_next_plc_source() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let down = spawn_status_server(StatusCode::BAD_GATEWAY).await; let up = spawn_doc_server(did_doc_json(&did, pds_service("https://pds.example"))).await; let client = reqwest::Client::builder() .no_proxy() .resolve("down.example", down) .resolve("up.example", up) .build() .unwrap(); let resolver = Resolver::with_plc_urls_and_client( vec![ Url::parse("http://down.example/").unwrap(), Url::parse("http://up.example/").unwrap(), ], 10, PublicHttpClient::test(client), );
assert_eq!( resolver.resolve_doc(&did).await.unwrap().pds.as_str(), "https://pds.example/" ); }
#[tokio::test] async fn fresh_resolution_bypasses_cache_until_explicitly_installed() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let doc = |pds: &str| { serde_json::json!({ "@context": ["https://www.w3.org/ns/did/v1"], "id": did.as_str(), "alsoKnownAs": ["at://alice.test"], "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": pds }], "verificationMethod": [] }) .to_string() }; let (address, current) = spawn_mutable_doc_server(doc("https://pds-a.example")).await; let resolver = plc_resolver(address);
assert_eq!( resolver.resolve_doc(&did).await.unwrap().pds.as_str(), "https://pds-a.example/" ); *current.write().await = doc("https://pds-b.example");
let fresh = resolver.resolve_doc_fresh(&did).await.unwrap(); assert_eq!(fresh.pds.as_str(), "https://pds-b.example/"); assert_eq!( resolver.resolve_doc(&did).await.unwrap().pds.as_str(), "https://pds-a.example/" );
resolver.cache_doc(did.clone(), fresh).await; assert_eq!( resolver.resolve_doc(&did).await.unwrap().pds.as_str(), "https://pds-b.example/" ); }
#[tokio::test] async fn raw_did_doc_json_contract_and_size_limit() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let body = serde_json::json!({ "@context": ["https://www.w3.org/ns/did/v1"], "id": did.as_str(), "alsoKnownAs": ["at://alice.test"], "service": [{ "id": "#atproto_pds", "type": "AtprotoPersonalDataServer", "serviceEndpoint": "https://pds.example" }], "verificationMethod": [] }) .to_string(); let address = spawn_doc_server(body).await; let resolver = plc_resolver(address);
let (data, handle) = resolver.resolve_raw_doc(&did).await.unwrap(); let json = serde_json::to_value(data).unwrap(); assert_eq!(json["id"], did.as_str()); assert_eq!(json["service"][0]["serviceEndpoint"], "https://pds.example"); assert_eq!(handle.unwrap().as_str(), "alice.test");
let oversized = plc_resolver(spawn_doc_server("x".repeat(MAX_DID_DOC_BYTES + 1)).await); let error = oversized.resolve_raw_doc(&did).await.unwrap_err(); assert!( error .to_string() .contains(&format!("exceeds {MAX_DID_DOC_BYTES} bytes")), "unexpected error: {error}" ); }}