didbot_lookup/

fetch.rs

//! The one round trip behind a cache miss, and what this reader will not
//! make it for.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use didbot_http::{AddressError, PrivateAddresses};
use didbot_identity::DidDocument;
use serde::Deserialize;
use serde_json::Value;
use url::Url;

use crate::record::RecordRef;

/// The most of one record's JSON this reader holds in memory.
///
/// **Reasonable range: 8 KiB to 256 KiB.** The records a gate is asked
/// about are small — a reply gate lists at most five rules, an AI
/// preference four axes — and this bound is what makes the cache's memory
/// arithmetic in [`Limits`](crate::Limits) hold: entries times this.
/// Smaller than [`didbot_http::MAX_JSON_BODY`], which bounds one document
/// held once rather than [`Limits::entries`](crate::Limits::entries) of
/// them held at the same time.
pub const MAX_RECORD_BYTES: usize = 64 * 1024;

/// The atproto error name a repository answers with for "there is no
/// record at that key" -- the one 400 this reader reads as absence rather
/// than as a failure, exactly as `didbot_serve::operator_poll` does.
const RECORD_NOT_FOUND: &str = "RecordNotFound";

/// A future a [`Fetch`] implementation hands back, boxed so the trait stays
/// object-safe and a reader can be held as `Arc<dyn Fetch>`.
pub type FetchFuture<'a> = Pin<Box<dyn Future<Output = Fetched> + Send + 'a>>;

/// Why a record's value is not known.
///
/// Every variant is distinct from [`Answer::Absent`](crate::Answer), which
/// is the repository having answered that it holds nothing at that key.
/// This reader never collapses the two: a rule that refuses on absence
/// would otherwise start refusing the moment somebody else's server went
/// down.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Unknown {
    /// Nothing has fetched this record, so the cache holds no answer for
    /// it. What a read gets before its write's prefetch has run, and again
    /// once an answer has aged out.
    NotFetched,
    /// The fetch did not finish inside its deadline.
    Deadline,
    /// This reader will not fetch from there. Carries why.
    Refused(String),
    /// There is no way to resolve that DID to a server.
    Unresolvable(String),
    /// The repository did not answer, or answered with something other
    /// than a record.
    Unreachable(String),
}

impl Unknown {
    /// A short word for a log line and a counter.
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Unknown::NotFetched => "not-fetched",
            Unknown::Deadline => "deadline",
            Unknown::Refused(_) => "refused",
            Unknown::Unresolvable(_) => "unresolvable",
            Unknown::Unreachable(_) => "unreachable",
        }
    }
}

/// What one round trip established.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Fetched {
    /// The repository holds records there. Always an array, each element
    /// `{"uri": …, "value": …}` — one element for a reference naming one
    /// record, one per record for a reference naming a collection. One
    /// shape means a caller reads both the same way, and the uri is what a
    /// refusal carries so a person can open the record it came from.
    Present(Value),
    /// The repository answered, and holds nothing there: no record at that
    /// key, or no record in that collection.
    Absent,
    /// Neither was established; see [`Unknown`].
    Unknown(Unknown),
}

/// One round trip to a repository this deployment does not hold.
///
/// Separate from the cache so the cache's own behaviour — its bounds, its
/// TTLs, its counters — is testable with no network, and so a deployment
/// that resolves repositories some other way substitutes this rather than
/// the reader.
pub trait Fetch: Send + Sync {
    /// Read `record` from the repository that holds it.
    fn fetch<'a>(&'a self, record: &'a RecordRef) -> FetchFuture<'a>;
}

/// What this deployment already holds, so the reader never fetches it over
/// the network.
///
/// A repository this server hosts is read from its own store, and a name
/// this server answers to is this server. Both are refusals rather than
/// fetches: a DID document that points back here, whether stale or
/// planted, would otherwise turn every foreign read into a request this
/// deployment makes of itself.
pub trait LocalRepositories: Send + Sync {
    /// Whether this deployment holds `did`'s repository.
    fn holds(&self, did: &str) -> bool;
    /// Whether `host` is a name this deployment answers to.
    fn serves(&self, host: &str) -> bool;
}

/// A deployment that holds nothing, for a caller with no registry to ask —
/// a test, or a reader pointed only at the public network.
#[derive(Debug, Clone, Copy, Default)]
pub struct NoLocalRepositories;

