use jacquard_common::types::crypto::PublicKey;
use jacquard_common::types::tid::Tid;
use miette::IntoDiagnostic;
use thiserror::Error;
use crate::ingest::stream::{Commit, RepoOpAction, Sync};
use crate::mst::{AtpCommit, MstNode};
use crate::types::RepoState;
/// describes which size limit was exceeded
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SizeLimitKind {
/// msg.blocks.len() > 2 MiB
BlocksField,
/// msg.ops.len() > 200
OpCount,
/// individual record block > 1 MiB
RecordSize,
}
impl std::fmt::Display for SizeLimitKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::BlocksField => write!(f, "blocks field exceeds 2MiB"),
Self::OpCount => write!(f, "op count exceeds 200"),
Self::RecordSize => write!(f, "record block exceeds 1MiB"),
}
}
}
#[derive(Debug, Error)]
pub enum CommitValidationError {
/// rev is not greater than the last known rev
#[error("stale rev")]
StaleRev,
/// rev timestamp exceeds the clock skew window
#[error("future rev")]
FutureRev,
/// CAR could not be parsed, or a required block is missing
#[error("malformed CAR: {0}")]
MalformedCar(miette::Report),
/// wire message fields differ from the signed commit object
#[error("field mismatch in {field}")]
FieldMismatch { field: &'static str },
/// signature verification failed.
#[error("signature verification failed")]
SigFailure,
/// a block, op count, or record exceeds the ATProto size limits
#[error("size limit exceeded: {0}")]
SizeLimitExceeded(SizeLimitKind),
/// MST inversion check failed (checked with verify_mst or when the signature was verified)
#[error("MST inversion failed: {0}")]
MstInvalid(miette::Report),
/// commit arrived from a host that is not the authoritative PDS for this DID
/// (enforced in phase 2 relay worker)
#[allow(dead_code)]
#[error("commit from wrong host")]
WrongHost,
}
#[derive(Debug, Error)]
pub enum SyncValidationError {
/// blocks field exceeds 2MiB
#[error("size limit exceeded")]
SizeLimitExceeded,
/// CAR could not be parsed
#[error("malformed CAR: {0}")]
MalformedCar(miette::Report),
#[error("future rev")]
FutureRev,
/// wire message fields differ from the signed commit object
#[error("field mismatch in {field}")]
FieldMismatch { field: &'static str },
/// signature verification failed
#[error("signature verification failed")]
SigFailure,
}
/// indicates that the commit's chain pointers do not match the last known repo state.
/// this is not a hard rejection so callers can decide whta they want to do
#[derive(Default, Debug)]
pub struct ChainBreak {
/// msg.since is present and does not match the last known rev
pub since_mismatch: bool,
/// msg.prev_data does not match the last known data CID
pub prev_data_mismatch: bool,
}
impl ChainBreak {
pub fn is_broken(&self) -> bool {
self.since_mismatch || self.prev_data_mismatch
}
}
/// a successfully validated `#commit` message, carrying pre-parsed data for apply_commit
pub struct ValidatedCommit<'c> {
#[cfg_attr(not(feature = "indexer"), allow(dead_code))]
pub commit: &'c Commit<'c>,
/// the parsed car, kept so apply_commit does not re-parse it
pub parsed_blocks: crate::car::CommitCar,
pub commit_obj: AtpCommit,
pub chain_break: ChainBreak,
}
/// a successfully validated `#sync` message
pub struct ValidatedSync {
pub commit_obj: AtpCommit,
}
pub struct ValidationOptions {
/// clock drift window for future-rev rejection (seconds)
pub rev_clock_skew_secs: i64,
/// run MST inversion validation (expensive)
pub verify_mst: bool,
/// reject CAR blocks whose content does not hash to their claimed CID
pub verify_cids: bool,
}
impl From<&crate::config::Config> for ValidationOptions {
fn from(config: &crate::config::Config) -> Self {
Self {
rev_clock_skew_secs: config.rev_clock_skew_secs,
verify_mst: config.verify_mst,
verify_cids: config.verify_cids,
}
}
}
/// `validate_commit` panics if called outside a tokio runtime context.
pub struct ValidationContext<'a> {
pub opts: &'a ValidationOptions,
}
impl ValidationContext<'_> {
pub fn validate_commit<'c>(
&self,
msg: &'c Commit<'c>,
repo_state: &RepoState,
signing_key: Option<&PublicKey>,
) -> Result<ValidatedCommit<'c>, CommitValidationError> {
validate_commit(msg, repo_state, signing_key, self.opts)
}
pub fn validate_sync(
&self,
msg: &Sync<'_>,
signing_key: Option<&PublicKey>,
) -> Result<ValidatedSync, SyncValidationError> {
validate_sync(msg, signing_key, self.opts)
}
}
/// validate an incoming `#commit` message.
///
/// on success, returns a `ValidatedCommit` carrying pre-parsed data so that
/// `apply_commit` does not need to repeat the work.
///
/// chain-break (since/prevData mismatch) is NOT an error. callers check
/// `validated.chain_break.is_some()` and decide how to respond.
///
/// - `signing_key`: `None` when signature verification is disabled.
///
/// panics if called outside a tokio runtime context.
pub fn validate_commit<'c>(
msg: &'c Commit<'c>,
repo_state: &RepoState,
signing_key: Option<&PublicKey>,
opts: &ValidationOptions,
) -> Result<ValidatedCommit<'c>, CommitValidationError> {
const MAX_BLOCKS_BYTES: usize = 2_097_152; // 2 MiB
const MAX_OPS: usize = 200;
const MAX_RECORD_BYTES: usize = 1_048_576; // 1 MiB
// 1. size limits
if msg.blocks.len() > MAX_BLOCKS_BYTES {
return Err(CommitValidationError::SizeLimitExceeded(
SizeLimitKind::BlocksField,
));
}
if msg.ops.len() > MAX_OPS {
return Err(CommitValidationError::SizeLimitExceeded(
SizeLimitKind::OpCount,
));
}
// 2. stale rev, skip if msg.rev <= last known rev (lexicographic order)
if let Some(root) = &repo_state.root
&& msg.rev.as_str() <= root.rev.to_tid().as_str()
{
return Err(CommitValidationError::StaleRev);
}
// 3. future rev, reject if rev timestamp is more than clock_skew_secs ahead of now
if is_future_rev(&msg.rev, opts.rev_clock_skew_secs) {
return Err(CommitValidationError::FutureRev);
}
// 4. CAR parse, checking each block's cid as it is read
let parsed = crate::car::parse_commit_car(msg.blocks.clone(), opts.verify_cids)
.map_err(CommitValidationError::MalformedCar)?;
let root_bytes = parsed.blocks.get(&parsed.root).ok_or_else(|| {
CommitValidationError::MalformedCar(miette::miette!("root block missing from CAR"))
})?;
// 5. commit object deserialization
let commit_obj = AtpCommit::from_cbor(root_bytes)
.map_err(|e| CommitValidationError::MalformedCar(miette::miette!("{e}")))?;
// 6. field consistency: wire message vs signed commit object
if commit_obj.did.as_str() != msg.repo.as_str() {
return Err(CommitValidationError::FieldMismatch { field: "repo" });
}
if commit_obj.rev.as_str() != msg.rev.as_str() {
return Err(CommitValidationError::FieldMismatch { field: "rev" });
}
let msg_commit = msg.commit.to_ipld().map_err(|e| {
CommitValidationError::MalformedCar(miette::miette!("invalid commit CID: {e}"))
})?;
if msg_commit != parsed.root {
return Err(CommitValidationError::FieldMismatch { field: "commit" });
}
// 7. signature verification
if let Some(key) = signing_key {
commit_obj
.verify(key)
.map_err(|_| CommitValidationError::SigFailure)?;
}
// 8. chain break checks
let chain_break = repo_state
.root
.as_ref()
.map(|r| breaks_chain(msg, r))
.unwrap_or_default();
// 9–10. per-record size limits and basic CBOR validity
for op in &msg.ops {
let Some(cid_link) = &op.cid else { continue };
let cid = cid_link.to_ipld().map_err(|e| {
CommitValidationError::MalformedCar(miette::miette!("invalid op CID: {e}"))
})?;
let Some(block) = parsed.blocks.get(&cid) else {
return Err(CommitValidationError::MalformedCar(miette::miette!(
"block for op CID {cid} missing from CAR"
)));
};
if block.len() > MAX_RECORD_BYTES {
return Err(CommitValidationError::SizeLimitExceeded(
SizeLimitKind::RecordSize,
));
}
serde_ipld_dagcbor::from_slice::<jacquard_common::Data>(block).map_err(|e| {
CommitValidationError::MalformedCar(miette::miette!("record is not valid CBOR: {e}"))
})?;
}
// 11. MST inversion
if opts.verify_mst || signing_key.is_some() {
verify_mst(msg, &parsed, &commit_obj).map_err(CommitValidationError::MstInvalid)?;
}
Ok(ValidatedCommit {
commit: msg,
parsed_blocks: parsed,
commit_obj,
chain_break,
})
}
pub fn validate_sync<'c>(
msg: &'c Sync<'c>,
signing_key: Option<&PublicKey>,
opts: &ValidationOptions,
) -> Result<ValidatedSync, SyncValidationError> {
const MAX_BLOCKS_BYTES: usize = 2_097_152;
// 1. size limit
if msg.blocks.len() > MAX_BLOCKS_BYTES {
return Err(SyncValidationError::SizeLimitExceeded);
}
// a future rev would stick as the repo's rev and make honest commits after it look stale
if Tid::new(msg.rev.as_ref()).is_ok_and(|rev| is_future_rev(&rev, opts.rev_clock_skew_secs)) {
return Err(SyncValidationError::FutureRev);
}
// 2. CAR parse, checking each block's cid as it is read
let parsed = crate::car::parse_commit_car(msg.blocks.clone(), opts.verify_cids)
.map_err(SyncValidationError::MalformedCar)?;
let root_bytes = parsed.blocks.get(&parsed.root).ok_or_else(|| {
SyncValidationError::MalformedCar(miette::miette!("root block missing from CAR"))
})?;
// 3. commit object deserialization
let commit_obj = AtpCommit::from_cbor(root_bytes)
.map_err(|e| SyncValidationError::MalformedCar(miette::miette!("{e}")))?;
// 4. field consistency
if commit_obj.did.as_str() != msg.did.as_str() {
return Err(SyncValidationError::FieldMismatch { field: "did" });
}
if commit_obj.rev.as_str() != msg.rev.as_str() {
return Err(SyncValidationError::FieldMismatch { field: "rev" });
}
// 5. signature verification
if let Some(key) = signing_key {
commit_obj
.verify(key)
.map_err(|_| SyncValidationError::SigFailure)?;
}
Ok(ValidatedSync { commit_obj })
}
pub(crate) fn is_future_rev(rev: &Tid, skew_secs: i64) -> bool {
let rev_us = rev.timestamp() as i64;
rev_us > chrono::Utc::now().timestamp_micros() + skew_secs * 1_000_000
}
/// whether `msg` picks up where `root` left off, as far as it says. a missing prevData isn't held
/// against it here, unlike in [`breaks_chain`], since this asks again about a commit that went
/// through that check when it came in
#[cfg(feature = "indexer")]
pub(crate) fn follows(msg: &Commit<'_>, root: &crate::types::Commit) -> bool {
let chain = breaks_chain(msg, root);
!chain.since_mismatch && (msg.prev_data.is_none() || !chain.prev_data_mismatch)
}
fn breaks_chain(msg: &Commit<'_>, root: &crate::types::Commit) -> ChainBreak {
// since should equal the rev of the previous commit; only flag when since is present and wrong
let since_mismatch = msg
.since
.as_ref()
.map(|since| since.as_str() != root.rev.to_tid().as_str())
.unwrap_or(false);
// prev_data must equal the last known data CID when both are present
let prev_data_mismatch = match &msg.prev_data {
Some(prev_link) => match prev_link.to_ipld() {
Ok(cid) => cid != root.data,
Err(_) => true, // unparseable CID is a mismatch
},
None => true, // no prev_data but we have a previous state is a chain break
};
ChainBreak {
since_mismatch,
prev_data_mismatch,
}
}
/// apply the inverse of each op (in reverse order) to the new MST and verify the resulting root
/// equals `msg.prev_data`. called when `opts.verify_mst` is set or the signature was verified.
fn verify_mst(
msg: &Commit<'_>,
parsed: &crate::car::CommitCar,
commit_obj: &AtpCommit,
) -> miette::Result<()> {
let blocks = parsed.blocks.map();
let mut mst = MstNode::load(commit_obj.data);
for op in msg.ops.iter().rev() {
let key = op.path.as_ref();
let ok = match &op.action {
RepoOpAction::Create => {
let cid_link = op
.cid
.as_ref()
.ok_or_else(|| miette::miette!("create op missing CID"))?;
let expected = cid_link.to_ipld().into_diagnostic()?;
mst.delete(blocks, key)
.is_ok_and(|found| found == expected)
}
RepoOpAction::Update => {
let cid_link = op
.cid
.as_ref()
.ok_or_else(|| miette::miette!("update op missing CID"))?;
let Some(prev_link) = op.prev.as_ref() else {
// prev is optional in inductive firehose (v3); skip if absent
continue;
};
let expected = cid_link.to_ipld().into_diagnostic()?;
let prev = prev_link.to_ipld().into_diagnostic()?;
if mst.get(blocks, key).ok().flatten() == Some(expected) {
if expected != prev {
mst.add(blocks, key, prev)?;
}
true
} else {
false
}
}
RepoOpAction::Delete => {
let Some(prev_link) = op.prev.as_ref() else {
// prev is optional in inductive firehose (v3); skip if absent
continue;
};
let prev = prev_link.to_ipld().into_diagnostic()?;
if mst.get(blocks, key).ok().flatten().is_some() {
false
} else {
mst.add(blocks, key, prev)?;
true
}
}
RepoOpAction::Other(action) => {
return Err(miette::miette!("unknown op action: {action}"));
}
};
if !ok {
return Err(miette::miette!(
"MST inversion inconsistent with tree state for op on {}",
op.path
));
}
}
// verify the resulting root CID equals prev_data (skip for genesis commits)
if let Some(prev_link) = &msg.prev_data {
let expected = prev_link.to_ipld().into_diagnostic()?;
let root_cid = mst.root_cid(blocks)?;
if root_cid != expected {
return Err(miette::miette!(
"MST inversion root mismatch: expected {expected}, got {root_cid}"
));
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ingest::stream::types::Datetime;
use jacquard_common::types::cid::CidLink;
use jacquard_common::types::string::Did;
#[test]
fn default_config_rejects_forged_car_blocks() {
let opts = ValidationOptions::from(&crate::config::Config::default());
let rt = tokio::runtime::Runtime::new().unwrap();
let _guard = rt.enter();
let claimed = crate::car::compute_cid(b"trusted");
let blocks: bytes::Bytes = rt
.block_on(async {
let mut car = Vec::new();
let header = crate::car::CarHeader::new_v1(vec![claimed]);
let mut writer = crate::car::CarWriter::new(header, &mut car);
writer.write(claimed, b"forged".to_vec()).await.unwrap();
writer.finish().await.unwrap();
car
})
.into();
let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap();
let rev = "3l2abcdefgh22";
let time = Datetime(chrono::Utc::now().fixed_offset());
let sync = Sync {
blocks: blocks.clone(),
did: did.clone(),
rev: rev.into(),
seq: 1,
time: time.clone(),
};
let Err(SyncValidationError::MalformedCar(err)) = validate_sync(&sync, None, &opts) else {
panic!("forged #sync block was accepted");
};
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
let commit = Commit {
blobs: Vec::new(),
blocks,
commit: CidLink::from(claimed),
ops: Vec::new(),
prev_data: None,
rebase: false,
repo: did,
rev: rev.parse().unwrap(),
seq: 2,
since: None,
time,
too_big: false,
};
let Err(CommitValidationError::MalformedCar(err)) =
validate_commit(&commit, &RepoState::synced(), None, &opts)
else {
panic!("forged #commit block was accepted");
};
assert!(err.to_string().contains("CAR block CID mismatch"), "{err}");
}
#[test]
fn a_sync_dated_past_the_clock_is_rejected_before_its_car_is_read() {
let opts = ValidationOptions::from(&crate::config::Config::default());
let sync = |rev: &str| Sync {
blocks: bytes::Bytes::new(),
did: Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap(),
rev: rev.to_owned().into(),
seq: 1,
time: Datetime(chrono::Utc::now().fixed_offset()),
};
assert!(matches!(
validate_sync(&sync("jzzzzzzzzzzzz"), None, &opts),
Err(SyncValidationError::FutureRev)
));
// the rev is fine here, it's the empty car that fails
assert!(matches!(
validate_sync(&sync(Tid::now_0().as_str()), None, &opts),
Err(SyncValidationError::MalformedCar(_))
));
}
#[test]
fn record_dag_cbor_projection_accepts_data_and_rejects_invalid_bytes() {
let encoded = serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": "app.bsky.feed.post",
"text": "stinkpot",
}))
.unwrap();
let decoded = serde_ipld_dagcbor::from_slice::<jacquard_common::Data>(&encoded).unwrap();
assert!(matches!(decoded, jacquard_common::Data::Object(_)));
assert!(
serde_ipld_dagcbor::from_slice::<jacquard_common::Data>(&[0xff]).is_err(),
"break marker is not a standalone record value"
);
}
}