avatars follow a pds's getBlob redirect, and profiles written under sh.tangled are read
Signed-off-by: dawn <90008@klbr.net>
25 unmodified lines26272829303130313233343536373839404142394 unmodified linesExpand all43743843944043444144244344444544644744834 unmodified lines48348448548647747847948048148248348748848949049149249349449549649749849950050150250350450550650750850951051151295 unmodified lines25 unmodified lines
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 TANGLED_PROFILE: &str = "org.tangled.actor.profile";const PROFILE_COLLECTIONS: [&str; 2] = [TANGLED_PROFILE, "app.bsky.actor.profile"];/// tangled's appview writes the profile under its older sh.tangled nameconst TANGLED_PROFILES: [&str; 2] = ["org.tangled.actor.profile", "sh.tangled.actor.profile"];const PROFILE_COLLECTIONS: [&str; 3] = [ TANGLED_PROFILES[0], TANGLED_PROFILES[1], "app.bsky.actor.profile",];/// a pds may send getBlob on to the host that keeps its blobsconst AVATAR_REDIRECTS: usize = 3;/// 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;
394 unmodified linesExpand all if response.status.is_server_error() { return None; } let record: Value = serde_json::from_slice(&response.body).unwrap_or_default(); if collection == TANGLED_PROFILE && response.status.is_success() { if profile.is_none() && TANGLED_PROFILES.contains(&collection) && response.status.is_success() { profile = record["value"] .is_object() .then(|| super::actor_profile::sanitized(&record["value"])); }34 unmodified lines { return Some(Some((name, ext.to_string()))); } let query = [("did", did), ("cid", name.as_str())]; let response = match pds .execute(xrpc(endpoint, "com.atproto.sync.getBlob", &query)?) .await { Ok(response) => response, Err(NetworkError::TooLarge { .. }) => return Some(None), Err(_) => return None, let mut url = xrpc(endpoint, "com.atproto.sync.getBlob", &query)?.url; let mut hops = 0; let response = loop { let response = match pds.execute(HttpRequest::get(url.clone())).await { Ok(response) => response, Err(NetworkError::TooLarge { .. }) => return Some(None), Err(_) => return None, }; if !response.status.is_redirection() { break response; } let next = response .headers .get(http::header::LOCATION) .and_then(|location| location.to_str().ok()) .and_then(|location| url.join(location).ok()) .filter(|next| next.scheme() == "https"); hops += 1; match next { Some(next) if hops <= AVATAR_REDIRECTS => url = next, _ => return Some(None), } }; if response.status.is_server_error() { return None; }95 unmodified lines744 unmodified linesExpand all745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881547 unmodified linesExpand all744 unmodified linesExpand all let cached = fake_pds(tangled(&cid), None); assert_eq!(avatar(&cached, &avatars, &endpoint).await, fetched);}
/// getBlob answers with `hops` redirects before the blob host answers with `blob`fn redirecting_pds( profile: serde_json::Value, hops: Vec<&'static str>, blob: Vec<u8>,) -> impl knot_runtime::HttpTransport { use http::StatusCode; knot_runtime::FakeHttp::new(move |request| { let url = request.url.as_str().to_string(); let mut headers = http::HeaderMap::new(); let (status, body) = if request.url.path() == "/xrpc/com.atproto.repo.getRecord" { let collection = request .url .query_pairs() .find(|(key, _)| key == "collection") .map(|(_, value)| value.into_owned()); if collection.as_deref() == Some("org.tangled.actor.profile") { ( StatusCode::OK, serde_json::json!({ "value": profile }) .to_string() .into_bytes(), ) } else { (StatusCode::BAD_REQUEST, Vec::new()) } } else { let at = if request.url.path() == "/xrpc/com.atproto.sync.getBlob" { 0 } else { hops.iter() .position(|hop| url.ends_with(hop.trim_start_matches('.'))) .map_or(usize::MAX, |at| at + 1) }; match hops.get(at) { Some(next) => { headers.insert(http::header::LOCATION, next.parse().unwrap()); (StatusCode::FOUND, Vec::new()) } None if at == hops.len() => (StatusCode::OK, blob.clone()), None => (StatusCode::NOT_FOUND, Vec::new()), } }; Ok(knot_runtime::HttpResponse { status, headers, body: body.into(), }) })}
#[tokio::test]async fn a_blob_the_pds_redirects_is_fetched_from_where_it_points() { let root = tempfile::tempdir().unwrap(); let endpoint = url::Url::parse("https://pds.example.test/").unwrap(); let cid = blob_cid(PNG); let fetched = Some(Some((cid.clone(), "png".to_string()))); let cases = [ (vec!["https://blobs.example.test/blocks/a"], fetched.clone()), ( vec!["https://blobs.example.test/blocks/a", "./b"], fetched.clone(), ), (vec!["http://blobs.example.test/blocks/a"], Some(None)), ( vec![ "https://blobs.example.test/1", "https://blobs.example.test/2", "https://blobs.example.test/3", "https://blobs.example.test/4", ], Some(None), ), ]; for (n, (hops, expected)) in cases.into_iter().enumerate() { let avatars = root.path().join(n.to_string()); let pds = redirecting_pds(avatar_ref(&cid), hops.clone(), PNG.to_vec()); assert_eq!( avatar(&pds, &avatars, &endpoint).await, expected, "{hops:?}" ); }}
#[tokio::test]async fn an_avatar_past_the_lexicons_size_is_left_out_wherever_it_comes_from() { let root = tempfile::tempdir().unwrap(); let endpoint = url::Url::parse("https://pds.example.test/").unwrap(); let mut big = PNG.to_vec(); big.resize(1_000_001, 0); let cid = blob_cid(&big); for hops in [vec![], vec!["https://blobs.example.test/blocks/a"]] { let avatars = root.path().join(hops.len().to_string()); let pds = redirecting_pds(avatar_ref(&cid), hops, big.clone()); assert_eq!(avatar(&pds, &avatars, &endpoint).await, Some(None)); assert!(!avatars.exists()); }}
#[tokio::test]async fn a_profile_written_under_sh_tangled_is_read_too() { use http::StatusCode; use serde_json::json; let root = tempfile::tempdir().unwrap(); let avatars = root.path().join("avatars"); let endpoint = url::Url::parse("https://pds.example.test/").unwrap(); let cid = blob_cid(PNG); let mut profile = avatar_ref(&cid); profile["$type"] = json!("sh.tangled.actor.profile"); profile["location"] = json!("Nicaragua"); profile["links"] = json!(["", "", "", "", ""]); let pds = fake_pds( BTreeMap::from([("sh.tangled.actor.profile", profile)]), Some((StatusCode::OK, PNG.to_vec())), ); let found = profile_of(&pds, &avatars, "did:plc:verified", &endpoint).await; assert_eq!( found, Some(Fetched { avatar: Some((cid, "png".to_string())), profile: json!({ "location": "Nicaragua", "links": [] }) .as_object() .cloned(), }) );}
#[tokio::test]async fn the_tangled_profile_is_kept_within_the_lexicon() { use http::StatusCode; use serde_json::json; let root = tempfile::tempdir().unwrap();547 unmodified linesExpand all