impl LocalRepositories for NoLocalRepositories {
    fn holds(&self, _did: &str) -> bool {
        false
    }
    fn serves(&self, _host: &str) -> bool {
        false
    }
}

/// What `com.atproto.repo.getRecord` answers with on success.
#[derive(Debug, Deserialize)]
struct GetRecordResponse {
    value: Value,
}

/// What `com.atproto.repo.listRecords` answers with on success.
#[derive(Debug, Deserialize)]
struct ListRecordsResponse {
    records: Vec<ListedRecord>,
    #[serde(default)]
    cursor: Option<String>,
}

/// One record in a [`ListRecordsResponse`].
#[derive(Debug, Deserialize)]
struct ListedRecord {
    uri: String,
    value: Value,
}

/// The most records one collection read takes from a repository.
///
/// **Reasonable range: 10 to 100.** It is the `limit` the request carries
/// and the page it reads; a repository holding more than this in the
/// collection asked for is answered [`Unknown::Unreachable`] rather than
/// with a part of the set, because a rule that resolved a preference from
/// the first page would be deciding on records it chose not to read. One
/// hundred is atproto's own maximum for the method, and the sets this
/// reader was built for -- a global default and a handful of scopes -- are
/// two orders of magnitude under it.
pub const MAX_COLLECTION_RECORDS: usize = 100;

/// The real fetch: resolve the DID to its `#atproto_pds` endpoint, then
/// `com.atproto.repo.getRecord` there.
///
/// Both hops go through [`didbot_http::pinned_client`], which resolves the
/// host once, refuses it at a private address, pins the connection to what
/// it resolved to and follows no redirect. Both hops read a bounded body.
/// Every URL here is named by somebody else — the DID's own document says
/// which server holds the repository — which is the same posture
/// `didbot_serve::tree::oidc` takes toward an issuer a token names.
pub struct Network {
    private: PrivateAddresses,
    local: Arc<dyn LocalRepositories>,
}

impl Network {
    /// A fetcher that refuses what `local` already holds, and refuses a
    /// private address unless `private` allows one.
    #[must_use]
    pub fn new(local: Arc<dyn LocalRepositories>, private: PrivateAddresses) -> Self {
        Self { private, local }
    }

    /// The `#atproto_pds` endpoint `did`'s document names.
    async fn endpoint(&self, did: &str) -> Result<String, Unknown> {
        let document_url = didbot_identity::document_url(did)
            .map_err(|error| Unknown::Unresolvable(format!("{did}: {error}")))?;
        let body = self
            .get_bounded(&document_url, didbot_http::MAX_JSON_BODY)
            .await?;
        let document: DidDocument = serde_json::from_slice(&body)
            .map_err(|error| Unknown::Unreachable(format!("parsing {document_url}: {error}")))?;
        // The check `didbot_identity::resolve` and the operator poll both
        // make: a document served at the right URL claiming a different
        // subject is a document about somebody else, and the server it
        // names holds somebody else's repository.
        if document.id != did {
            return Err(Unknown::Unreachable(format!(
                "the did document at {document_url} claims {:?}, expected {did:?}",
                document.id
            )));
        }
        let endpoint = document.pds_endpoint().ok_or_else(|| {
            Unknown::Unreachable(format!(
                "{did}'s did document names no #atproto_pds service"
            ))
        })?;
        Ok(endpoint.to_owned())
    }

    /// A GET through a client pinned to `url`'s own addresses, with its
    /// body read against `max_bytes`.
    async fn get_bounded(&self, url: &str, max_bytes: usize) -> Result<Vec<u8>, Unknown> {
        let response = self.send(url).await?;
        let response = response
            .error_for_status()
            .map_err(|error| Unknown::Unreachable(format!("{url} answered: {error}")))?;
        didbot_http::read_bounded(response, max_bytes)
            .await
            .map_err(|error| Unknown::Unreachable(format!("reading {url}: {error}")))
    }

