diff --git a/hubble-sync/src/commit/commit_object.rs b/hubble-sync/src/commit/commit_object.rs index ed32388..2840444 100644 --- a/hubble-sync/src/commit/commit_object.rs +++ b/hubble-sync/src/commit/commit_object.rs @@ -5,7 +5,7 @@ use jacquard_repo::commit::Commit as JacqCommit; use repo_stream::Commit as RSCommit; use serde::Serialize; -use crate::identity::DidError; +use crate::identity::{DidError, SignatureError, SigningKey}; use crate::tid::TidParseError; use crate::{DaslCid, Did, Tid}; @@ -37,7 +37,10 @@ pub struct CommitObject { /// /// values too far in the future are invalid (not checked here) pub rev: Tid, - /// unused, and will possibly break our commit reconstruction if used + /// usually unused + /// + /// in previous versions `prev` held the previous commit's CID; in v3 it's + /// not needed. someeeeeee producers still set this (eg., bridgy) pub prev: Option, /// cryptographic signature of this commit, as raw bytes pub sig: Vec, @@ -47,6 +50,14 @@ impl CommitObject { pub fn data_ipld(&self) -> IpldCid { IpldCid::read_bytes(self.data.as_bytes()).expect("dasl cid to be valid") } + /// verify `sig` over this commit's canonical unsigned encoding + /// + /// resolving the correct key for the repo is the caller's problem. + pub fn verify_signature(&self, key: &SigningKey) -> Result<(), SignatureError> { + let unsigned: UnsignedCommit = self.into(); + let signed_bytes = drisl::to_vec(&unsigned).expect("vec encode infallible"); + key.verify_bytes(&signed_bytes, &self.sig) + } /// get encoded signed commit bytes pub fn to_drisl_bytes(&self) -> Vec { drisl::to_vec(&SignedCommit { @@ -116,7 +127,7 @@ pub struct UnsignedCommit<'a> { version: u64, // always 3 data: DaslCid, rev: String, // tid isn't string, so not easy to borrow - prev: Option, // always none in v3 but must be present (null) + prev: Option, // must be present (null when unset); some PDSes still set it in v3 } impl<'a> From<&'a CommitObject> for UnsignedCommit<'a> { @@ -138,7 +149,7 @@ struct SignedCommit<'a> { version: u64, // always 3 data: DaslCid, rev: String, - prev: Option, // always null in v3 + prev: Option, // null for most v3 repos, but some PDSes still set it #[serde(with = "serde_bytes")] sig: &'a [u8], } diff --git a/hubble-sync/src/commit/mod.rs b/hubble-sync/src/commit/mod.rs index a0547bb..c54514e 100644 --- a/hubble-sync/src/commit/mod.rs +++ b/hubble-sync/src/commit/mod.rs @@ -1,5 +1,5 @@ mod commit_object; mod commit_view; -pub use commit_object::{CommitConvertError, CommitObject, UnsignedCommit}; +pub use commit_object::{CommitConvertError, CommitObject}; pub use commit_view::{Commit, Op, OpKind}; diff --git a/hubble-sync/src/lib.rs b/hubble-sync/src/lib.rs index 8c6dce8..f6614d9 100644 --- a/hubble-sync/src/lib.rs +++ b/hubble-sync/src/lib.rs @@ -33,7 +33,7 @@ pub use firehose::{ }; pub use host::{Host, HostRegistry, HostRegistryConfig, HostRequestError, Sync11State}; pub use hubble_sync::{HubbleSync, HubbleSyncError}; -pub use identity::{Did, DidMethod, HubbleSyncResolver, Resolve, SigningKey, did}; +pub use identity::{Did, DidMethod, HubbleSyncResolver, Resolve, SignatureError, SigningKey, did}; pub use metrics::{ BIG_REPO_WAIT_SECONDS_BUCKETS, REPO_ACTOR_ALIVE_SECONDS_BUCKETS, RESYNC_COUNT_BUCKETS, RESYNC_DURATION_SECONDS_BUCKETS, RESYNC_PHASE_SECONDS_BUCKETS, RESYNC_SIZE_BUCKETS, diff --git a/hubble-sync/src/repo_actor/task_processor/event_validation.rs b/hubble-sync/src/repo_actor/task_processor/event_validation.rs index e56208d..97dfb77 100644 --- a/hubble-sync/src/repo_actor/task_processor/event_validation.rs +++ b/hubble-sync/src/repo_actor/task_processor/event_validation.rs @@ -1,6 +1,4 @@ -use dasl::drisl; - -use crate::commit::{CommitObject, UnsignedCommit}; +use crate::commit::CommitObject; use crate::identity::{SignatureError, SigningKey}; /// Apply remaining commit event validations @@ -15,7 +13,5 @@ pub fn validate_commit_signature( commit: &CommitObject, ) -> Result<(), SignatureError> { // 4. verify the commit signature - let unsigned: UnsignedCommit = commit.into(); - let signed_bytes = drisl::to_vec(&unsigned).expect("infallible encode"); - key.verify_bytes(&signed_bytes, &commit.sig) + commit.verify_signature(key) } diff --git a/hubble-sync/src/repo_actor/task_processor/mod.rs b/hubble-sync/src/repo_actor/task_processor/mod.rs index 4184f44..d624f05 100644 --- a/hubble-sync/src/repo_actor/task_processor/mod.rs +++ b/hubble-sync/src/repo_actor/task_processor/mod.rs @@ -19,7 +19,7 @@ use super::{Task, WorkingGauge}; use crate::Commit; use crate::firehose::{FirehoseEvent, FirehosePayload, Seq}; use crate::host::{HostRequestError, LimitWaited, ResolvedHost}; -use crate::identity::{ResolutionError, Resolve, ResolvedIdentity, Validity}; +use crate::identity::{ResolutionError, Resolve, ResolvedIdentity, SigningKey, Validity}; use crate::metrics::{ ACCOUNT_OUTCOMES_TOTAL, COMMIT_OUTCOMES_TOTAL, IDENTITY_REFRESH_OUTCOMES_TOTAL, RESYNC_COUNT, RESYNC_DURATION_SECONDS, RESYNC_OUTCOMES_TOTAL, RESYNC_PHASE_SECONDS, RESYNC_SIZE, @@ -70,6 +70,10 @@ impl<'a> RepoContext<'a> { pub fn pds(&self) -> &Arc { &self.0.info().identity.pds_host } + /// the repo's current signing key + pub fn signing_key(&self) -> &SigningKey { + &self.0.info().identity.signing_key + } pub fn is_active(&self) -> bool { self.status().is_active() } diff --git a/hubble/src/main.rs b/hubble/src/main.rs index 851cd1c..7b31019 100644 --- a/hubble/src/main.rs +++ b/hubble/src/main.rs @@ -209,6 +209,10 @@ struct Args { #[arg(long, default_value_t = 1, value_parser = clap::value_parser!(u32).range(0..=9))] gzip_level: u32, + /// verify reconstructed commit signatures for getRepo (just logs errors) + #[arg(long)] + verify_export_sigs: bool, + /// prometheus metrics exporter listen #[arg(long, default_value = "0.0.0.0:9000")] metrics_bind: SocketAddr, @@ -358,6 +362,7 @@ async fn main() -> color_eyre::Result<()> { zstd: args.zstd_level, gzip: args.gzip_level, }, + verify_export_sigs: args.verify_export_sigs, }; let mut tasks: JoinSet> = JoinSet::new(); diff --git a/hubble/src/metrics.rs b/hubble/src/metrics.rs index f562331..65f9039 100644 --- a/hubble/src/metrics.rs +++ b/hubble/src/metrics.rs @@ -37,6 +37,15 @@ pub const GETREPO_EXPORT_RECORDS: &str = "hubble_getrepo_export_records"; /// | `star-lite`, `encoding` = `identity` | `gzip` | `zstd` pub const GETREPO_EXPORT_SECONDS: &str = "hubble_getrepo_export_seconds"; +/// getRepo exports by stored commit-`prev` state. `prev` = `some` | `null` | +/// `unknown` (`unknown` = slot predates prev tracking; the export guesses null) +pub const GETREPO_EXPORT_PREV: &str = "hubble_getrepo_export_prev_total"; + +/// signature checks of reconstructed export commits (only recorded with +/// `--verify-export-sigs`). `result` = `ok` | `fail`, `prev` as in +/// [`GETREPO_EXPORT_PREV`] +pub const GETREPO_EXPORT_SIG_CHECKS: &str = "hubble_getrepo_export_sig_checks_total"; + /// RAII handle on [`HTTP_REQUESTS_IN_FLIGHT`]: increments on start, decrements /// on drop -- so a request future cancelled mid-flight still releases its count pub struct InFlightGuard(Gauge); @@ -87,6 +96,14 @@ pub fn describe_metrics() { GETREPO_EXPORT_SECONDS, "seconds per completed getRepo export incl. compression, format=car|star-lite, encoding=identity|gzip|zstd" ); + describe_counter!( + GETREPO_EXPORT_PREV, + "getRepo exports by stored commit-prev state, prev=some|null|unknown" + ); + describe_counter!( + GETREPO_EXPORT_SIG_CHECKS, + "reconstructed export commit sig checks (--verify-export-sigs), result=ok|fail, prev=some|null|unknown" + ); } /// buckets for hubble's own histograms, for the exporter to install diff --git a/hubble/src/serve/get_repo.rs b/hubble/src/serve/get_repo.rs index d1b2d25..3d0fca1 100644 --- a/hubble/src/serve/get_repo.rs +++ b/hubble/src/serve/get_repo.rs @@ -21,12 +21,15 @@ use std::sync::Arc; use std::time::Instant; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; -use tracing::{debug, error}; +use tracing::{debug, error, warn}; use super::encoding::{CompressedWriter, CompressionLevels, Encoding}; use super::{MEDIA_AVAILABLE, MEDIA_CAR, MEDIA_STAR_LITE, ServerState}; -use crate::metrics::{GETREPO_EXPORT_RECORDS, GETREPO_EXPORT_SECONDS, GETREPO_RESPONSES_TOTAL}; -use crate::store::record; +use crate::metrics::{ + GETREPO_EXPORT_PREV, GETREPO_EXPORT_RECORDS, GETREPO_EXPORT_SECONDS, GETREPO_EXPORT_SIG_CHECKS, + GETREPO_RESPONSES_TOTAL, +}; +use crate::store::{StoredPrev, record}; #[derive(Debug, thiserror::Error)] enum ArchiveBuildError { @@ -134,7 +137,7 @@ pub async fn get_repo( return Err(map_status(&ctx.status())); } - let Some((rev, data, commit)) = super::synced_commit(&view) else { + let Some((rev, data, commit_state)) = super::synced_commit(&view) else { return Err(not_found("repo not synchronized yet")); }; let info_slot = view.info(); @@ -143,14 +146,35 @@ pub async fn get_repo( Some(id) => record::RepoKey::Interned(id), None => record::RepoKey::Did(ctx.did().clone()), }; + // old slots are assumed to have `null` for `prev`, which means their + // reconstructed commit object will fail signature check if it had a value. + let (prev, prev_label) = match commit_state.prev { + StoredPrev::Known(Some(cid)) => (Some(cid), "some"), + StoredPrev::Known(None) => (None, "null"), + StoredPrev::Legacy => (None, "unknown"), + }; + counter!(GETREPO_EXPORT_PREV, "prev" => prev_label).increment(1); let commit = CommitObject { did: did.clone(), data, rev, - prev: None, - sig: commit.signature.clone(), + prev, + sig: commit_state.signature.clone(), }; + if state.verify_export_sigs { + // verifies with the *current* identity key, so a key rotation since + // the last commit can fail this spuriously (we just log it, for now). + let result = match commit.verify_signature(ctx.signing_key()) { + Ok(()) => "ok", + Err(err) => { + warn!(%did, prev = prev_label, %err, "reconstructed export commit failed signature check"); + "fail" + } + }; + counter!(GETREPO_EXPORT_SIG_CHECKS, "result" => result, "prev" => prev_label).increment(1); + } + let content_disposition = archive_format .attachment_disposition(&did, &commit.rev) .inspect_err(|e| debug!("failed to render attachment disposition: {e}")) @@ -237,7 +261,10 @@ fn write_archive( version: 3, data: StarLiteCid::read_bytes(spec.commit.data.as_bytes()).expect("valid dasl cid"), rev: spec.commit.rev.to_string(), - prev: None, + prev: spec + .commit + .prev + .map(|c| StarLiteCid::read_bytes(c.as_bytes()).expect("valid dasl cid")), sig: spec.commit.sig.clone(), }; let b = StarLiteBuilder::new(&mut compressed, &star_commit)?; diff --git a/hubble/src/serve/mod.rs b/hubble/src/serve/mod.rs index ae7386c..b311189 100644 --- a/hubble/src/serve/mod.rs +++ b/hubble/src/serve/mod.rs @@ -39,6 +39,8 @@ pub struct ServerState { pub car_spill_dir: PathBuf, /// compression effort for getRepo responses pub compression: CompressionLevels, + /// verify reconstructed commit signatures on getRepo export (just logs) + pub verify_export_sigs: bool, } /// repo archive media types, negotiable on getRepo and listed by getRepoInfo @@ -62,6 +64,10 @@ fn lex_datetime(t: SystemTime) -> Datetime { } /// all the bits needed to be able to reconstruct and serve a repo archive +/// +/// fun terminology collision: hubble-sync's `prev` is the previous commit's MST +/// root node CID (used for inductive proof), not the commit object's `prev` +/// field (unused in repo v3, used to be the previous *commit object*'s CID). fn synced_commit(view: &RepoView) -> Option<(Tid, DaslCid, &CommitState)> { let ctx = view.context(); match (ctx.rev(), ctx.data(), view.commit()) { diff --git a/hubble/src/store/commit_slot.rs b/hubble/src/store/commit_slot.rs index 64c1aff..5b23b19 100644 --- a/hubble/src/store/commit_slot.rs +++ b/hubble/src/store/commit_slot.rs @@ -1,17 +1,158 @@ -use hubble_sync::{RepoSlot, SlotDecodeError}; +use hubble_sync::{DaslCid, RepoSlot, SlotDecodeError}; +/// per-commit repo state: the rest of what hubble-sync doesn't already track to +/// be able to reconstruct a signed commit object #[derive(Debug, Clone)] pub struct CommitState { pub signature: Vec, + pub prev: StoredPrev, } +/// the commit object's `prev` field +/// +/// *almost* always null in v3 commits, but some PDSes (bridgy) still set it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StoredPrev { + /// slot written before hubble tracked `prev`: value unknown. on export it's + /// assumed to be null, but that's wrong if it the commit had it set. + Legacy, + Known(Option), +} + +// stored signatures are exactly SIG_LEN bytes: both atproto curves (k256, p256) +// use 64-byte signatures. a bare-signature slot (legacy, not tracking prev) +// only encodes the 64 bytes of signature. new slots write a tag first. +const SIG_LEN: usize = 64; +const CID_LEN: usize = 36; + +const TAG_PREV_NULL: u8 = 0x01; +const TAG_PREV_CID: u8 = 0x02; + +const LEN_LEGACY: usize = SIG_LEN; +const LEN_PREV_NULL: usize = 1 + SIG_LEN; +const LEN_PREV_CID: usize = 1 + CID_LEN + SIG_LEN; + impl RepoSlot for CommitState { fn encode(&self) -> Vec { - self.signature.clone() + assert_eq!( + self.signature.len(), + SIG_LEN, + "commit slot signatures must be verified 64-byte compact ECDSA", + ); + let mut out = Vec::with_capacity(LEN_PREV_CID); + match self.prev { + StoredPrev::Legacy => { + unreachable!("legacy commit slots are decode-only, never written back") + } + StoredPrev::Known(None) => out.push(TAG_PREV_NULL), + StoredPrev::Known(Some(cid)) => { + out.push(TAG_PREV_CID); + out.extend_from_slice(cid.as_bytes()); + } + } + out.extend_from_slice(&self.signature); + out } fn decode(bytes: &[u8]) -> Result { - Ok(Self { - signature: bytes.to_vec(), - }) + match bytes.len() { + LEN_LEGACY => Ok(Self { + signature: bytes.to_vec(), + prev: StoredPrev::Legacy, + }), + LEN_PREV_NULL if bytes[0] == TAG_PREV_NULL => Ok(Self { + signature: bytes[1..].to_vec(), + prev: StoredPrev::Known(None), + }), + LEN_PREV_CID if bytes[0] == TAG_PREV_CID => { + let cid = DaslCid::from_bytes_raw(&bytes[1..1 + CID_LEN]) + .map_err(|e| SlotDecodeError(format!("commit slot prev cid: {e}")))?; + Ok(Self { + signature: bytes[1 + CID_LEN..].to_vec(), + prev: StoredPrev::Known(Some(cid)), + }) + } + n => Err(SlotDecodeError(format!( + "commit slot: tag {:?} disagrees with length {n}", + bytes.first(), + ))), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_sig(fill: u8) -> Vec { + vec![fill; SIG_LEN] + } + + fn test_cid() -> DaslCid { + let mut bytes = [0u8; 36]; + bytes[0] = 0x01; // cidv1 + bytes[1] = 0x71; // dag-cbor + bytes[2] = 0x12; // sha256 + bytes[3] = 0x20; // 32-byte digest + DaslCid::from_bytes_raw(&bytes).expect("test cid") + } + + #[test] + fn round_trips_known_null_prev() { + let state = CommitState { + signature: test_sig(0xAB), + prev: StoredPrev::Known(None), + }; + let decoded = CommitState::decode(&state.encode()).expect("decode"); + assert_eq!(decoded.signature, state.signature); + assert_eq!(decoded.prev, StoredPrev::Known(None)); + } + + #[test] + fn round_trips_known_prev_cid() { + let state = CommitState { + signature: test_sig(0xCD), + prev: StoredPrev::Known(Some(test_cid())), + }; + let decoded = CommitState::decode(&state.encode()).expect("decode"); + assert_eq!(decoded.signature, state.signature); + assert_eq!(decoded.prev, StoredPrev::Known(Some(test_cid()))); + } + + #[test] + fn decodes_legacy_bare_signature() { + // slots written before prev tracking: exactly the raw 64-byte sig + let decoded = CommitState::decode(&test_sig(0x01)).expect("decode"); + assert_eq!(decoded.signature, test_sig(0x01)); + assert_eq!(decoded.prev, StoredPrev::Legacy); + } + + #[test] + fn rejects_unknown_lengths() { + for len in [0, 1, SIG_LEN - 1, LEN_PREV_NULL + 1, LEN_PREV_CID - 1] { + assert!( + CommitState::decode(&vec![TAG_PREV_NULL; len]).is_err(), + "length {len} must not decode", + ); + } + } + + #[test] + fn rejects_tag_length_mismatch() { + let mut null_sized = vec![TAG_PREV_CID; LEN_PREV_NULL]; + assert!(CommitState::decode(&null_sized).is_err()); + null_sized[0] = 0x7F; + assert!(CommitState::decode(&null_sized).is_err()); + + let mut cid_sized = vec![0u8; LEN_PREV_CID]; + cid_sized[0] = TAG_PREV_NULL; + assert!(CommitState::decode(&cid_sized).is_err()); + } + + #[test] + fn rejects_invalid_prev_cid() { + // right tag and length, but the cid bytes don't parse as a dasl cid + let mut bytes = vec![0u8; LEN_PREV_CID]; + bytes[0] = TAG_PREV_CID; + assert!(CommitState::decode(&bytes).is_err()); } } diff --git a/hubble/src/store/mod.rs b/hubble/src/store/mod.rs index 29128f5..e662b68 100644 --- a/hubble/src/store/mod.rs +++ b/hubble/src/store/mod.rs @@ -4,7 +4,7 @@ mod id_interner; mod info_slot; pub mod record; -pub use commit_slot::CommitState; +pub use commit_slot::{CommitState, StoredPrev}; pub use id_interner::{IdInterner, IdInternerError}; pub use info_slot::InfoState; diff --git a/hubble/src/sync.rs b/hubble/src/sync.rs index 9bc79e5..e7996f5 100644 --- a/hubble/src/sync.rs +++ b/hubble/src/sync.rs @@ -1,6 +1,8 @@ use crate::Hubble; use crate::metrics::{REPO_SST_INGESTIONS, RESYNC_GENERATIONS_OVERFLOW, RESYNC_TOTAL_BLOCK_BYTES}; -use crate::store::{CommitState, IdInternerError, InfoState, counts, record, repo_sst_opts}; +use crate::store::{ + CommitState, IdInternerError, InfoState, StoredPrev, counts, record, repo_sst_opts, +}; use hubble_sync::{ AccountStatus, AppResult, Commit, ConsumerAppError, RepoContext, RepoSlots, ResyncData, StatefulSyncConsumer, StorageBatch, StorageEngine, @@ -27,9 +29,10 @@ impl StatefulSyncConsumer for Hubble { ctx: &RepoContext, batch: &mut ::Batch, ) -> AppResult { - // stash the signature so we can reconstruct a commit object for export + // stash the signature+prev so we can reconstruct a commit object later slots.commit.set(CommitState { signature: commit.object.sig.clone(), + prev: StoredPrev::Known(commit.object.prev), }); let info_slot = slots.info.get(); let generation = info_slot.map_or(0, |i| i.resync_generation); @@ -176,9 +179,10 @@ impl StatefulSyncConsumer for Hubble { resync_generation: next_gen, did_id, }); - // and of course, update the commit signature + // and of course, update the commit signature (and prev) slots.commit.set(CommitState { signature: data.commit.sig.clone(), + prev: StoredPrev::Known(data.commit.prev), }); // aaaaand get those counters