Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528use super::FirehoseError;use super::types::{Info, SubscribeReposMessage};#[cfg(feature = "tangled-event")]use super::types::{TANGLED_PUSH_NSID, TangledPushEvent};use serde::{Deserialize, Serialize};
#[derive(Debug, Deserialize, Serialize)]struct EventHeader { op: i64, t: Option<String>,}
#[derive(Deserialize)]struct ErrorFrame { error: String, message: Option<String>,}
pub fn decode_frame<'i>(bytes: &'i [u8]) -> Result<SubscribeReposMessage<'i>, FirehoseError> { let mut de = serde_ipld_dagcbor::de::Deserializer::from_slice(bytes); let header = EventHeader::deserialize(&mut de)?;
match header.op { -1 => { let err = ErrorFrame::deserialize(&mut de)?; return match err.error.as_str() { "FutureCursor" => Err(FirehoseError::FutureCursor { message: err.message, }), _ => Err(FirehoseError::RelayError { error: err.error, message: err.message, }), }; } 1 => {} op => return Err(FirehoseError::UnknownOp(op)), }
let t = header.t.ok_or(FirehoseError::MissingType)?;
let msg = match t.as_str() { "#commit" => SubscribeReposMessage::Commit(Box::new(Deserialize::deserialize(&mut de)?)), "#account" => SubscribeReposMessage::Account(Box::new(Deserialize::deserialize(&mut de)?)), "#identity" => { SubscribeReposMessage::Identity(Box::new(Deserialize::deserialize(&mut de)?)) } "#sync" => SubscribeReposMessage::Sync(Box::new(Deserialize::deserialize(&mut de)?)), "#info" => { let info: Info<'i> = Deserialize::deserialize(&mut de)?; SubscribeReposMessage::Info(Box::new(info)) } #[cfg(feature = "tangled-event")] TANGLED_PUSH_NSID => { // `sig` covers the body verbatim: re-reading the header from the top leaves // `body` on exactly those bytes, which `de` has already consumed past. let mut body = bytes; serde_ipld_dagcbor::de::from_reader_once::<EventHeader, _>(&mut body) .map_err(|err| FirehoseError::Cbor(err.to_string()))?; let mut event: TangledPushEvent<'i> = Deserialize::deserialize(&mut de)?; event.payload = bytes::Bytes::copy_from_slice(body); SubscribeReposMessage::TangledPush(Box::new(event)) } other => return Err(FirehoseError::UnknownType(other.to_string())), };
Ok(msg)}
#[cfg(feature = "relay")]#[derive(Serialize)]struct EncodeHeader<'a> { op: i64, t: &'a str,}
#[cfg(feature = "relay")]#[derive(Serialize)]struct EncodeErrorHeader { op: i64,}
#[cfg(feature = "relay")]#[derive(Serialize)]struct EncodeErrorFrame<'a> { error: &'a str, #[serde(skip_serializing_if = "Option::is_none")] message: Option<&'a str>,}
#[cfg(feature = "relay")]pub fn encode_frame<T: serde::Serialize>(t: &str, msg: &T) -> miette::Result<bytes::Bytes> { encode_frame_body( t, &serde_ipld_dagcbor::to_vec(msg).map_err(|e| miette::miette!("encode_frame body: {e}"))?, )}
#[cfg(feature = "relay")]pub fn encode_frame_body(t: &str, body: &[u8]) -> miette::Result<bytes::Bytes> { let mut buf = serde_ipld_dagcbor::to_vec(&EncodeHeader { op: 1, t }) .map_err(|e| miette::miette!("encode_frame header: {e}"))?; buf.extend_from_slice(body); Ok(bytes::Bytes::from(buf))}
#[cfg(feature = "relay")]pub fn encode_error_frame(error: &str, message: Option<&str>) -> miette::Result<bytes::Bytes> { let mut buf = serde_ipld_dagcbor::to_vec(&EncodeErrorHeader { op: -1 }) .map_err(|e| miette::miette!("encode_error_frame header: {e}"))?; buf.extend_from_slice( &serde_ipld_dagcbor::to_vec(&EncodeErrorFrame { error, message }) .map_err(|e| miette::miette!("encode_error_frame body: {e}"))?, ); Ok(bytes::Bytes::from(buf))}
#[cfg(test)]mod tests { use super::*; use crate::ingest::stream::{Datetime, Identity, InfoName, Sync}; use bytes::Bytes; #[cfg(feature = "tangled-event")] use ipld_core::ipld::Ipld; use jacquard_common::{CowStr, types::string::Did};
fn frame<T: Serialize>(header: EventHeader, body: &T) -> Vec<u8> { let mut bytes = serde_ipld_dagcbor::to_vec(&header).unwrap(); bytes.extend_from_slice(&serde_ipld_dagcbor::to_vec(body).unwrap()); bytes }
#[test] fn firehose_frame_fixture_decodes_identity() { let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); let body = Identity { did, handle: None, seq: 42, time: serde_json::from_str::<Datetime>(r#""2026-08-10T12:34:56Z""#).unwrap(), }; let bytes = frame( EventHeader { op: 1, t: Some("#identity".to_owned()), }, &body, ); assert_eq!( hex::encode(&bytes), "a2617469236964656e74697479626f7001a36364696478206469643a706c633a65777669376e787a796f756e367a687872687336346f697a63736571182a6474696d6574323032362d30382d31305431323a33343a35365a" );
let decoded = decode_frame(&bytes).unwrap(); let SubscribeReposMessage::Identity(identity) = decoded else { panic!("expected identity"); }; assert_eq!(identity.did.as_str(), "did:plc:ewvi7nxzyoun6zhxrhs64oiz"); assert_eq!(identity.seq, 42); assert_eq!(identity.time.0.to_rfc3339(), "2026-08-10T12:34:56+00:00"); assert!(identity.handle.is_none()); }
#[test] fn firehose_frame_mock_decodes_sync_binary_blocks() { let bytes = frame( EventHeader { op: 1, t: Some("#sync".to_owned()), }, &Sync { blocks: Bytes::from_static(b"stinkpot"), did: Did::new_static("did:web:example.com").unwrap(), rev: CowStr::Borrowed("3jzfcijpj2z2a"), seq: 42, time: serde_json::from_str::<Datetime>(r#""2026-08-10T12:34:56Z""#).unwrap(), }, ); assert_eq!( hex::encode(&bytes), "a26174652373796e63626f7001a563646964736469643a7765623a6578616d706c652e636f6d637265766d336a7a6663696a706a327a326163736571182a6474696d6574323032362d30382d31305431323a33343a35365a66626c6f636b73487374696e6b706f74" );
let SubscribeReposMessage::Sync(sync) = decode_frame(&bytes).unwrap() else { panic!("expected sync"); }; assert_eq!(sync.blocks.as_ref(), b"stinkpot"); assert_eq!(sync.did.as_str(), "did:web:example.com"); assert_eq!(sync.rev.as_ref(), "3jzfcijpj2z2a"); assert_eq!(sync.seq, 42); assert_eq!(sync.time.0.to_rfc3339(), "2026-08-10T12:34:56+00:00"); }
#[test] fn firehose_frame_rejects_header_and_relay_errors() { let unknown_op = serde_ipld_dagcbor::to_vec(&EventHeader { op: 7, t: Some("#identity".to_owned()), }) .unwrap(); assert!(matches!( decode_frame(&unknown_op), Err(FirehoseError::UnknownOp(7)) ));
let missing_type = serde_ipld_dagcbor::to_vec(&EventHeader { op: 1, t: None }).unwrap(); assert!(matches!( decode_frame(&missing_type), Err(FirehoseError::MissingType) ));
let unknown_type = serde_ipld_dagcbor::to_vec(&EventHeader { op: 1, t: Some("#unknown".to_owned()), }) .unwrap(); assert!(matches!( decode_frame(&unknown_type), Err(FirehoseError::UnknownType(value)) if value == "#unknown" ));
let relay_error = frame( EventHeader { op: -1, t: None }, &serde_json::json!({ "error": "ConsumerTooSlow", "message": "retry later", }), ); assert!(matches!( decode_frame(&relay_error), Err(FirehoseError::RelayError { error, message }) if error == "ConsumerTooSlow" && message.as_deref() == Some("retry later") )); }
#[test] fn firehose_frame_preserves_outdated_cursor_info() { let bytes = frame( EventHeader { op: 1, t: Some("#info".to_owned()), }, &Info { message: Some("cursor is outside the retained window".into()), name: InfoName::OutdatedCursor, }, ); let SubscribeReposMessage::Info(info) = decode_frame(&bytes).unwrap() else { panic!("expected info"); }; assert_eq!(info.name, InfoName::OutdatedCursor); }
/// a real `org.tangled.git.pushEvent` frame emitted by knot2. regenerate from /// knot-xrpc's `firehose.rs` if the wire shape changes: a disagreement between the /// two repositories about what `sig` covers should fail here, not in production. #[cfg(feature = "tangled-event")] const KNOT_PUSH_FRAME: &str = "a2617478196f72672e74616e676c65642e6769742e707573684576656e74626f7001a7637365711b00060a24181e4000637369675840d1fc7602df9d5e3e2e81f7b966e4dc812531109adefdfeff3ce0691a4fe01ab1018b2ac21e5360f80498529974a443af0c3d12bdca3e0a9c1b7e4ae3c57580bb647265706f6d6469643a706c633a73717569646474696d65781b323032332d31312d31345432323a31333a32302e3030303030305a65656469747383a3636e6577782831313131313131313131313131313131313131313131313131313131313131313131313131313131636f6c64f6646e616d656f726566732f68656164732f6d61696ea3636e6577782833333333333333333333333333333333333333333333333333333333333333333333333333333333636f6c64782832323232323232323232323232323232323232323232323232323232323232323232323232323232646e616d656e726566732f68656164732f646576a3636e6577f6636f6c64782834343434343434343434343434343434343434343434343434343434343434343434343434343434646e616d656f726566732f68656164732f676f6e65667075736865726f6469643a706c633a6f63746f707573676f7074696f6e73816a766572626f73652d6369";
#[cfg(feature = "tangled-event")] const KNOT_PUSH_PUBKEY: &str = "02989c0b76cb563971fdc9bef31ec06c3560f3249d6ee9e5d83c57625596e05f6f";
#[cfg(feature = "tangled-event")] fn knot_push_key() -> jacquard_common::types::crypto::PublicKey<'static> { jacquard_common::types::crypto::PublicKey { codec: jacquard_common::types::crypto::KeyCodec::Secp256k1, bytes: std::borrow::Cow::Owned(hex::decode(KNOT_PUSH_PUBKEY).unwrap()), } }
#[cfg(feature = "tangled-event")] #[test] fn tngl_push_frame_decodes_its_fields_and_keeps_its_body() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); };
assert_eq!(push.repo.as_str(), "did:plc:squid"); assert_eq!(push.pusher.as_str(), "did:plc:octopus"); assert_eq!(push.seq, 1_700_000_000_000_000); assert_eq!(push.time.0.to_rfc3339(), "2023-11-14T22:13:20+00:00"); assert_eq!(push.sig.len(), 64, "secp256k1 signatures are 64 bytes"); assert_eq!( push.options.as_ref().map(|o| o.len()), Some(1), "push options survive decoding" );
let edits = push .edits .iter() .map(|e| { ( e.name.as_ref(), e.old.as_ref().map(|o| &o.as_ref()[..4]), e.new.as_ref().map(|n| &n.as_ref()[..4]), ) }) .collect::<Vec<_>>(); assert_eq!( edits, [ ("refs/heads/main", None, Some("1111")), ("refs/heads/dev", Some("2222"), Some("3333")), ("refs/heads/gone", Some("4444"), None), ] );
assert_eq!( push.payload.as_ref(), &bytes[bytes.len() - push.payload.len()..], "payload must be the tail of the frame, header excluded" ); }
#[cfg(feature = "tangled-event")] #[test] fn tngl_push_signature_verifies_against_the_repo_signing_key() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); };
assert!( crate::ingest::validation::validate_tngl_push(&push, Some(&knot_push_key())).is_ok(), "a frame straight off a knot must verify" ); }
#[cfg(feature = "tangled-event")] #[test] fn tngl_push_signature_rejects_a_tampered_body() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); };
let mut forged = push.clone(); let Ipld::Map(mut map) = serde_ipld_dagcbor::from_slice(&push.payload).unwrap() else { panic!("payload is a map"); }; map.insert( "pusher".to_owned(), Ipld::String("did:plc:impostor".to_owned()), ); forged.payload = bytes::Bytes::from(serde_ipld_dagcbor::to_vec(&Ipld::Map(map)).unwrap());
assert!( matches!( crate::ingest::validation::validate_tngl_push(&forged, Some(&knot_push_key())), Err(crate::ingest::validation::TangledPushValidationError::SigFailure) ), "rewriting a signed field must fail verification" );
let mut wrong_sig = push.clone(); wrong_sig.sig = bytes::Bytes::from(vec![0u8; 64]); assert!(matches!( crate::ingest::validation::validate_tngl_push(&wrong_sig, Some(&knot_push_key())), Err(crate::ingest::validation::TangledPushValidationError::SigFailure) )); }
/// the reason `seq` sits outside the signature: hydrant-as-relay reassigns it. #[cfg(all(feature = "relay", feature = "tangled-event"))] #[test] fn tngl_push_survives_a_relay_reassigning_its_seq() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); };
let reemitted = super::encode_frame_body(TANGLED_PUSH_NSID, &push.with_seq(9_001).unwrap()).unwrap(); let SubscribeReposMessage::TangledPush(relayed) = decode_frame(&reemitted).unwrap() else { panic!("a re-emitted push event is still a push event"); };
assert_eq!(relayed.seq, 9_001, "the relay's sequence is what goes out"); assert_eq!(relayed.repo, push.repo); assert_eq!(relayed.sig, push.sig); assert_eq!( relayed.edits.len(), push.edits.len(), "re-sequencing must not disturb the body" ); assert!( crate::ingest::validation::validate_tngl_push(&relayed, Some(&knot_push_key())).is_ok(), "a consumer downstream of the relay must still be able to verify" ); }
/// a body carrying a field this build does not model must survive re-emission, /// or the signature dies at the first relay that upgrades late. #[cfg(all(feature = "relay", feature = "tangled-event"))] #[test] fn tngl_push_reemission_preserves_unmodelled_fields() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); };
let Ipld::Map(mut map) = serde_ipld_dagcbor::from_slice(&push.payload).unwrap() else { panic!("payload is a map"); }; map.insert("tomorrow".to_owned(), Ipld::String("stinkpot".to_owned())); let mut extended = push.clone(); extended.payload = bytes::Bytes::from(serde_ipld_dagcbor::to_vec(&Ipld::Map(map)).unwrap());
let reemitted = super::encode_frame_body(TANGLED_PUSH_NSID, &extended.with_seq(9_001).unwrap()) .unwrap(); let SubscribeReposMessage::TangledPush(relayed) = decode_frame(&reemitted).unwrap() else { panic!("expected a push event"); }; let Ipld::Map(out) = serde_ipld_dagcbor::from_slice(&relayed.payload).unwrap() else { panic!("payload is a map"); }; assert_eq!( out.get("tomorrow"), Some(&Ipld::String("stinkpot".to_owned())), "an unknown field must be forwarded, not dropped" ); }
#[cfg(feature = "tangled-event")] #[test] fn tngl_push_rejects_bodies_over_the_lexicon_limits() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); }; use crate::ingest::validation::{ TangledPushSizeLimitKind, TangledPushValidationError, validate_tngl_push, };
let mut too_many_edits = push.clone(); too_many_edits.edits = std::iter::repeat_n(push.edits[0].clone(), 201).collect(); assert!(matches!( validate_tngl_push(&too_many_edits, Some(&knot_push_key())), Err(TangledPushValidationError::SizeLimitExceeded( TangledPushSizeLimitKind::EditCount )) ));
let mut too_many_options = push.clone(); too_many_options.options = Some(vec![CowStr::Borrowed("ci"); 11]); assert!(matches!( validate_tngl_push(&too_many_options, Some(&knot_push_key())), Err(TangledPushValidationError::SizeLimitExceeded( TangledPushSizeLimitKind::OptionCount )) ));
let long = "x".repeat(1025); let mut option_too_long = push.clone(); option_too_long.options = Some(vec![CowStr::Owned(long.as_str().into())]); assert!(matches!( validate_tngl_push(&option_too_long, Some(&knot_push_key())), Err(TangledPushValidationError::SizeLimitExceeded( TangledPushSizeLimitKind::OptionSize )) ));
assert!(validate_tngl_push(&push, Some(&knot_push_key())).is_ok()); }
#[cfg(feature = "tangled-event")] #[test] fn tngl_push_without_a_key_skips_verification_but_keeps_the_limits() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); }; let mut forged = push.clone(); forged.sig = bytes::Bytes::from(vec![0u8; 64]);
assert!( crate::ingest::validation::validate_tngl_push(&forged, None).is_ok(), "verification off means the signature is not consulted" );
let mut too_many_edits = push.clone(); too_many_edits.edits = std::iter::repeat_n(push.edits[0].clone(), 201).collect(); assert!( crate::ingest::validation::validate_tngl_push(&too_many_edits, None).is_err(), "size limits hold whether or not signatures are verified" ); }
#[cfg(feature = "tangled-event")] #[test] fn tngl_push_with_an_empty_payload_is_malformed_not_unsigned() { let bytes = hex::decode(KNOT_PUSH_FRAME).unwrap(); let SubscribeReposMessage::TangledPush(push) = decode_frame(&bytes).unwrap() else { panic!("expected a push event"); }; let mut orphaned = push.clone(); orphaned.payload = bytes::Bytes::new();
assert!( matches!( crate::ingest::validation::validate_tngl_push(&orphaned, Some(&knot_push_key())), Err(crate::ingest::validation::TangledPushValidationError::MalformedPayload(_)) ), "a body that cannot be reduced is malformed, not merely unverified" ); }
#[test] fn firehose_frame_maps_future_cursor_error() { let bytes = frame( EventHeader { op: -1, t: None }, &serde_json::json!({ "error": "FutureCursor", "message": "requested cursor is ahead of the relay", }), ); assert!(matches!( decode_frame(&bytes), Err(FirehoseError::FutureCursor { message }) if message.as_deref() == Some("requested cursor is ahead of the relay") )); }}