use 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; 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, identity_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. const HANDLE_VERIFY_TTL: Duration = Duration::from_secs(30); /// how long a failed handle verdict is reused. it lives apart from the good ones, which are /// checked first, so it never hides a success, only saves asking again for a handle that keeps /// failing const HANDLE_UNVERIFIED_TTL: Duration = Duration::from_secs(60); trait HandleDns: Send + Sync { fn resolve(&self, handle: Handle) -> BoxFuture<'static, Option>; } 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> { 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 { let value = txt .txt_data .iter() .flat_map(|part| part.iter().copied()) .collect::>(); 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> { 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), #[error(transparent)] NotFound(DocumentNotFound), #[error(transparent)] NoSigningKey(NoSigningKeyError), } impl From for ResolverError { fn from(report: miette::Report) -> Self { ResolverError::Generic(report) } } /// what a failed resolve says about the identity, as opposed to about the source we asked #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum ResolveFailure { /// the method's authority says the did has no document NotFound, /// it says the did had one and doesn't any more, for good (a 410, a plc tombstone) Tombstoned, MissingPds, NoSigningKey, Ratelimited, /// the source timed out, refused, or answered with a server error Unreachable, Other, } impl ResolverError { pub(crate) fn failure(&self) -> ResolveFailure { match self { Self::Ratelimited => ResolveFailure::Ratelimited, Self::Transport(_) => ResolveFailure::Unreachable, Self::MissingPds(_) => ResolveFailure::MissingPds, Self::NotFound(DocumentNotFound { gone: true, .. }) => ResolveFailure::Tombstoned, Self::NotFound(_) => ResolveFailure::NotFound, Self::NoSigningKey(_) => ResolveFailure::NoSigningKey, Self::Generic(_) | Self::HandleResolutionExhausted(_) => ResolveFailure::Other, } } } /// a 404 or 410 from an identity source #[derive(Debug, Diagnostic, Error)] #[error("identity source {url} has no document")] pub struct DocumentNotFound { url: Url, /// a 410: it had one and it's gone for good. a 404 can also mean a mirror that hasn't /// caught up yet gone: bool, } #[derive(Clone, Debug)] pub struct MiniDoc { pub pds: Url, pub handle: Option, pub key: Option>, } #[derive(Clone)] enum HandleSource { HttpsWellKnown, Xrpc(Url), } #[derive(Deserialize)] struct ResolveHandleOutput { did: Did, } struct ResolverInner { plc_urls: Vec, handle_sources: Vec, documents: PublicHttpClient, handles: PublicHttpClient, handle_dns: Arc, next_idx: AtomicUsize, cache: HashCache, /// when a positive `(did, handle)` verdict stops being reused. verified_handles: HashCache<(Did, Handle), Instant>, /// when a negative one stops being reused. unverified_handles: HashCache<(Did, Handle), Instant>, } #[derive(Clone)] pub struct Resolver { inner: Arc, } fn production_handle_sources() -> Vec { vec![ HandleSource::HttpsWellKnown, HandleSource::Xrpc( Url::parse(PUBLIC_HANDLE_RESOLVER).expect("public resolver URL is valid"), ), ] } impl Resolver { pub fn new(plc_urls: Vec, identity_cache_size: u64) -> Self { Self::with_clients( plc_urls, production_handle_sources(), identity_cache_size, identity_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, 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, handle_sources: Vec, identity_cache_size: u64, documents: PublicHttpClient, handles: PublicHttpClient, handle_dns: Arc, ) -> 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), unverified_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 { 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 { 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 { Self::decode_mini_doc(did, &self.fetch_document(did).await?) } /// the pds of a cached doc, without going to the network for it. #[cfg(feature = "indexer")] pub(crate) fn cached_pds(&self, did: &Did) -> Option { self.inner .cache .get_sync(did) .map(|entry| entry.get().pds.clone()) } 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), ResolverError> { let mini = self.resolve_doc(did).await?; Ok((mini.pds, mini.handle)) } pub async fn resolve_signing_key( &self, did: &Did, ) -> Result, ResolverError> { let did = did.clone(); let mini = self.resolve_doc(&did).await?; mini.key .ok_or(ResolverError::NoSigningKey(NoSigningKeyError(did))) } /// 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), 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, 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 { 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: atproto_signing_key(&doc), }) } async fn get( &self, client: &PublicHttpClient, url: &Url, max_body_bytes: usize, ) -> Result, 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" )), PublicHttpError::Reqwest(error) if error.is_timeout() => { ResolverError::Transport(format!("identity source {url} timed out").into()) } 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), status @ (StatusCode::NOT_FOUND | StatusCode::GONE) => { Err(ResolverError::NotFound(DocumentNotFound { url: url.clone(), gone: status == StatusCode::GONE, })) } // 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 { 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 { if let Some(did) = self.inner.handle_dns.resolve(handle.clone()).await { return Ok(did); } let (mut ratelimited, mut answered) = (false, 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, // neither is a source we couldn't reach Err(ResolverError::Transport(_)) => {} Err(_) => answered = true, } } if ratelimited { return Err(ResolverError::Ratelimited); } if !answered { return Err(ResolverError::Transport( format!("no handle source answered for {handle}").into(), )); } Err(ResolverError::HandleResolutionExhausted(miette::miette!( "failed to resolve handle {handle}" ))) } async fn resolve_handle_source( &self, source: &HandleSource, handle: &Handle, ) -> Result { 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::(&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. /// /// a lookup that failed or timed out answers `false`, so callers can use /// `handle.invalid` (atproto.com/specs/handle#invalid-handles), but a rate /// limit is an error, since it says nothing about the handle. a verdict /// belongs to one pair, so a renamed account ages out rather than being /// invalidated. a positive one is reused for [`HANDLE_VERIFY_TTL`], and a /// negative one for [`HANDLE_UNVERIFIED_TTL`] only when a source answered, /// because every describeRepo of an account whose handle doesn't resolve /// would otherwise go out to its host again. one that nobody could answer /// isn't kept, it's not confirmed. pub async fn verify_handle(&self, did: &Did, handle: &Handle) -> Result { let key = (did.clone(), handle.clone()); let fresh = |_: &(Did, Handle), expires_at: &Instant| *expires_at > Instant::now(); if self.inner.verified_handles.read_async(&key, fresh).await == Some(true) { return Ok(true); } if self.inner.unverified_handles.read_async(&key, fresh).await == Some(true) { return Ok(false); } let verdict = self .verify_handle_with_timeout(did, handle, HANDLE_VERIFY_TIMEOUT) .await?; // concurrent verdicts of one pair can land in either order, and the // loser only re-dates or overwrites an entry that is about as fresh let (cache, other, ttl) = match verdict { HandleVerdict::Valid => ( &self.inner.verified_handles, &self.inner.unverified_handles, HANDLE_VERIFY_TTL, ), HandleVerdict::Invalid => ( &self.inner.unverified_handles, &self.inner.verified_handles, HANDLE_UNVERIFIED_TTL, ), HandleVerdict::Unknown => return Ok(false), }; other.remove_async(&key).await; let _ = cache.entry_async(key).await.put_entry(Instant::now() + ttl); Ok(verdict == HandleVerdict::Valid) } async fn verify_handle_with_timeout( &self, did: &Did, handle: &Handle, timeout: Duration, ) -> Result { Self::handle_verdict(did, self.resolve_handle_with_timeout(handle, timeout).await) } fn handle_verdict( did: &Did, result: Result, ) -> Result { match result { Ok(resolved_did) if resolved_did.as_str() == did.as_str() => Ok(HandleVerdict::Valid), Ok(_) | Err(ResolverError::HandleResolutionExhausted(_)) => Ok(HandleVerdict::Invalid), Err(ResolverError::Transport(_)) => Ok(HandleVerdict::Unknown), Err(error) => Err(error), } } } /// what a handle lookup says about one did and handle pair #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum HandleVerdict { Valid, /// a source answered, and the handle points at another did or nowhere Invalid, /// no source could answer, it timed out or failed on the way Unknown, } /// the account's atproto signing key: the first `verificationMethod` whose id /// ends in `#atproto`, typed `Multikey` and controlled by the DID itself /// (https://atproto.com/specs/did). jacquard-common's `atproto_public_key` takes /// the first `Multikey` whatever its id, so a document listing a rotated-out /// `#atproto_old` first got its commits checked against the old key. an /// undecodable key is a missing key. fn atproto_signing_key(doc: &DidDocument) -> Option> { let did = doc.id.as_str(); doc.verification_method .as_ref()? .iter() .find(|m| { m.id.ends_with("#atproto") && m.r#type == "Multikey" && m.controller.as_ref().is_none_or(|c| c == did) }) .and_then(|m| m.public_key_multibase.as_ref()) .and_then(|key| PublicKey::decode(key).ok()) } /// parses a DID document and rejects one that answers for another DID. fn parse_document(did: &Did, body: &[u8]) -> Result { 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 { 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 } pub(crate) fn did_web_url(did: &Did) -> Result { 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::>(); 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(pub(crate) 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, } struct FixedHandleDns { did: Option, calls: Arc, } impl HandleDns for FixedHandleDns { fn resolve(&self, _handle: Handle) -> BoxFuture<'static, Option> { 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> { 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, } async fn resolve_handle(State(counters): State) -> axum::Json { 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) { 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) -> 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_eq!(verdict, HandleVerdict::Unknown); } #[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), 2, "a negative verdict is reused too" ); } #[tokio::test] async fn an_unresolvable_handle_is_asked_once_per_ttl() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let (address, calls) = spawn_counting_status_server(StatusCode::NOT_FOUND).await; let resolver = xrpc_handle_resolver(address); for _ in 0..10 { assert!(!resolver.verify_handle(&did, &handle).await.unwrap()); } assert_eq!(calls.load(Ordering::SeqCst), 1); } /// a source that's down says nothing about the handle, so the next ask goes out again #[tokio::test] async fn an_unreachable_handle_source_is_asked_again() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let (address, calls) = spawn_counting_status_server(StatusCode::BAD_GATEWAY).await; let resolver = xrpc_handle_resolver(address); for _ in 0..2 { assert!(!resolver.verify_handle(&did, &handle).await.unwrap()); } assert_eq!( calls.load(Ordering::SeqCst), 2, "an outage was kept as a verdict" ); } async fn spawn_counting_status_server( status: StatusCode, ) -> (std::net::SocketAddr, Arc) { let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); let counter = calls.clone(); let app = Router::new().fallback(move || { counter.fetch_add(1, Ordering::SeqCst); 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, calls) } 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 )); } /// a 404 may be a mirror that's behind, a 410 is plc saying the did is tombstoned #[tokio::test] async fn a_missing_document_is_a_typed_not_found() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); for (status, failure) in [ (StatusCode::NOT_FOUND, ResolveFailure::NotFound), (StatusCode::GONE, ResolveFailure::Tombstoned), ] { let resolver = plc_resolver(spawn_status_server(status).await); assert_eq!( resolver .resolve_doc_fresh(&did) .await .unwrap_err() .failure(), failure, "{status}" ); } let resolver = plc_resolver(spawn_status_server(StatusCode::BAD_GATEWAY).await); assert_eq!( resolver .resolve_doc_fresh(&did) .await .unwrap_err() .failure(), ResolveFailure::Unreachable ); } #[tokio::test] async fn a_missing_key_is_typed() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let body = did_doc_json(&did, pds_service("https://pds.example")); let resolver = plc_resolver(spawn_doc_server(body).await); assert_eq!( resolver .resolve_signing_key(&did) .await .unwrap_err() .failure(), ResolveFailure::NoSigningKey ); } #[tokio::test] async fn a_document_host_that_never_answers_is_unreachable() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let (silent, _) = spawn_silent_listener().await; let client = reqwest::Client::builder() .no_proxy() .resolve("public.example", silent) .timeout(Duration::from_millis(200)) .build() .unwrap(); let resolver = Resolver::with_plc_urls_and_client( vec![Url::parse("http://public.example/").unwrap()], 10, PublicHttpClient::test(client), ); let error = resolver.resolve_doc_fresh(&did).await.unwrap_err(); assert_eq!(error.failure(), ResolveFailure::Unreachable); assert!(error.to_string().contains("timed out"), "{error}"); } /// a rate limit says nothing about the handle, so it's no verdict, and it isn't kept either #[tokio::test] async fn a_rate_limited_handle_source_is_an_error_not_an_invalid_handle() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let handle = Handle::new_static("fast.test").unwrap(); let (address, calls) = spawn_counting_status_server(StatusCode::TOO_MANY_REQUESTS).await; let resolver = xrpc_handle_resolver(address); for _ in 0..2 { assert!(matches!( resolver.verify_handle(&did, &handle).await.unwrap_err(), ResolverError::Ratelimited )); } assert_eq!( calls.load(Ordering::SeqCst), 2, "a rate limit was kept as a verdict" ); // resolving the handle itself still can't call a rate limit an absent mapping assert!(matches!( resolver .resolve_did(&AtIdentifier::Handle(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>) { 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 the_signing_key_is_the_atproto_entry_not_the_first_multikey() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let old = "zQ3shNzzLK25imF9m6tjPYgSLpAHBH1fzUtdkePghss94jwwZ"; let current = "zQ3shQcbUU2kNp1GqhhuSuV1ykhAUtrCdCB2RYCj28kZbEVER"; let mut extra = pds_service("https://pds.example"); extra.as_object_mut().unwrap().insert( "verificationMethod".into(), serde_json::json!([ { "id": format!("{}#atproto_old", did.as_str()), "type": "Multikey", "controller": did.as_str(), "publicKeyMultibase": old }, { "id": format!("{}#atproto", did.as_str()), "type": "Multikey", "controller": did.as_str(), "publicKeyMultibase": current } ]), ); let resolver = plc_resolver(spawn_doc_server(did_doc_json(&did, extra)).await); let key = resolver.resolve_signing_key(&did).await.unwrap(); assert_eq!(key.bytes, PublicKey::decode(current).unwrap().bytes); } #[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}" ); } }