diff --git a/src/app.rs b/src/app.rs index 5ff7efc..29aab9b 100644 --- a/src/app.rs +++ b/src/app.rs @@ -6,7 +6,7 @@ use crate::{ domain::{DeviceId, EntryKind, FolderRoster, MemberRole, RosterMember, is_conflict_artifact}, iroh::NodeHost, protocol::{INVITATION_PROTOCOL_VERSION, Invite, decode_invite, encode_invite}, - storage::{AppPaths, FolderConfig, ManifestRevision, PeerInfo, StateStore}, + storage::{AppPaths, AuditVerification, FolderConfig, ManifestRevision, PeerInfo, StateStore}, }; pub type AppResult = anyhow::Result; @@ -225,6 +225,16 @@ impl AppaService { self.state_store.manifest_history(folder.id) } + pub fn audit_log(&self, folder_path: &Path) -> AppResult> { + let folder = self.require_folder(folder_path)?; + self.state_store.audit_events(folder.id) + } + + pub fn verify_audit(&self, folder_path: &Path) -> AppResult { + let folder = self.require_folder(folder_path)?; + self.state_store.verify_audit(folder.id) + } + pub fn conflicts(&self, folder_path: &Path) -> AppResult> { let folder = self.require_folder(folder_path)?; let manifest = self.state_store.load_manifest(folder.id)?; diff --git a/src/app/sync.rs b/src/app/sync.rs index b417e4c..66c73a5 100644 --- a/src/app/sync.rs +++ b/src/app/sync.rs @@ -6,7 +6,7 @@ use crate::{ RemoteManifestResult, apply_remote_manifest, recover_pending_materialization, }, }, - domain::{FolderRoster, manifest_root_hash}, + domain::{AuditEvent, AuditEventKind, FolderRoster, manifest_root_hash}, iroh::{FolderSession, NodeHost, PeerSummaryResponse}, storage::FolderConfig, }; @@ -147,10 +147,34 @@ impl AppaService { node: &NodeHost, manifest: &crate::domain::Manifest, ) -> AppResult<()> { - self.state_store.save_manifest(manifest)?; + let audit_event = self.manifest_audit_event(manifest)?; + self.state_store.save_manifest_with_audit(manifest, &audit_event)?; node.publish_manifest(manifest.clone()).await } + fn manifest_audit_event(&self, manifest: &crate::domain::Manifest) -> AppResult { + let identity = self.paths.load_identity()?; + let roster = self.load_roster(manifest.folder_id)?; + let author_device_id = identity.public().to_string(); + if !roster.can_publish(&author_device_id) { + anyhow::bail!("a non-writing member cannot sign a manifest audit event"); + } + let (sequence, parent_hash) = self + .state_store + .next_audit_sequence_and_parent(manifest.folder_id, &author_device_id)?; + let mut event = AuditEvent::create( + manifest.folder_id, + author_device_id, + sequence, + parent_hash, + AuditEventKind::ManifestCommitted, + Some(manifest_root_hash(manifest)?), + roster.hash()?, + ); + event.sign(&identity)?; + Ok(event) + } + pub(super) fn save_discovered_peers( &self, folder: &FolderConfig, diff --git a/src/cli.rs b/src/cli.rs index 88ca51e..8ea2105 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -158,6 +158,13 @@ enum Command { #[arg(value_name = "FOLDER", help = "Folder whose history to show")] folder: String, }, + #[command(about = "Show or verify the signed folder audit log")] + Audit { + #[arg(value_name = "FOLDER", help = "Folder whose audit log to inspect")] + folder: String, + #[arg(long, help = "Verify signatures and chain links")] + verify: bool, + }, #[command(about = "Restore a saved folder revision")] Restore { #[arg(value_name = "FOLDER", help = "Folder to restore")] @@ -350,6 +357,9 @@ async fn run_app_command(command: Command, offline: bool) -> anyhow::Result<()> println!("{}", client.peers(PathBuf::from(folder), json).await?) } Command::History { folder } => println!("{}", client.history(PathBuf::from(folder)).await?), + Command::Audit { folder, verify } => { + println!("{}", client.audit(PathBuf::from(folder), verify).await?) + } Command::Restore { folder, revision } => { let count = client .restore_revision(PathBuf::from(folder), revision) diff --git a/src/daemon.rs b/src/daemon.rs index d798f9b..f4de058 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -138,6 +138,10 @@ async fn handle_command( Command::History { folder_path } => { service_history(service, &folder_path).map(Response::Text) } + Command::Audit { + folder_path, + verify, + } => service_audit(service, &folder_path, verify).map(Response::Text), Command::Conflicts { folder_path } => { service_conflicts(service, &folder_path).map(Response::Text) } @@ -200,6 +204,38 @@ fn service_history(service: &AppaService, folder_path: &Path) -> anyhow::Result< .collect()) } +fn service_audit(service: &AppaService, folder_path: &Path, verify: bool) -> anyhow::Result { + if verify { + let verification = service.verify_audit(folder_path)?; + if verification.is_valid() { + return Ok(format!( + "Audit log is valid. Events: {}. Author chains: {}.\n", + verification.event_count, + verification.heads.len() + )); + } + let faults = verification + .faults + .into_iter() + .map(|fault| format!(" {}\n", fault.detail)) + .collect::(); + return Ok(format!("Audit log has integrity faults:\n{faults}")); + } + Ok(service + .audit_log(folder_path)? + .into_iter() + .map(|event| { + format!( + "{}\t{}\t{}\t{}\n", + event.sequence, + event.author_device_id, + format!("{:?}", event.kind), + event.hash().unwrap_or_else(|_| "invalid".to_owned()) + ) + }) + .collect()) +} + fn service_conflicts(service: &AppaService, folder_path: &Path) -> anyhow::Result { Ok(service .conflicts(folder_path)? diff --git a/src/domain.rs b/src/domain.rs index aaebdeb..9e6fce3 100644 --- a/src/domain.rs +++ b/src/domain.rs @@ -4,6 +4,7 @@ use serde::{Deserialize, Serialize}; use time::OffsetDateTime; use uuid::Uuid; +mod audit; mod merkle; mod reconciliation; mod roster; @@ -293,3 +294,4 @@ mod tests { ); } } +pub(crate) use audit::{AuditEvent, AuditEventKind}; diff --git a/src/domain/audit.rs b/src/domain/audit.rs new file mode 100644 index 0000000..672e34b --- /dev/null +++ b/src/domain/audit.rs @@ -0,0 +1,150 @@ +use std::collections::BTreeMap; + +use anyhow::Context; +use iroh::{EndpointId, SecretKey, Signature}; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; + +use super::{DeviceId, FolderId, canonical_json_bytes}; + +pub const AUDIT_PROTOCOL_VERSION: u16 = 1; + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum AuditEventKind { + ManifestCommitted, + RosterUpdated, + OwnerIdentityRestored, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] +pub struct AuditEvent { + pub protocol_version: u16, + pub folder_id: FolderId, + pub author_device_id: DeviceId, + pub sequence: u64, + pub parent_hash: Option, + pub kind: AuditEventKind, + pub manifest_root_hash: Option, + pub roster_hash: String, + pub occurred_at: OffsetDateTime, + pub metadata: BTreeMap, + pub signature: Option, +} + +impl AuditEvent { + pub fn create( + folder_id: FolderId, + author_device_id: DeviceId, + sequence: u64, + parent_hash: Option, + kind: AuditEventKind, + manifest_root_hash: Option, + roster_hash: String, + ) -> Self { + Self { + protocol_version: AUDIT_PROTOCOL_VERSION, + folder_id, + author_device_id, + sequence, + parent_hash, + kind, + manifest_root_hash, + roster_hash, + occurred_at: OffsetDateTime::now_utc(), + metadata: BTreeMap::new(), + signature: None, + } + } + + pub fn sign(&mut self, identity: &SecretKey) -> anyhow::Result<()> { + if identity.public().to_string() != self.author_device_id { + anyhow::bail!("audit event author does not match its signing identity"); + } + self.signature = Some(identity.sign(&self.signing_bytes()?)); + Ok(()) + } + + pub fn validate(&self) -> anyhow::Result<()> { + if self.protocol_version != AUDIT_PROTOCOL_VERSION { + anyhow::bail!("unsupported audit protocol version"); + } + if self.sequence == 0 { + anyhow::bail!("audit event sequence must be positive"); + } + let author = self.author_device_id.parse::()?; + let signature = self + .signature + .as_ref() + .ok_or_else(|| anyhow::anyhow!("audit event is unsigned"))?; + author + .verify(&self.signing_bytes()?, signature) + .context("audit event signature is invalid") + } + + pub fn hash(&self) -> anyhow::Result { + Ok(blake3::hash(&self.signing_bytes()?).to_hex().to_string()) + } + + fn signing_bytes(&self) -> anyhow::Result> { + canonical_json_bytes(&UnsignedAuditEvent::from(self)) + } +} + +#[derive(Serialize)] +struct UnsignedAuditEvent<'a> { + protocol_version: u16, + folder_id: FolderId, + author_device_id: &'a str, + sequence: u64, + parent_hash: &'a Option, + kind: &'a AuditEventKind, + manifest_root_hash: &'a Option, + roster_hash: &'a str, + occurred_at: OffsetDateTime, + metadata: &'a BTreeMap, + signature: Option, +} + +impl<'a> From<&'a AuditEvent> for UnsignedAuditEvent<'a> { + fn from(event: &'a AuditEvent) -> Self { + Self { + protocol_version: event.protocol_version, + folder_id: event.folder_id, + author_device_id: &event.author_device_id, + sequence: event.sequence, + parent_hash: &event.parent_hash, + kind: &event.kind, + manifest_root_hash: &event.manifest_root_hash, + roster_hash: &event.roster_hash, + occurred_at: event.occurred_at, + metadata: &event.metadata, + signature: None, + } + } +} + +#[cfg(test)] +mod tests { + use super::{AuditEvent, AuditEventKind}; + use uuid::Uuid; + + #[test] + fn rejects_a_changed_signed_event() -> anyhow::Result<()> { + let identity = iroh::SecretKey::generate(); + let mut event = AuditEvent::create( + Uuid::new_v4(), + identity.public().to_string(), + 1, + None, + AuditEventKind::ManifestCommitted, + Some("manifest".to_owned()), + "roster".to_owned(), + ); + event.sign(&identity)?; + event.sequence = 2; + + assert!(event.validate().is_err()); + Ok(()) + } +} diff --git a/src/domain/roster.rs b/src/domain/roster.rs index b814480..c4eb2b9 100644 --- a/src/domain/roster.rs +++ b/src/domain/roster.rs @@ -83,6 +83,10 @@ impl FolderRoster { .context("folder roster signature is invalid") } + pub fn hash(&self) -> anyhow::Result { + Ok(blake3::hash(&self.signing_bytes()?).to_hex().to_string()) + } + pub fn contains_member(&self, device_id: &str) -> bool { self.members.contains_key(device_id) } diff --git a/src/ipc.rs b/src/ipc.rs index 013c5e2..b37f4d5 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -62,6 +62,10 @@ pub(crate) enum Command { History { folder_path: PathBuf, }, + Audit { + folder_path: PathBuf, + verify: bool, + }, Conflicts { folder_path: PathBuf, }, @@ -183,6 +187,10 @@ impl DaemonClient { self.request_text(Command::History { folder_path }).await } + pub async fn audit(&self, folder_path: PathBuf, verify: bool) -> anyhow::Result { + self.request_text(Command::Audit { folder_path, verify }).await + } + pub async fn conflicts(&self, folder_path: PathBuf) -> anyhow::Result { self.request_text(Command::Conflicts { folder_path }).await } diff --git a/src/storage.rs b/src/storage.rs index 9f4ea37..3f3aac4 100644 --- a/src/storage.rs +++ b/src/storage.rs @@ -17,6 +17,7 @@ use crate::{ }; mod fingerprints; +mod audit; mod manifests; mod materializations; mod paths; @@ -35,12 +36,16 @@ const FOLDER_SCOPED_TABLES: &[&str] = &[ "sync_state", "file_fingerprints", "pending_materializations", + "audit_events", + "audit_heads", + "audit_integrity_faults", ]; pub use fingerprints::FileFingerprint; pub use manifests::ManifestRevision; pub use paths::AppPaths; pub use peers::PeerInfo; +pub use audit::AuditVerification; const DEFAULT_HISTORY_REVISIONS: usize = 100; diff --git a/src/storage/audit.rs b/src/storage/audit.rs new file mode 100644 index 0000000..9f87826 --- /dev/null +++ b/src/storage/audit.rs @@ -0,0 +1,216 @@ +use rusqlite::{OptionalExtension, params}; + +use crate::{ + domain::{AuditEvent, DeviceId, FolderId}, + storage::StateStore, +}; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AuditHead { + pub author_device_id: DeviceId, + pub sequence: u64, + pub event_hash: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AuditIntegrityFault { + pub author_device_id: DeviceId, + pub detail: String, +} + +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct AuditVerification { + pub event_count: usize, + pub heads: Vec, + pub faults: Vec, +} + +impl AuditVerification { + pub fn is_valid(&self) -> bool { + self.faults.is_empty() + } +} + +impl StateStore { + pub fn next_audit_sequence_and_parent( + &self, + folder_id: FolderId, + author_device_id: &str, + ) -> anyhow::Result<(u64, Option)> { + let head = self.audit_head(folder_id, author_device_id)?; + Ok(match head { + Some(head) => (head.sequence + 1, Some(head.event_hash)), + None => (1, None), + }) + } + + #[cfg(test)] + pub fn append_audit_event(&self, event: &AuditEvent) -> anyhow::Result<()> { + event.validate()?; + let transaction = self.connection.unchecked_transaction()?; + append_audit_event(&transaction, event)?; + transaction.commit()?; + Ok(()) + } + + pub fn audit_events(&self, folder_id: FolderId) -> anyhow::Result> { + let mut statement = self.connection.prepare( + "SELECT event FROM audit_events WHERE folder_id = ?1 ORDER BY author_device_id, sequence", + )?; + statement + .query_map(params![folder_id.to_string()], |row| row.get::<_, String>(0))? + .map(|event| serde_json::from_str(&event?).map_err(Into::into)) + .collect() + } + + pub fn audit_heads(&self, folder_id: FolderId) -> anyhow::Result> { + let mut statement = self.connection.prepare( + "SELECT author_device_id, sequence, event_hash FROM audit_heads WHERE folder_id = ?1 ORDER BY author_device_id", + )?; + statement + .query_map(params![folder_id.to_string()], |row| { + Ok(AuditHead { + author_device_id: row.get(0)?, + sequence: read_sequence(row.get::<_, i64>(1)?)?, + event_hash: row.get(2)?, + }) + })? + .collect::, _>>() + .map_err(Into::into) + } + + pub fn verify_audit(&self, folder_id: FolderId) -> anyhow::Result { + let events = self.audit_events(folder_id)?; + let mut faults = Vec::new(); + let mut expected_by_author = std::collections::BTreeMap::new(); + for event in &events { + if let Err(error) = validate_event_link(event, &mut expected_by_author) { + faults.push(AuditIntegrityFault { + author_device_id: event.author_device_id.clone(), + detail: error.to_string(), + }); + } + } + faults.extend(self.audit_integrity_faults(folder_id)?); + Ok(AuditVerification { + event_count: events.len(), + heads: self.audit_heads(folder_id)?, + faults, + }) + } + + fn audit_head(&self, folder_id: FolderId, author_device_id: &str) -> anyhow::Result> { + self.connection + .query_row( + "SELECT sequence, event_hash FROM audit_heads WHERE folder_id = ?1 AND author_device_id = ?2", + params![folder_id.to_string(), author_device_id], + |row| Ok(AuditHead { + author_device_id: author_device_id.to_owned(), + sequence: read_sequence(row.get::<_, i64>(0)?)?, + event_hash: row.get(1)?, + }), + ) + .optional() + .map_err(Into::into) + } + + fn audit_integrity_faults(&self, folder_id: FolderId) -> anyhow::Result> { + let mut statement = self.connection.prepare( + "SELECT author_device_id, detail FROM audit_integrity_faults WHERE folder_id = ?1 ORDER BY detected_at", + )?; + statement + .query_map(params![folder_id.to_string()], |row| Ok(AuditIntegrityFault { + author_device_id: row.get(0)?, + detail: row.get(1)?, + }))? + .collect::, _>>() + .map_err(Into::into) + } +} + +pub(super) fn append_audit_event( + transaction: &rusqlite::Transaction<'_>, + event: &AuditEvent, +) -> anyhow::Result<()> { + event.validate()?; + let current_head = transaction + .query_row( + "SELECT sequence, event_hash FROM audit_heads WHERE folder_id = ?1 AND author_device_id = ?2", + params![event.folder_id.to_string(), event.author_device_id], + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)), + ) + .optional()?; + validate_new_event(event, current_head)?; + let event_hash = event.hash()?; + transaction.execute( + "INSERT INTO audit_events (folder_id, author_device_id, sequence, event_hash, parent_hash, event) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![event.folder_id.to_string(), event.author_device_id, i64::try_from(event.sequence)?, event_hash, event.parent_hash, serde_json::to_string(event)?], + )?; + transaction.execute( + "INSERT INTO audit_heads (folder_id, author_device_id, sequence, event_hash) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(folder_id, author_device_id) DO UPDATE SET sequence = excluded.sequence, event_hash = excluded.event_hash", + params![event.folder_id.to_string(), event.author_device_id, i64::try_from(event.sequence)?, event.hash()?], + )?; + Ok(()) +} + +fn validate_new_event(event: &AuditEvent, current_head: Option<(i64, String)>) -> anyhow::Result<()> { + match current_head { + None if event.sequence == 1 && event.parent_hash.is_none() => Ok(()), + Some((sequence, hash)) + if event.sequence == u64::try_from(sequence)? + 1 + && event.parent_hash.as_deref() == Some(&hash) => Ok(()), + None => anyhow::bail!("first audit event must have sequence 1 and no parent hash"), + Some(_) => anyhow::bail!("audit event does not extend the current author chain"), + } +} + +fn read_sequence(sequence: i64) -> rusqlite::Result { + u64::try_from(sequence).map_err(|error| rusqlite::Error::FromSqlConversionFailure( + 0, + rusqlite::types::Type::Integer, + Box::new(error), + )) +} + +fn validate_event_link( + event: &AuditEvent, + expected_by_author: &mut std::collections::BTreeMap, +) -> anyhow::Result<()> { + event.validate()?; + let current_head = expected_by_author.get(&event.author_device_id).cloned(); + let current_head = match current_head { + Some((sequence, hash)) => Some((i64::try_from(sequence)?, hash)), + None => None, + }; + validate_new_event(event, current_head)?; + expected_by_author.insert(event.author_device_id.clone(), (event.sequence, event.hash()?)); + Ok(()) +} + +#[cfg(test)] +mod tests { + use tempfile::TempDir; + use uuid::Uuid; + + use crate::{domain::{AuditEvent, AuditEventKind}, storage::{AppPaths, StateStore}}; + + #[test] + fn verifies_a_signed_author_chain() -> anyhow::Result<()> { + let directory = TempDir::new()?; + let store = StateStore::open(&AppPaths::from_data_directory(directory.path().join("state"))?)?; + let identity = iroh::SecretKey::generate(); + let folder_id = Uuid::new_v4(); + let mut first = AuditEvent::create(folder_id, identity.public().to_string(), 1, None, AuditEventKind::ManifestCommitted, Some("one".to_owned()), "roster".to_owned()); + first.sign(&identity)?; + store.append_audit_event(&first)?; + let (sequence, parent) = store.next_audit_sequence_and_parent(folder_id, &identity.public().to_string())?; + let mut second = AuditEvent::create(folder_id, identity.public().to_string(), sequence, parent, AuditEventKind::ManifestCommitted, Some("two".to_owned()), "roster".to_owned()); + second.sign(&identity)?; + store.append_audit_event(&second)?; + + let verification = store.verify_audit(folder_id)?; + assert!(verification.is_valid()); + assert_eq!(verification.event_count, 2); + Ok(()) + } +} diff --git a/src/storage/manifests.rs b/src/storage/manifests.rs index a8a5267..441130a 100644 --- a/src/storage/manifests.rs +++ b/src/storage/manifests.rs @@ -6,8 +6,9 @@ use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use crate::{ domain::{Entry, FolderId, Manifest}, - storage::{StateStore, prune_manifest_history}, + storage::{StateStore, audit::append_audit_event, prune_manifest_history}, }; +use crate::domain::AuditEvent; const HISTORY_CHECKPOINT_INTERVAL: u64 = 20; const COMPRESSION_LEVEL: i32 = 3; @@ -43,6 +44,22 @@ impl StateStore { } pub fn save_manifest(&self, manifest: &Manifest) -> anyhow::Result<()> { + self.save_manifest_with_optional_audit(manifest, None) + } + + pub fn save_manifest_with_audit( + &self, + manifest: &Manifest, + audit_event: &AuditEvent, + ) -> anyhow::Result<()> { + self.save_manifest_with_optional_audit(manifest, Some(audit_event)) + } + + fn save_manifest_with_optional_audit( + &self, + manifest: &Manifest, + audit_event: Option<&AuditEvent>, + ) -> anyhow::Result<()> { let serialized = serde_json::to_string(manifest)?; let existing = self.load_existing_manifest(manifest.folder_id)?; if existing.as_deref() == Some(&serialized) { @@ -55,6 +72,9 @@ impl StateStore { transaction.execute("INSERT INTO manifest_history_compact (folder_id, saved_at, kind, payload) VALUES (?1, ?2, ?3, ?4)", params![manifest.folder_id.to_string(), OffsetDateTime::now_utc().format(&Rfc3339)?, payload.kind, payload.bytes])?; checkpoint_retention_boundary(&transaction, manifest.folder_id, self.history_limit)?; prune_manifest_history(&transaction, manifest.folder_id, self.history_limit)?; + if let Some(audit_event) = audit_event { + append_audit_event(&transaction, audit_event)?; + } transaction.commit()?; Ok(()) } diff --git a/src/storage/schema.rs b/src/storage/schema.rs index eeec171..97e360b 100644 --- a/src/storage/schema.rs +++ b/src/storage/schema.rs @@ -1,6 +1,6 @@ use rusqlite::Connection; -const CURRENT_SCHEMA_VERSION: u32 = 1; +const CURRENT_SCHEMA_VERSION: u32 = 2; struct Migration { version: u32, @@ -70,6 +70,32 @@ const MIGRATIONS: &[Migration] = &[Migration { resulting_manifest TEXT NOT NULL );", ], +}, Migration { + version: CURRENT_SCHEMA_VERSION, + statements: &[ + "CREATE TABLE IF NOT EXISTS audit_events ( + folder_id TEXT NOT NULL, + author_device_id TEXT NOT NULL, + sequence INTEGER NOT NULL, + event_hash TEXT NOT NULL UNIQUE, + parent_hash TEXT, + event TEXT NOT NULL, + PRIMARY KEY (folder_id, author_device_id, sequence) + );", + "CREATE TABLE IF NOT EXISTS audit_heads ( + folder_id TEXT NOT NULL, + author_device_id TEXT NOT NULL, + sequence INTEGER NOT NULL, + event_hash TEXT NOT NULL, + PRIMARY KEY (folder_id, author_device_id) + );", + "CREATE TABLE IF NOT EXISTS audit_integrity_faults ( + folder_id TEXT NOT NULL, + author_device_id TEXT NOT NULL, + detected_at TEXT NOT NULL, + detail TEXT NOT NULL + );", + ], }]; pub(super) fn initialize(connection: &mut Connection) -> anyhow::Result<()> {