    /// A GET through a client pinned to `url`'s own addresses, with the
    /// status left to the caller.
    async fn send(&self, url: &str) -> Result<reqwest::Response, Unknown> {
        let parsed =
            Url::parse(url).map_err(|_| Unknown::Unreachable(format!("{url:?} is not a URL")))?;
        if let Some(host) = parsed.host_str() {
            if self.local.serves(host) {
                return Err(Unknown::Refused(format!(
                    "{host} is this deployment; a repository it holds is read from its own store"
                )));
            }
        }
        let client = didbot_http::pinned_client(&parsed, self.private)
            .await
            .map_err(|error| match error {
                AddressError::Refused { .. } => Unknown::Refused(error.to_string()),
                other => Unknown::Unreachable(other.to_string()),
            })?;
        client
            .get(parsed)
            .send()
            .await
            .map_err(|error| Unknown::Unreachable(format!("fetching {url}: {error}")))
    }

    /// `com.atproto.repo.listRecords` for every record in `collection` at
    /// `endpoint`, as an array of `{"uri": …, "value": …}`.
    ///
    /// An empty collection is [`Fetched::Absent`], the same answer a key
    /// nothing holds gets: the repository answered, and holds nothing
    /// there.
    async fn list_records(&self, endpoint: &str, collection: &RecordRef) -> Fetched {
        let url = format!(
            "{}/xrpc/com.atproto.repo.listRecords?repo={}&collection={}&limit={MAX_COLLECTION_RECORDS}",
            endpoint.trim_end_matches('/'),
            urlencode(&collection.did),
            urlencode(&collection.collection),
        );
        let body = match self.get_bounded(&url, MAX_RECORD_BYTES).await {
            Ok(body) => body,
            Err(unknown) => return Fetched::Unknown(unknown),
        };
        let listed: ListRecordsResponse = match serde_json::from_slice(&body) {
            Ok(listed) => listed,
            Err(error) => {
                return Fetched::Unknown(Unknown::Unreachable(format!("parsing {url}: {error}")))
            }
        };
        if listed.cursor.is_some() && listed.records.len() >= MAX_COLLECTION_RECORDS {
            return Fetched::Unknown(Unknown::Unreachable(format!(
                "{} holds more than {MAX_COLLECTION_RECORDS} records, which is more of somebody \
                 else's repository than one write reads",
                collection.at_uri()
            )));
        }
        if listed.records.is_empty() {
            return Fetched::Absent;
        }
        Fetched::Present(Value::Array(
            listed
                .records
                .into_iter()
                .map(|record| serde_json::json!({ "uri": record.uri, "value": record.value }))
                .collect(),
        ))
    }

    /// `com.atproto.repo.getRecord` for `record` at `endpoint`, as an array
    /// holding the one `{"uri": …, "value": …}` it found, so a caller reads
    /// one record and a whole collection the same way.
    async fn get_record(&self, endpoint: &str, record: &RecordRef, rkey: &str) -> Fetched {
        let url = format!(
            "{}/xrpc/com.atproto.repo.getRecord?repo={}&collection={}&rkey={}",
            endpoint.trim_end_matches('/'),
            urlencode(&record.did),
            urlencode(&record.collection),
            urlencode(rkey),
        );
        let response = match self.send(&url).await {
            Ok(response) => response,
            Err(unknown) => return Fetched::Unknown(unknown),
        };
        // atproto answers "no such record" as a 400 naming `RecordNotFound`
        // in the error body, and a well-behaved implementation may answer
        // 404 instead, so both can mean absence. Every other 400 is a
        // repository that was reached and refused the request -- a
        // collection it does not know, a proxy rewriting the query -- and
        // reading those as absence would have a rule decide on a fact
        // nobody established.
        let status = response.status().as_u16();
        if status == 404 {
            return Fetched::Absent;
        }
        if status == 400 {
            // Under the bound a record is read under: a body past it names
            // no error, and the answer is unknown rather than absent.
            let body = didbot_http::read_bounded(response, MAX_RECORD_BYTES)
                .await
                .unwrap_or_default();
            let named = serde_json::from_slice::<Value>(&body)
                .ok()
                .and_then(|value| {
                    value
                        .get("error")
                        .and_then(Value::as_str)
                        .map(str::to_owned)
                });
            return match named.as_deref() {
                Some(RECORD_NOT_FOUND) => Fetched::Absent,
                Some(other) => Fetched::Unknown(Unknown::Unreachable(format!(
                    "{url} answered 400 {other}, which is not {RECORD_NOT_FOUND}"
                ))),
                None => Fetched::Unknown(Unknown::Unreachable(format!(
                    "{url} answered 400 with no atproto error name"
                ))),
            };
        }
        let response = match response.error_for_status() {
            Ok(response) => response,
            Err(error) => {
                return Fetched::Unknown(Unknown::Unreachable(format!("{url} answered: {error}")))
            }
        };
        let body = match didbot_http::read_bounded(response, MAX_RECORD_BYTES).await {
            Ok(body) => body,
            Err(error) => {
                return Fetched::Unknown(Unknown::Unreachable(format!("reading {url}: {error}")))
            }
        };
        match serde_json::from_slice::<GetRecordResponse>(&body) {
            Ok(body) => Fetched::Present(Value::Array(vec![serde_json::json!({
                "uri": record.at_uri(),
                "value": body.value,
            })])),
            Err(error) => Fetched::Unknown(Unknown::Unreachable(format!("parsing {url}: {error}"))),
        }
    }
}

