verify the handles of everyone the social records name, not only owners
Signed-off-by: dawn <90008@klbr.net>
2 unmodified lines345678910117 unmodified lines19202122222324252627282930313233343559 unmodified lines9596979895991001011021031001041051061071081091101111121021031131141151161171181051191201211225 unmodified lines12812913013111912012113213313413513613713812513914014114214315 unmodified lines1591601611621631641651661671681691701711721731531741751761771571781791801811821831841851861871881891901911921934 unmodified lines19819920020120220320420520620711 unmodified lines219220221222223224225226227228229152 unmodified linesExpand all2 unmodified linesuse std::path::Path;use std::time::Duration;
use anyhow::Context as _;use futures::StreamExt as _;use jacquard_repo::storage::BlockStore as _;use knot_atproto::Atproto;use knot_git::Repo;use knot_record::blocks::GitBlocks;7 unmodified linesuse crate::config::ValidatedConfig;use crate::datastore::{Datastore, schema};use crate::manifest::{GenerationManifest, RepoEntry};use crate::plan::repo_key;use crate::xrpc::social::AVATARS_DIR;use crate::xrpc::social::{ACCOUNTS_FILE, AVATARS_DIR};
const REFRESH_TIMEOUT: Duration = Duration::from_secs(3);const AVATAR_TIMEOUT: Duration = Duration::from_secs(10);const AVATAR_MAX_BYTES: u64 = 1_000_000;const PROFILE_COLLECTIONS: [&str; 2] = ["org.tangled.actor.profile", "app.bsky.actor.profile"];/// owners are verified on every sync, everyone else the social records name this oftenpub(crate) const AUTHOR_REVERIFY_SECS: u64 = 6 * 60 * 60;const REFRESHES_IN_FLIGHT: usize = 16;
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]pub(crate) struct StoredHandles { #[serde(default)]59 unmodified lines let gen_dir = config.output_root.join("generations").join(current); let Ok(manifest) = GenerationManifest::load(&gen_dir) else { return; }; let mut accounts: BTreeSet<String> = manifest let mut owners: BTreeSet<String> = manifest .repos .iter() .map(|entry| entry.owner.clone()) .collect(); accounts.insert(config.owner.as_str().to_string()); owners.insert(config.owner.as_str().to_string()); let store = Datastore::at(&config.output_root); let record = store.record::<schema::OwnerHandles>(&()); let mut stored = record.read().ok().flatten().unwrap_or_default(); let mut authors: BTreeSet<String> = std::fs::read(gen_dir.join(ACCOUNTS_FILE)) .ok() .and_then(|bytes| serde_json::from_slice(&bytes).ok()) .unwrap_or_default(); match crate::admin::keys::accounts(config) { Ok(found) => accounts.extend(found.iter().map(|did| did.as_str().to_string())), Err(error) => tracing::warn!(%error, "read collaborators for the handle cache"), Ok(found) => owners.extend(found.iter().map(|did| did.as_str().to_string())), Err(error) => { tracing::warn!(%error, "read collaborators for the handle cache"); authors.extend(stored.entries.keys().cloned()); } } let owners: BTreeSet<&str> = accounts.iter().map(String::as_str).collect(); let limits = refresh_limits(config); let Ok(transport) = ReqwestHttp::new(limits) else { return; };5 unmodified lines }; let Ok(atproto) = config.atproto(transport) else { return; }; let store = Datastore::at(&config.output_root); let record = store.record::<schema::OwnerHandles>(&()); let mut stored = record.read().ok().flatten().unwrap_or_default(); let before = stored.clone(); let avatars = config.output_root.join(AVATARS_DIR); let now = crate::publish::unix_now(); let named = Named { owners: owners.iter().map(String::as_str).collect(), authors: authors.iter().map(String::as_str).collect(), }; refresh_handles_with(&atproto, &pds, &avatars, &owners, &mut stored, now).await; refresh_handles_with(&atproto, &pds, &avatars, &named, &mut stored, now).await; if stored != before && record.put(&stored).is_ok() { sweep_avatars(&avatars, &stored); }}15 unmodified lines } }}
/// the accounts a generation names. a cached handle for any other is droppedpub(crate) struct Named<'a> { pub owners: BTreeSet<&'a str>, /// everyone else the social records name pub authors: BTreeSet<&'a str>,}
pub(crate) async fn refresh_handles_with<H: HttpTransport, C: Clock, P: HttpTransport>( atproto: &Atproto<H, C>, pds: &P, avatars: &Path, owners: &BTreeSet<&str>, named: &Named<'_>, stored: &mut StoredHandles, now_unix: u64,) { let refreshed = futures::future::join_all(owners.iter().map(|&raw_did| async move { let stale = |did: &&str| { stored.entries.get(*did).is_none_or(|entry| { now_unix.saturating_sub(entry.verified_at_unix) >= AUTHOR_REVERIFY_SECS }) }; let due: BTreeSet<&str> = named .owners .iter() .copied() .chain(named.authors.iter().copied().filter(stale)) .collect(); let refreshed: Vec<_> = futures::stream::iter(due.into_iter().map(|raw_did| async move { let (handle, endpoint) = tokio::time::timeout(REFRESH_TIMEOUT, verify_owner_handle(atproto, raw_did)) .await .ok()4 unmodified lines .ok() .flatten(); Some((raw_did, handle, avatar)) })) .buffer_unordered(REFRESHES_IN_FLIGHT) .collect() .await; for (raw_did, handle, avatar) in refreshed.into_iter().flatten() { let previous = stored.entries.get(raw_did); let (avatar_cid, avatar_ext) = match avatar {11 unmodified lines avatar_ext, }, ); } stored.entries.retain(|did, _| { named.owners.contains(did.as_str()) || named.authors.contains(did.as_str()) });}
async fn verify_owner_handle<H: HttpTransport, C: Clock>( atproto: &Atproto<H, C>,152 unmodified linesExpand all11234567371 unmodified linesExpand all3793803813823813833843853863871 unmodified line38939039139239139339439539639742 unmodified lines44044144244344444544644744844945045113 unmodified lines46546646746846947047147247347447547647747847948010 unmodified lines49149249349449549649749849950050150211 unmodified lines5145155165175025185195205055215225235245255265275285295305315323 unmodified lines5365375385395175405415425435443 unmodified lines548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585319 unmodified linesExpand alluse super::handles::{avatar_of, published_description, refresh_handles_with};use super::handles::{ AUTHOR_REVERIFY_SECS, Named, avatar_of, published_description, refresh_handles_with,};use super::*;use crate::config::IndexConfig;use crate::datastore::{Datastore, schema};use knot_atproto::Atproto;371 unmodified linesExpand all assert!(seen.len() > 5, "kaomoji picks across the table: {seen:?}");}
#[tokio::test]async fn cached_labels_and_refresh_keep_cached_handle_when_transport_fails() {async fn handles_verify_through_dns_or_well_known_and_survive_outages() { use bytes::Bytes; use http::StatusCode; use knot_atproto::PlcDirectory; use knot_runtime::{1 unmodified line SeededEntropy, Signer as _, }; use knot_types::{KnotId, UnixMicros}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
struct StubDns { offline: Arc<AtomicBool>, }42 unmodified lines };
let offline = Arc::new(AtomicBool::new(false)); let net_down = Arc::clone(&offline); let author_lookups = Arc::new(AtomicUsize::new(0)); let make_atproto = || { let net_down = Arc::clone(&net_down); let author_lookups = Arc::clone(&author_lookups); let doc = doc.clone(); let http = FakeHttp::new(move |req| { if net_down.load(Ordering::SeqCst) { return Err(NetworkError::Request("plc down".into()));13 unmodified lines )), "https://plc.directory/did:plc:other" => { Ok(reply(StatusCode::OK, doc("did:plc:other", "other.pet"))) } "https://plc.directory/did:plc:author" => { author_lookups.fetch_add(1, Ordering::SeqCst); Ok(reply(StatusCode::OK, doc("did:plc:author", "author.test"))) } "https://author.test/.well-known/atproto-did" => Ok(reply( StatusCode::OK, Bytes::from_static(b"did:plc:author\n"), )), _ => Ok(reply(StatusCode::NOT_FOUND, Bytes::new())), } }); Atproto::new(10 unmodified lines let root = tempfile::tempdir().unwrap(); let store = Datastore::at(root.path()); let record = store.record::<schema::OwnerHandles>(&()); let owners = BTreeSet::from(["did:plc:verified", "did:plc:unverified", "did:plc:missing"]); let named = Named { owners: owners.clone(), authors: BTreeSet::from(["did:plc:author"]), };
let no_cache = cached_owner_labels(root.path(), &owners); assert_eq!(no_cache["did:plc:verified"], "did:plc:verified"); assert_eq!(no_cache["did:plc:unverified"], "did:plc:unverified");11 unmodified lines ]), Some((StatusCode::OK, PNG.to_vec())), ); let mut stored = StoredHandles::default(); refresh_handles_with(&make_atproto(), &pds, &avatars, &owners, &mut stored, 1_000).await; refresh_handles_with(&make_atproto(), &pds, &avatars, &named, &mut stored, 1_000).await; record.put(&stored).unwrap(); let with_cache = cached_owner_labels(root.path(), &owners); assert_eq!(with_cache["did:plc:verified"], "ptr.pet"); assert_eq!( with_cache["did:plc:verified"], "ptr.pet", "through the TXT record" ); assert_eq!( stored.entries["did:plc:author"].handle, "author.test", "through .well-known" ); assert_eq!(with_cache["did:plc:unverified"], "did:plc:unverified"); assert_eq!(with_cache["did:plc:missing"], "did:plc:missing"); let verified = &stored.entries["did:plc:verified"]; assert_eq!(verified.avatar_cid.as_deref(), Some(cid.as_str()));3 unmodified lines PNG );
offline.store(true, Ordering::SeqCst); refresh_handles_with(&make_atproto(), &pds, &avatars, &owners, &mut stored, 2_000).await; refresh_handles_with(&make_atproto(), &pds, &avatars, &named, &mut stored, 2_000).await; record.put(&stored).unwrap(); let after_failure = cached_owner_labels(root.path(), &owners); assert_eq!( after_failure["did:plc:verified"], "ptr.pet",3 unmodified lines assert_eq!( stored.entries["did:plc:verified"].avatar_cid.as_deref(), Some(cid.as_str()) );
offline.store(false, Ordering::SeqCst); let lookups = author_lookups.load(Ordering::SeqCst); let soon = 1_000 + AUTHOR_REVERIFY_SECS - 1; refresh_handles_with(&make_atproto(), &pds, &avatars, &named, &mut stored, soon).await; assert_eq!( author_lookups.load(Ordering::SeqCst), lookups, "an author isn't verified again before its time" ); let later = 1_000 + AUTHOR_REVERIFY_SECS; refresh_handles_with(&make_atproto(), &pds, &avatars, &named, &mut stored, later).await; assert_eq!(author_lookups.load(Ordering::SeqCst), lookups + 1); assert_eq!(stored.entries["did:plc:author"].verified_at_unix, later);
let unnamed = Named { owners: owners.clone(), authors: BTreeSet::new(), }; refresh_handles_with( &make_atproto(), &pds, &avatars, &unnamed, &mut stored, later, ) .await; assert!(!stored.entries.contains_key("did:plc:author")); assert_eq!(stored.entries["did:plc:verified"].handle, "ptr.pet");}
const PNG: &[u8] = b"\x89PNG\r\n\x1a\n an avatar";
319 unmodified linesExpand all20 unmodified lines21222324252627282930268 unmodified linesExpand all299300301302303304305306307159 unmodified linesExpand all4674684694704714724734744759 unmodified lines48548648748848949049149249349449549649749849950050150250350428 unmodified lines20 unmodified linespub(crate) const AVATARS_WWW: &str = "www/_avatars";const AVATAR_MIMES: [&str; 3] = ["image/png", "image/jpeg", "image/webp"];const DIR: &str = "social";const INVALID_HANDLE: &str = "handle.invalid";/// the accounts the social answers name, which `sync_handles` verifies handles forpub(crate) const ACCOUNTS_FILE: &str = "accounts.json";
const MANIFEST: &str = "org.tangled.repo.manifest";const GIT_REFS: &str = "org.tangled.git.refs";const TICKET: &str = "org.tangled.track.ticket";268 unmodified linesExpand all .tickets .iter() .chain(self.comments.values().flatten()) .chain(self.reactions.values().flatten()) .chain(self.patches.values().flatten()) .chain(&self.labels) .chain(&self.artifacts); let editors = records .flat_map(|record| [record.author.as_deref(), record.editor.as_deref()])159 unmodified linesExpand all rows: BTreeMap::new(), extra: Vec::new(), }; let hosted: Vec<Hosted> = repos.iter().map(|repo| Hosted::of(repo)).collect(); write_accounts(generation_dir, &hosted)?; identity::emit(&mut out, &hosted, social)?; repo::emit(&mut out, &hosted, social)?; ticket::emit(&mut out, &hosted, social)?; let avatars = link_avatars(generation_dir, social);9 unmodified lines atomic_write(&path, maps.to_string().as_bytes()) .with_context(|| format!("write {}", path.display()))}
fn write_accounts(generation_dir: &Path, hosted: &[Hosted]) -> anyhow::Result<()> { let repos: BTreeSet<&str> = hosted.iter().map(|repo| repo.did).collect(); let accounts: BTreeSet<&str> = hosted .iter() .flat_map(Hosted::accounts) .filter(|did| !repos.contains(did) && knot_types::AccountDid::new(*did).is_ok()) .collect(); let path = generation_dir.join(ACCOUNTS_FILE); atomic_write(&path, &serde_json::to_vec(&accounts)?) .with_context(|| format!("write {}", path.display()))}
fn link_avatars(generation_dir: &Path, social: &Social) -> Vec<(String, String)> { let Some(dir) = &social.avatars_dir else { return Vec::new(); };28 unmodified lines8 unmodified lines91011121314151617156 unmodified linesExpand all174175176177177178179180181182103 unmodified linesExpand all286287288289290291292293294295296297298299300301302303394 unmodified linesExpand all8 unmodified linesconst REPO: &str = "did:plc:abcdefghijklmnopqrstuvwx";const OWNER: &str = "did:plc:owner";const ALICE: &str = "did:plc:alice";const BOB: &str = "did:plc:bob";const CAROL: &str = "did:plc:carol";const AVATAR: &str = "bafkreigh2akiscaildc7gnvtklbsfhdgwz72eolmpckbqr5ej26byp3uli";
struct Map { source: String,156 unmodified linesExpand all ), record( "org.tangled.review.patch", "v1", ALICE, CAROL, json!({ "ticket": "p1", "base": { "oid": "b1" }, "head": { "oid": "h1" }, "createdAt": "2024-01-04T00:00:00Z" }), ), record( "org.tangled.review.patch",103 unmodified linesExpand all assert_eq!(keys, json!({ "items": [] }));}
#[test]fn the_generation_lists_every_account_its_social_answers_name() { let served = served(); let accounts: Vec<String> = serde_json::from_slice( &std::fs::read(served.generation.join(crate::xrpc::social::ACCOUNTS_FILE)).unwrap(), ) .unwrap(); assert_eq!(accounts, [ALICE, BOB, CAROL, OWNER]);}
#[test]fn cached_avatars_are_linked_mapped_and_named_in_the_profile() { let served = served(); let actor = |did: &str| format!("at://{did}/org.tangled.actor.profile/self"); let profile = served394 unmodified linesExpand all