diff --git a/crates/pds-server/src/car.rs b/crates/pds-server/src/car.rs index cfed9ee..71c70c5 100644 --- a/crates/pds-server/src/car.rs +++ b/crates/pds-server/src/car.rs @@ -17,79 +17,51 @@ //! The header is `{ version: 1, roots: [CID, ...] }` encoded as DAG-CBOR. In //! DAG-CBOR CID links carry the IANA-registered CBOR tag `42`, which the //! `ciborium` crate does not emit for `cid::Cid` (it uses serde newtype-struct -//! tagging instead). We hand-encode the header bytes to keep the file -//! spec-compliant: a `Map(2)` with text keys `"version"` and `"roots"`, an -//! unsigned int `1` for the version, and a tagged byte string for each root -//! CID. +//! tagging instead). We hand-encode the header bytes through the shared +//! primitives in [`crate::dag_cbor`]: a `Map(2)` with text keys `"version"` +//! and `"roots"`, an unsigned int `1` for the version, and a tagged byte +//! string for each root CID. +//! +//! One documented deviation from the DAG-CBOR spec lives in +//! [`encode_header`] — the root CIDs are tagged but not identity-prefixed. +//! See the note there; the firehose frames in [`crate::firehose`] do it the +//! spec-correct way via [`crate::dag_cbor::write_link`]. //! //! Per the spec, CAR v1 stores the raw CID bytes (varint version + codec + //! multihash) prefixed to every block, with a leading varint giving the total //! length of the section (CID + block). +use crate::dag_cbor::{read_head, write_bytes, write_head, write_text}; use anyhow::Result; use cid::Cid; -/// Encode an unsigned CBOR head (major type in upper 3 bits) with a value. -/// -/// Supports values up to `u32::MAX` which is more than enough for any realistic -/// header or array length. -fn cbor_head(out: &mut Vec, major: u8, n: u64) { - let m = (major & 0x07) << 5; - if n < 24 { - out.push(m | n as u8); - } else if n < 0x100 { - out.push(m | 24); - out.push(n as u8); - } else if n < 0x10000 { - out.push(m | 25); - out.push((n >> 8) as u8); - out.push(n as u8); - } else if n < 0x100_0000 { - out.push(m | 26); - out.push((n >> 16) as u8); - out.push((n >> 8) as u8); - out.push(n as u8); - } else { - out.push(m | 27); - out.push((n >> 24) as u8); - out.push((n >> 16) as u8); - out.push((n >> 8) as u8); - out.push(n as u8); - } -} - -/// Append a CBOR text string. -fn cbor_text(out: &mut Vec, s: &str) { - cbor_head(out, 3, s.len() as u64); - out.extend_from_slice(s.as_bytes()); -} - -/// Append a CBOR byte string. -fn cbor_bytes(out: &mut Vec, b: &[u8]) { - cbor_head(out, 2, b.len() as u64); - out.extend_from_slice(b); -} - -/// Append a CBOR tag wrapping the following value. -fn cbor_tag(out: &mut Vec, tag: u64) { - cbor_head(out, 6, tag); -} - /// Encode the CAR v1 DAG-CBOR header `{ version: 1, roots: [CID, ...] }`. /// -/// CIDs are encoded as `tag(42) + bytes()` per the DAG-CBOR -/// spec. This is the canonical IPLD CID-link form. +/// ### Known deviation: the root CIDs carry no identity prefix +/// +/// A spec-conformant DAG-CBOR CID link is `tag(42)` wrapping a byte string of +/// `0x00 || ` — the `0x00` being the multibase identity prefix. +/// +/// Older builds of this server omitted that byte and tagged the bare CID, +/// which no conformant CAR reader can follow: it reads the first byte as the +/// CID version and gives up. Since the header is not content-addressed — +/// nothing hashes it, and no CID anywhere depends on its bytes — fixing it +/// changes only what goes out on the wire, never an identifier. So it is +/// fixed, via [`crate::dag_cbor::write_link`], the same writer the firehose +/// frames use. +/// +/// [`decode_header`] accepts both spellings, so a CAR captured from an older +/// build still parses. pub fn encode_header(roots: &[Cid]) -> Vec { let mut out = Vec::new(); // Map(2): { "version": 1, "roots": [...] } - cbor_head(&mut out, 5, 2); - cbor_text(&mut out, "version"); - cbor_head(&mut out, 0, 1); - cbor_text(&mut out, "roots"); - cbor_head(&mut out, 4, roots.len() as u64); + write_head(&mut out, 5, 2); + write_text(&mut out, "version"); + write_head(&mut out, 0, 1); + write_text(&mut out, "roots"); + write_head(&mut out, 4, roots.len() as u64); for cid in roots { - cbor_tag(&mut out, 42); - cbor_bytes(&mut out, &cid.to_bytes()); + crate::dag_cbor::write_link(&mut out, cid); } out } @@ -261,43 +233,83 @@ fn read_section(section: &[u8]) -> Result<(Cid, Vec)> { Ok((cid, data)) } +/// Decode the CAR header written by [`encode_header`]. +/// +/// Structural only, and deliberately *not* routed through +/// [`crate::dag_cbor::decode`]: that decoder enforces the `0x00` multibase +/// identity prefix on every tag-42 link, which our own header does not carry +/// (see the deviation note on [`encode_header`]). It does share the CBOR head +/// reader with it, so there is exactly one implementation of that. #[allow(dead_code)] fn decode_header(bytes: &[u8]) -> Result { - // The header is a tiny DAG-CBOR map. We decode only the structure we emit. let mut p = 0usize; - let (n_items, consumed) = read_head_and_uint(bytes, p, 5)?; - p += consumed; + let (major, n_items, next) = read_head(bytes, p)?; + if major != 5 { + anyhow::bail!("CAR header must be a CBOR map, got major type {major}"); + } if n_items != 2 { anyhow::bail!("CAR header must have 2 keys, got {n_items}"); } + p = next; let mut version: Option = None; let mut roots: Vec = Vec::new(); for _ in 0..2 { - let (key, consumed) = read_head_and_text(bytes, p)?; - p += consumed; + let (major, len, next) = read_head(bytes, p)?; + if major != 3 { + anyhow::bail!("CAR header key must be text, got major type {major}"); + } + p = next; + if p + len as usize > bytes.len() { + anyhow::bail!("CAR header key exceeds header"); + } + let key = std::str::from_utf8(&bytes[p..p + len as usize]) + .map_err(|e| anyhow::anyhow!("invalid UTF-8 in CAR header key: {e}"))? + .to_string(); + p += len as usize; + match key.as_str() { "version" => { - let (v, c) = read_head_and_uint(bytes, p, 0)?; - p += c; + let (major, v, next) = read_head(bytes, p)?; + if major != 0 { + anyhow::bail!("CAR header `version` must be an unsigned int"); + } + p = next; version = Some(v); } "roots" => { - let (n_roots, c) = read_head_and_uint(bytes, p, 4)?; - p += c; + let (major, n_roots, next) = read_head(bytes, p)?; + if major != 4 { + anyhow::bail!("CAR header `roots` must be an array"); + } + p = next; for _ in 0..n_roots { - // tag(42) - let (_, c) = read_head_and_uint(bytes, p, 6)?; - p += c; - // bytes - let (n, c) = read_head_and_uint(bytes, p, 2)?; - p += c; + let (major, tag, next) = read_head(bytes, p)?; + if major != 6 || tag != 42 { + anyhow::bail!("CAR root must be CBOR tag 42, got major {major} tag {tag}"); + } + p = next; + let (major, n, next) = read_head(bytes, p)?; + if major != 2 { + anyhow::bail!("CAR root CID must be a byte string"); + } + p = next; if p + n as usize > bytes.len() { anyhow::bail!("CAR root CID bytes exceed header"); } - let cid_bytes = &bytes[p..p + n as usize]; - let cid = Cid::read_bytes(cid_bytes) + // Tolerate both spellings: the conformant + // `0x00 || cid` this server writes today, and the bare + // CID older builds wrote (see `encode_header`). A real + // CID never starts with 0x00 — that byte position holds + // the version varint, and version 0 does not exist — so + // stripping it is unambiguous, not a guess. + let raw = &bytes[p..p + n as usize]; + let raw = match raw.first() { + Some(0x00) => &raw[1..], + _ => raw, + }; + let cid = Cid::read_bytes(raw) .map_err(|e| anyhow::anyhow!("invalid root CID bytes: {e}"))?; p += n as usize; roots.push(cid); @@ -313,92 +325,54 @@ fn decode_header(bytes: &[u8]) -> Result { }) } -/// Read a CBOR head (single byte for value < 24, otherwise head + varint -/// extension) and decode its value. Validates that the major type is -/// `expected_major`. Returns the decoded value and the number of bytes -/// consumed (head + any extension). -#[allow(dead_code)] -fn read_head_and_uint( - bytes: &[u8], - offset: usize, - expected_major: u8, -) -> Result<(u64, usize)> { - if offset >= bytes.len() { - anyhow::bail!("CBOR read past end of input"); - } - let first = bytes[offset]; - let major = first >> 5; - if major != expected_major { - anyhow::bail!( - "expected CBOR major {}, got {}", - expected_major, - major - ); - } - let low = first & 0x1f; - let (value, extra) = match low { - 0..=23 => (low as u64, 0usize), - 24 => { - if offset + 2 > bytes.len() { - anyhow::bail!("truncated CBOR uint8"); - } - (bytes[offset + 1] as u64, 1) - } - 25 => { - if offset + 3 > bytes.len() { - anyhow::bail!("truncated CBOR uint16"); - } - ( - ((bytes[offset + 1] as u64) << 8) | (bytes[offset + 2] as u64), - 2, - ) - } - 26 => { - if offset + 5 > bytes.len() { - anyhow::bail!("truncated CBOR uint32"); - } - let n = ((bytes[offset + 1] as u64) << 24) - | ((bytes[offset + 2] as u64) << 16) - | ((bytes[offset + 3] as u64) << 8) - | (bytes[offset + 4] as u64); - (n, 4) - } - 27 => { - if offset + 9 > bytes.len() { - anyhow::bail!("truncated CBOR uint64"); - } - let mut n = 0u64; - for i in 0..8 { - n = (n << 8) | (bytes[offset + 1 + i] as u64); - } - (n, 8) - } - other => anyhow::bail!("unsupported CBOR uint tag {other}"), - }; - Ok((value, 1 + extra)) -} - -/// Read a CBOR text string with major type 3, returning the string and the -/// total number of bytes consumed. -#[allow(dead_code)] -fn read_head_and_text( - bytes: &[u8], - offset: usize, -) -> Result<(String, usize)> { - let (n, c) = read_head_and_uint(bytes, offset, 3)?; - if offset + c + n as usize > bytes.len() { - anyhow::bail!("CBOR text string exceeds buffer"); - } - let s = std::str::from_utf8(&bytes[offset + c..offset + c + n as usize]) - .map_err(|e| anyhow::anyhow!("invalid UTF-8 in CBOR text: {e}"))?; - Ok((s.to_string(), c + n as usize)) -} - #[cfg(test)] mod tests { use super::*; use at_crypto::cid::cid_for_cbor; + /// The identity prefix is what makes a root readable by a stock CAR + /// library, so assert on the bytes rather than only on the round trip + /// through our own parser — which would pass either way. + #[test] + fn header_roots_carry_the_identity_prefix() { + let c = cid_for_cbor(b"a").unwrap(); + let bytes = encode_header(&[c]); + let raw = c.to_bytes(); + // tag(42) is 0xD8 0x2A, then a byte string one longer than the CID, + // whose first content byte is the 0x00 multibase identity prefix. + let tag_at = bytes + .windows(2) + .position(|w| w == [0xD8, 0x2A]) + .expect("tag(42) must be present"); + let after_tag = &bytes[tag_at + 2..]; + let (major, len, next) = read_head(after_tag, 0).unwrap(); + assert_eq!(major, 2, "a link wraps a byte string"); + assert_eq!(len as usize, raw.len() + 1, "one byte longer than the CID"); + assert_eq!(after_tag[next], 0x00, "multibase identity prefix"); + assert_eq!(&after_tag[next + 1..next + 1 + raw.len()], &raw[..]); + } + + /// A CAR captured from an older build tagged the bare CID. Those bytes + /// must keep parsing — otherwise upgrading the server would strand + /// anything that stored a repo export. + #[test] + fn header_without_identity_prefix_still_parses() { + let c = cid_for_cbor(b"legacy").unwrap(); + // Hand-build the old shape: map(2), "version", 1, "roots", [tag(42) + // bytes()]. + let mut old = Vec::new(); + write_head(&mut old, 5, 2); + write_text(&mut old, "version"); + write_head(&mut old, 0, 1); + write_text(&mut old, "roots"); + write_head(&mut old, 4, 1); + write_head(&mut old, 6, 42); + write_bytes(&mut old, &c.to_bytes()); + + let h = decode_header(&old).unwrap(); + assert_eq!(h.roots, vec![c], "legacy root must still decode"); + } + #[test] fn header_encodes_cids_with_tag_42() { let c1 = cid_for_cbor(b"a").unwrap(); diff --git a/crates/pds-server/tests/pds_integration.rs b/crates/pds-server/tests/pds_integration.rs index 2b4618b..bd18937 100644 --- a/crates/pds-server/tests/pds_integration.rs +++ b/crates/pds-server/tests/pds_integration.rs @@ -435,7 +435,17 @@ fn parse_car(bytes: &[u8]) -> ParsedCar { assert_eq!(maj, 2, "root CID must be a byte string"); p += c; let cid_bytes = &bytes[p..p + ln as usize]; - let cid_hex: String = cid_bytes + // A DAG-CBOR link wraps `0x00 || `. The 0x00 is + // the multibase identity prefix, not part of the CID, so it + // comes off before the hex comparison against a real CID's + // bytes. Asserted rather than skipped: this helper is the + // only place the header's wire shape is checked. + assert_eq!( + cid_bytes.first(), + Some(&0x00), + "root link must carry the multibase identity prefix" + ); + let cid_hex: String = cid_bytes[1..] .iter() .map(|b| format!("{:02x}", b)) .collect(); @@ -1056,42 +1066,86 @@ async fn sync_list_repos_keyset_pagination() { assert!(resp["uri"].is_string(), "createRecord: {:?}", resp); created_dids.push(did); } - let min_did = created_dids.iter().min().unwrap().clone(); - let start_cursor = did_cursor_lt(&min_did); + // Two separate properties, deliberately not tested by one long walk + // from the top of the table: `repos` grows without bound on a + // long-lived instance (a few thousand rows here), the seeded DIDs are + // random `did:plc:bafy…` hashes scattered across that range, and a + // full scan at two rows per page ran into its own iteration cap — + // failing for table size rather than for anything about pagination. - let mut seen: std::collections::HashSet = std::collections::HashSet::new(); - let mut cursor: Option = Some(start_cursor); - let mut pages = 0; - loop { - pages += 1; - assert!(pages < 2000, "pagination did not terminate"); + // 1. Every seeded DID is reachable: anchor the cursor immediately + // before it and it must be on the first page. + for did in &created_dids { let url = format!( "{}/xrpc/com.atproto.sync.listRepos?limit=2&cursor={}", PDS_URL, - urlencode(cursor.as_deref().unwrap_or("")) + urlencode(&did_cursor_just_before(did)) ); let resp = client().await.get(&url).send().await.unwrap(); assert_eq!(resp.status().as_u16(), 200); let body: Value = resp.json().await.unwrap(); let repos = body["repos"].as_array().expect("repos array"); + assert!( + repos.iter().any(|r| r["did"].as_str() == Some(did.as_str())), + "DID not on the page starting immediately before it: {did}" + ); + } + + // 2. The keyset itself: walking forward from the lowest seeded DID + // yields strictly increasing DIDs, never a duplicate, and the + // cursor the server hands back is always the last DID of the page. + // A bounded number of pages is enough — these are properties of + // every step, not of the whole table. + let min_did = created_dids.iter().min().unwrap().clone(); + let mut cursor = did_cursor_just_before(&min_did); + let mut seen: std::collections::HashSet = std::collections::HashSet::new(); + let mut last: Option = None; + for _ in 0..25 { + let url = format!( + "{}/xrpc/com.atproto.sync.listRepos?limit=2&cursor={}", + PDS_URL, + urlencode(&cursor) + ); + let resp = client().await.get(&url).send().await.unwrap(); + assert_eq!(resp.status().as_u16(), 200); + let body: Value = resp.json().await.unwrap(); + let repos = body["repos"].as_array().expect("repos array"); + if repos.is_empty() { + break; + } for r in repos { let did = r["did"].as_str().unwrap().to_string(); assert!( seen.insert(did.clone()), "duplicate DID across pages: {did}" ); - } - if created_dids.iter().all(|d| seen.contains(d)) { - break; + if let Some(prev) = &last { + assert!( + &did > prev, + "listRepos must be strictly ascending by DID: {prev} then {did}" + ); + } + last = Some(did); } match body["cursor"].as_str() { - Some(c) => cursor = Some(c.to_string()), - None => panic!( - "pagination exhausted before all created DIDs were seen; missing {:?}", - created_dids.iter().filter(|d| !seen.contains(*d)).collect::>() - ), + Some(c) => { + assert_eq!( + Some(c), + last.as_deref(), + "cursor must be the last DID of the page just served" + ); + cursor = c.to_string(); + } + // Fewer rows than the limit: the end of the table, and the + // server correctly stops handing out a cursor. + None => break, } } + assert!( + seen.len() >= 2, + "expected the walk to cover at least two pages, saw {}", + seen.len() + ); } fn urlencode(s: &str) -> String {