impl Fetch for Network {
    fn fetch<'a>(&'a self, record: &'a RecordRef) -> FetchFuture<'a> {
        Box::pin(async move {
            if self.local.holds(&record.did) {
                return Fetched::Unknown(Unknown::Refused(format!(
                    "{} is hosted here; its records are read from this server's own store",
                    record.did
                )));
            }
            let endpoint = match self.endpoint(&record.did).await {
                Ok(endpoint) => endpoint,
                Err(unknown) => return Fetched::Unknown(unknown),
            };
            match &record.rkey {
                Some(rkey) => self.get_record(&endpoint, record, rkey).await,
                None => self.list_records(&endpoint, record).await,
            }
        })
    }
}

/// Percent-encodes everything a query parameter's value may not carry
/// literally, so a DID, an NSID or a record key reaches the remote server
/// as one value rather than as extra parameters.
fn urlencode(value: &str) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.bytes() {
        match byte {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
                encoded.push(byte as char);
            }
            other => encoded.push_str(&format!("%{other:02X}")),
        }
    }
    encoded
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A record key is `*` -- among the characters atproto allows -- and a
    /// DID carries colons. Both have to survive the query string as one
    /// value each.
    #[test]
    fn a_query_value_carries_no_syntax_of_its_own() {
        assert_eq!(urlencode("did:plc:abc"), "did%3Aplc%3Aabc");
        assert_eq!(urlencode("3k-a.b_c~d"), "3k-a.b_c~d");
        assert_eq!(urlencode("a&b=c"), "a%26b%3Dc");
    }

    #[tokio::test]
    async fn a_repository_this_deployment_holds_is_never_fetched() {
        struct Ours;
        impl LocalRepositories for Ours {
            fn holds(&self, did: &str) -> bool {
                did == "did:web:agent.pds.example"
            }
            fn serves(&self, _host: &str) -> bool {
                false
            }
        }
        let network = Network::new(Arc::new(Ours), PrivateAddresses::Refused);
        let record = RecordRef::new(
            "did:web:agent.pds.example",
            "app.bsky.feed.threadgate",
            "3kabc",
        );
        assert!(matches!(
            network.fetch(&record).await,
            Fetched::Unknown(Unknown::Refused(_))
        ));
    }

    #[tokio::test]
    async fn a_did_method_nothing_here_resolves_is_unresolvable_not_unreachable() {
        let network = Network::new(Arc::new(NoLocalRepositories), PrivateAddresses::Refused);
        let record = RecordRef::new("did:example:abc", "app.bsky.feed.threadgate", "3kabc");
        assert!(matches!(
            network.fetch(&record).await,
            Fetched::Unknown(Unknown::Unresolvable(_))
        ));
    }

    /// A `did:web` naming this deployment's own hostname is refused before
    /// anything resolves it, so a document pointing back here cannot turn a
    /// foreign read into a request this server makes of itself.
    #[tokio::test]
    async fn this_deployments_own_name_is_refused() {
        struct Us;
        impl LocalRepositories for Us {
            fn holds(&self, _did: &str) -> bool {
                false
            }
            fn serves(&self, host: &str) -> bool {
                host == "pds.example"
            }
        }
        let network = Network::new(Arc::new(Us), PrivateAddresses::Allowed);
        let record = RecordRef::new("did:web:pds.example", "app.bsky.feed.threadgate", "3kabc");
        assert!(matches!(
            network.fetch(&record).await,
            Fetched::Unknown(Unknown::Refused(_))
        ));
    }
}