From a87b16dd852a3a6e7be7ad0f080b3ad97c75fdbc Mon Sep 17 00:00:00 2001 From: Sergio Date: Tue, 9 Jun 2026 14:43:07 +0000 Subject: [PATCH] =?UTF-8?q?Fase=203=20=E2=80=94=20diario=20de=20mutaciones?= =?UTF-8?q?=20(hammer-journal=20+=20fanotify=20+=20hammer=20journal)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nuevo crate hammer-journal y watcher fanotify en hammerd. Cierra el contrato del SDD 05 §3 ("toda mutación trazable/opaca registrada y consultable") y liquida la deuda Fase 2 sobre el hook al diario en commit. hammer-journal: - MutationEvent (op, path, by, content_hash?, note?), MutationOp Create/Replace/Edit/Delete, Source enum tagged HammerHydrate / HammerCommit / External (distingue trazable de opaco para `export`). - Journal: record append-only + sync_data, read_all/tail/follow (offset), resilencia a líneas malformadas (kernel-panic mid-write se ignora). - RFC 3339 UTC sin chrono (Hinnant civil_from_days, válido para todo i64). - 7 unit tests cubren append/read/tail/follow/malformed/serde/timestamp. Watcher (hammerd): - nix::sys::fanotify con FAN_CLASS_NOTIF + FAN_CLOSE_WRITE + FAN_EVENT_ON_CHILD sobre /bin /sbin /usr/bin /usr/sbin /lib /usr/lib /etc. - Resuelve path por readlink(/proc/self/fd/) — sin FAN_REPORT_DFID_NAME para Fase 3 v1; el refinamiento Create/Edit con metadata viene después. - Filtro: hammer_overlay::status(state_root) → ignora paths bajo un overlay activo (el ruido del experimento no entra hasta el commit). - Sin CAP_SYS_ADMIN, fanotify_init falla; hammerd lo reporta y sigue durmiendo (preparado para que Fase 5 monte el bus encima). - 3 unit tests sobre defaults/filtros/uid lookup. CLI: - hammer journal [--dir] [--tail N] [--follow] [--format pretty|json]. - hammer commit registra cada archivo promocionado/removido vía commit_with_journal; --no-journal opt-out. Hook overlay::commit: - commit_with_journal(id, root, journal): para cada copied/removed emite un MutationEvent con Source::HammerCommit { overlay }. Test directo del hook sin overlayfs (lógica pura) más el e2e existente que ya cubre los mounts. 57 tests workspace, 0 fallos. Co-Authored-By: Claude Opus 4.7 --- Cargo.lock | 41 ++++ Cargo.toml | 3 + crates/hammer-cli/Cargo.toml | 2 + crates/hammer-cli/src/main.rs | 94 +++++++- crates/hammer-journal/Cargo.toml | 19 ++ crates/hammer-journal/src/lib.rs | 370 +++++++++++++++++++++++++++++++ crates/hammer-overlay/Cargo.toml | 2 + crates/hammer-overlay/src/lib.rs | 105 ++++++++- crates/hammerd/Cargo.toml | 8 + crates/hammerd/src/main.rs | 46 +++- crates/hammerd/src/watcher.rs | 229 +++++++++++++++++++ docs/10-roadmap.md | 18 +- 12 files changed, 911 insertions(+), 26 deletions(-) create mode 100644 crates/hammer-journal/Cargo.toml create mode 100644 crates/hammer-journal/src/lib.rs create mode 100644 crates/hammerd/src/watcher.rs diff --git a/Cargo.lock b/Cargo.lock index a0eec2a3..cba81b11 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -124,6 +124,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + [[package]] name = "clap" version = "4.6.1" @@ -293,7 +299,9 @@ dependencies = [ "clap", "hammer-build", "hammer-core", + "hammer-journal", "hammer-overlay", + "serde_json", "tracing", "tracing-subscriber", ] @@ -312,12 +320,27 @@ dependencies = [ "toml", ] +[[package]] +name = "hammer-journal" +version = "0.0.1" +dependencies = [ + "anyhow", + "hammer-core", + "serde", + "serde_json", + "tempfile", + "thiserror", + "tracing", +] + [[package]] name = "hammer-overlay" version = "0.0.1" dependencies = [ "anyhow", "hammer-core", + "hammer-journal", + "libc", "serde", "serde_json", "tempfile", @@ -333,6 +356,12 @@ dependencies = [ "anyhow", "clap", "hammer-core", + "hammer-journal", + "hammer-overlay", + "libc", + "nix", + "tempfile", + "thiserror", "tracing", "tracing-subscriber", ] @@ -439,6 +468,18 @@ version = "2.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b947ae49db0d222b1dbc6b113ce7248a3fc3a6ca21b696717bfc000ba4484d8" +[[package]] +name = "nix" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" diff --git a/Cargo.toml b/Cargo.toml index 87619dbb..dc1a8497 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "crates/hammer-core", "crates/hammer-build", "crates/hammer-overlay", + "crates/hammer-journal", "crates/hammer-cli", "crates/hammerd", ] @@ -19,6 +20,7 @@ repository = "https://gitea.gioser.net/sergio/hammer" hammer-core = { path = "crates/hammer-core" } hammer-build = { path = "crates/hammer-build" } hammer-overlay = { path = "crates/hammer-overlay" } +hammer-journal = { path = "crates/hammer-journal" } # shared third-party deps, pinned at the workspace level anyhow = "1" @@ -30,6 +32,7 @@ toml = "0.8" blake3 = "1" sha2 = "0.10" hex = "0.4" +nix = { version = "0.30", default-features = false, features = ["fanotify", "fs", "user"] } tempfile = "3" clap = { version = "4", features = ["derive"] } tracing = "0.1" diff --git a/crates/hammer-cli/Cargo.toml b/crates/hammer-cli/Cargo.toml index 5f88e163..32c1dc57 100644 --- a/crates/hammer-cli/Cargo.toml +++ b/crates/hammer-cli/Cargo.toml @@ -15,6 +15,8 @@ path = "src/main.rs" hammer-core.workspace = true hammer-build.workspace = true hammer-overlay.workspace = true +hammer-journal.workspace = true +serde_json.workspace = true anyhow.workspace = true clap.workspace = true tracing.workspace = true diff --git a/crates/hammer-cli/src/main.rs b/crates/hammer-cli/src/main.rs index 0b6cc918..e09c7257 100644 --- a/crates/hammer-cli/src/main.rs +++ b/crates/hammer-cli/src/main.rs @@ -52,11 +52,18 @@ enum Cmd { #[arg(long)] state_root: Option, }, - /// [Fase 2] Fusiona un overlay al FHS real y desmonta. Sin diario aún (Fase 3). + /// [Fase 2+3] Fusiona un overlay al FHS real y desmonta. Por defecto registra cada + /// archivo promocionado en el diario; usa `--no-journal` para saltarse el registro. Commit { id: String, #[arg(long)] state_root: Option, + /// Directorio del diario donde anotar la promoción. + #[arg(long, default_value = "/var/lib/hammer/journal")] + journal: PathBuf, + /// Saltarse el registro en el diario (útil para experimentos efímeros). + #[arg(long)] + no_journal: bool, }, /// [Fase 2] Desmonta un overlay y borra su upper. El sistema vuelve al estado base. Discard { @@ -70,7 +77,20 @@ enum Cmd { state_root: Option, }, /// [Fase 3] Ver/seguir el diario de mutaciones. - Journal, + Journal { + /// Directorio del diario (mismo que `hammerd --journal`). + #[arg(long, default_value = "/var/lib/hammer/journal")] + dir: PathBuf, + /// Muestra los últimos N eventos. Default: 20. + #[arg(long, short, default_value = "20")] + tail: usize, + /// Sigue el archivo: imprime nuevos eventos según aparecen. + #[arg(long, short)] + follow: bool, + /// Formato. `pretty` (humano, default) o `json` (raw passthrough). + #[arg(long, default_value = "pretty")] + format: String, + }, /// [Fase 4] Aplica un manifiesto .swm (reproduce y lo deja en un overlay). Apply { file: String }, /// [Fase 4] Exporta el delta del sistema como manifiesto .swm. @@ -82,6 +102,39 @@ enum Cmd { Ctl { line: String }, } +fn print_event(ev: &hammer_journal::MutationEvent, format: &str) { + match format { + "json" => { + // Re-serializa. No imprimimos la línea original (no la tenemos a mano), pero el + // formato es estable porque MutationEvent es la fuente de verdad del schema. + match serde_json::to_string(ev) { + Ok(s) => println!("{s}"), + Err(e) => eprintln!("journal: error serializando: {e}"), + } + } + _ => { + let actor = match &ev.by.source { + hammer_journal::Source::HammerHydrate { artifact } => format!("hydrate {artifact}"), + hammer_journal::Source::HammerCommit { overlay, artifact } => match artifact { + Some(a) => format!("commit {overlay} ({a})"), + None => format!("commit {overlay}"), + }, + hammer_journal::Source::External => { + let pid = ev.by.pid.map(|p| format!("pid={p}")).unwrap_or_default(); + let uid = ev.by.uid.map(|u| format!("uid={u}")).unwrap_or_default(); + format!("external {pid} {uid}").trim().to_string() + } + }; + println!( + "{ts} {op:<7} {path} ← {actor}", + ts = ev.ts, + op = ev.op.as_str(), + path = ev.path.display(), + ); + } + } +} + fn main() -> anyhow::Result<()> { tracing_subscriber::fmt() .with_writer(std::io::stderr) @@ -134,14 +187,21 @@ fn main() -> anyhow::Result<()> { let id = hammer_overlay::try_overlay(&targets, &root)?; println!("{id}"); } - Cmd::Commit { id, state_root } => { + Cmd::Commit { id, state_root, journal, no_journal } => { let root = state_root .unwrap_or_else(|| PathBuf::from(hammer_overlay::DEFAULT_STATE_ROOT)); - let report = hammer_overlay::commit(&hammer_overlay::OverlayId(id), &root)?; + let id = hammer_overlay::OverlayId(id); + let report = if no_journal { + hammer_overlay::commit(&id, &root)? + } else { + let j = hammer_journal::Journal::open(&journal)?; + hammer_overlay::commit_with_journal(&id, &root, &j)? + }; println!( - "commit OK — {} archivo(s) promocionados, {} eliminado(s)", + "commit OK — {} archivo(s) promocionados, {} eliminado(s){}", report.copied.len(), - report.removed.len() + report.removed.len(), + if no_journal { "" } else { " (anotado en el diario)" } ); } Cmd::Discard { id, state_root } => { @@ -170,8 +230,26 @@ fn main() -> anyhow::Result<()> { } } } - Cmd::Journal => { - println!("[fase 3 pendiente] diario — ver docs/05-journal.md"); + Cmd::Journal { dir, tail, follow, format } => { + let j = hammer_journal::Journal::open(&dir)?; + let initial = j.tail(tail)?; + for ev in &initial { + print_event(ev, &format); + } + if follow { + use std::io::Write; + let _ = std::io::stdout().flush(); + let mut cursor = j.end_offset()?; + loop { + std::thread::sleep(std::time::Duration::from_millis(500)); + let (news, new_cursor) = j.read_from_offset(cursor)?; + cursor = new_cursor; + for ev in &news { + print_event(ev, &format); + } + let _ = std::io::stdout().flush(); + } + } } Cmd::Apply { file } => { println!("[fase 4 pendiente] apply {file} — ver docs/06-swm-format.md"); diff --git a/crates/hammer-journal/Cargo.toml b/crates/hammer-journal/Cargo.toml new file mode 100644 index 00000000..c3e36cb6 --- /dev/null +++ b/crates/hammer-journal/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "hammer-journal" +version.workspace = true +edition.workspace = true +license.workspace = true +authors.workspace = true +repository.workspace = true +description = "El diario de mutaciones de hammer: append-only JSON-líneas sobre eventos del FHS." + +[dependencies] +hammer-core.workspace = true +anyhow.workspace = true +thiserror.workspace = true +tracing.workspace = true +serde.workspace = true +serde_json.workspace = true + +[dev-dependencies] +tempfile.workspace = true diff --git a/crates/hammer-journal/src/lib.rs b/crates/hammer-journal/src/lib.rs new file mode 100644 index 00000000..6b38b659 --- /dev/null +++ b/crates/hammer-journal/src/lib.rs @@ -0,0 +1,370 @@ +//! El diario de mutaciones. Ver `docs/05-journal.md`. +//! +//! Formato: archivo único de **JSON-líneas append-only** (`mutations.jsonl`). Cada línea es +//! un `MutationEvent` autónomo; `cat` / `grep` / `jq` los procesan sin parseo extra. +//! +//! Esta crate es **lógica pura**: nada de fanotify ni de mounts. El watcher real vive en +//! `hammerd` y empuja eventos aquí vía `Journal::record`. El hook desde +//! `hammer_overlay::commit` también empuja por aquí. +//! +//! Diseño rationale (SDD 05): +//! - El diario no bloquea acciones; sólo observa. Pero también acepta eventos "trazables" +//! (de un `hydrate` o `commit`) que llegan con `actor.source = HammerHydrate { artifact }`. +//! - Las "mutaciones opacas" (pisada manual sin rastro de artefacto) llegan con +//! `actor.source = External { pid, uid }`. Eso permite que `hammer export` distinga +//! replays reproducibles de cambios que sólo se pueden compartir como warnings. + +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write}; +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; + +/// Operación sobre el FHS. Ver SDD 05 §2 — la lista evolucionará con los eventos que el +/// watcher fanotify aprenda a distinguir. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum MutationOp { + /// Archivo nuevo donde antes no había nada. + Create, + /// Pisado: el archivo existía y fue sustituido (mismo path, distinto contenido o inode). + Replace, + /// Edición en sitio: el archivo cambió sin que cambiara el path (típico de configs en /etc). + Edit, + /// Borrado. + Delete, +} + +impl MutationOp { + pub fn as_str(&self) -> &'static str { + match self { + MutationOp::Create => "create", + MutationOp::Replace => "replace", + MutationOp::Edit => "edit", + MutationOp::Delete => "delete", + } + } +} + +/// Quién originó la mutación. Distinguir trazable vs opaca es lo que más tarde permite +/// que `hammer export` produzca un `.swm` reproducible (artefactos) más una lista de +/// warnings (pisadas manuales). +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "kebab-case")] +pub enum Source { + /// Vino de un `hammer hydrate ` — la mutación es replay-able a partir del hash. + HammerHydrate { artifact: String }, + /// Vino de un `hammer commit ` — origen indirecto: el `actor` fue el overlay, + /// y `artifact` es opcional (puede no haber, si la promoción fue de un edit manual). + HammerCommit { + overlay: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + artifact: Option, + }, + /// Pisada manual fuera del flujo de hammer. La grabamos sin juzgar — el SDD insiste en + /// que es información, no error. + External, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Actor { + pub source: Source, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pid: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub uid: Option, +} + +/// Un evento atómico — una línea del diario. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MutationEvent { + /// RFC 3339 UTC. El watcher fija el timestamp en el momento de la observación. + pub ts: String, + pub op: MutationOp, + pub path: PathBuf, + pub by: Actor, + /// Opcional: hash blake3 del contenido del archivo tras la mutación. Permite a `export` + /// detectar mutaciones idempotentes (mismo contenido tras N replaces → un sólo evento). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content_hash: Option, + /// Nota libre (humana). Útil para que la IA explique su intención. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub note: Option, +} + +/// Handle del diario: una ruta a `mutations.jsonl`. Las operaciones son thread-safe en el +/// sentido de que cada `record` abre + append + cierra; suficiente para el throughput +/// esperado (cambios humanos, no log-de-tracing). +#[derive(Debug, Clone)] +pub struct Journal { + pub dir: PathBuf, +} + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("io: {0}")] + Io(#[from] std::io::Error), + #[error("json: {0}")] + Json(#[from] serde_json::Error), + #[error("journal: {0}")] + Other(String), +} + +pub type Result = std::result::Result; + +impl Journal { + /// Abre (creando el directorio si hace falta) un diario bajo `dir`. El archivo + /// `mutations.jsonl` se crea en el primer `record`. + pub fn open(dir: impl Into) -> Result { + let dir = dir.into(); + std::fs::create_dir_all(&dir)?; + Ok(Self { dir }) + } + + pub fn file_path(&self) -> PathBuf { + self.dir.join("mutations.jsonl") + } + + /// Append-only. Una línea JSON + `\n`. Maneja la creación del archivo si no existe. + pub fn record(&self, ev: &MutationEvent) -> Result<()> { + let line = serde_json::to_string(ev)?; + let mut f = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(self.file_path())?; + f.write_all(line.as_bytes())?; + f.write_all(b"\n")?; + f.sync_data()?; + Ok(()) + } + + /// Lee todos los eventos en orden de aparición. Líneas vacías o malformadas se loguean + /// y se saltan — el diario es append-only pero queremos resilencia a truncados de + /// emergencia (kernel-panic en medio de una escritura). + pub fn read_all(&self) -> Result> { + let path = self.file_path(); + if !path.is_file() { + return Ok(Vec::new()); + } + let f = std::fs::File::open(&path)?; + let rdr = BufReader::new(f); + let mut out = Vec::new(); + for (i, line) in rdr.lines().enumerate() { + let line = line?; + if line.trim().is_empty() { + continue; + } + match serde_json::from_str::(&line) { + Ok(ev) => out.push(ev), + Err(e) => { + tracing::warn!(line = i + 1, error = %e, "journal: línea malformada, saltando"); + } + } + } + Ok(out) + } + + /// Lee los últimos `n` eventos sin cargar todo el archivo. Implementación naïve por + /// ahora (delega a `read_all` y trunca); cuando el archivo crezca añadiremos un índice + /// por bloques. + pub fn tail(&self, n: usize) -> Result> { + let mut all = self.read_all()?; + if all.len() > n { + all.drain(..(all.len() - n)); + } + Ok(all) + } + + /// Devuelve el offset en bytes del fin de archivo actual. Útil para implementar un + /// "follow" en CLI: tras un read inicial, hacemos polling desde este offset. + pub fn end_offset(&self) -> Result { + let path = self.file_path(); + if !path.is_file() { + return Ok(0); + } + let m = std::fs::metadata(&path)?; + Ok(m.len()) + } + + /// Lee eventos desde un offset en bytes hasta el final actual. Devuelve los eventos y el + /// nuevo offset (para usar en la siguiente iteración del follow). + pub fn read_from_offset(&self, offset: u64) -> Result<(Vec, u64)> { + let path = self.file_path(); + if !path.is_file() { + return Ok((Vec::new(), 0)); + } + let mut f = std::fs::File::open(&path)?; + f.seek(SeekFrom::Start(offset))?; + let mut buf = String::new(); + f.read_to_string(&mut buf)?; + let new_offset = offset + buf.len() as u64; + let mut out = Vec::new(); + for line in buf.lines() { + if line.trim().is_empty() { + continue; + } + match serde_json::from_str::(line) { + Ok(ev) => out.push(ev), + Err(e) => tracing::warn!(error = %e, "journal: línea malformada en tail"), + } + } + Ok((out, new_offset)) + } +} + +/// Construye un timestamp RFC 3339 UTC del instante actual usando std-only (sin chrono). +pub fn now_rfc3339() -> String { + use std::time::SystemTime; + let d = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .unwrap_or_default(); + let secs = d.as_secs() as i64; + let nanos = d.subsec_nanos(); + let (year, month, day, hour, minute, second) = unix_to_ymdhms(secs); + format!( + "{year:04}-{month:02}-{day:02}T{hour:02}:{minute:02}:{second:02}.{:09}Z", + nanos + ) +} + +/// Conversión epoch → (year, month, day, hour, minute, second) UTC. Algoritmo de Howard +/// Hinnant (`civil_from_days`), válido para todo rango representable por i64. +fn unix_to_ymdhms(secs: i64) -> (i32, u32, u32, u32, u32, u32) { + let day_secs = 86_400i64; + let mut days = secs.div_euclid(day_secs); + let time_of_day = secs.rem_euclid(day_secs) as u32; + let hour = time_of_day / 3600; + let minute = (time_of_day / 60) % 60; + let second = time_of_day % 60; + + days += 719_468; // shift al epoch del algoritmo (0000-03-01) + let era = if days >= 0 { days } else { days - 146_096 } / 146_097; + let doe = (days - era * 146_097) as u64; // [0, 146_096] + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; // [0, 399] + let y = yoe as i64 + era * 400; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); // [0, 365] + let mp = (5 * doy + 2) / 153; // [0, 11] + let day = (doy - (153 * mp + 2) / 5 + 1) as u32; // [1, 31] + let month = if mp < 10 { mp + 3 } else { mp - 9 } as u32; // [1, 12] + let year = (y + if month <= 2 { 1 } else { 0 }) as i32; + + (year, month, day, hour, minute, second) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ev(op: MutationOp, path: &str) -> MutationEvent { + MutationEvent { + ts: now_rfc3339(), + op, + path: PathBuf::from(path), + by: Actor { + source: Source::External, + pid: Some(1234), + uid: Some(1000), + }, + content_hash: None, + note: None, + } + } + + #[test] + fn record_and_read_roundtrip() { + let d = tempfile::tempdir().unwrap(); + let j = Journal::open(d.path()).unwrap(); + j.record(&ev(MutationOp::Replace, "/bin/grep")).unwrap(); + j.record(&ev(MutationOp::Edit, "/etc/hosts")).unwrap(); + let read = j.read_all().unwrap(); + assert_eq!(read.len(), 2); + assert_eq!(read[0].op, MutationOp::Replace); + assert_eq!(read[0].path, PathBuf::from("/bin/grep")); + assert_eq!(read[1].op, MutationOp::Edit); + } + + #[test] + fn empty_journal_reads_empty() { + let d = tempfile::tempdir().unwrap(); + let j = Journal::open(d.path()).unwrap(); + assert!(j.read_all().unwrap().is_empty()); + assert_eq!(j.end_offset().unwrap(), 0); + } + + #[test] + fn tail_returns_last_n() { + let d = tempfile::tempdir().unwrap(); + let j = Journal::open(d.path()).unwrap(); + for i in 0..5 { + j.record(&ev(MutationOp::Create, &format!("/etc/{i}"))).unwrap(); + } + let last = j.tail(2).unwrap(); + assert_eq!(last.len(), 2); + assert_eq!(last[0].path, PathBuf::from("/etc/3")); + assert_eq!(last[1].path, PathBuf::from("/etc/4")); + } + + #[test] + fn follow_via_offset_yields_new_events_only() { + let d = tempfile::tempdir().unwrap(); + let j = Journal::open(d.path()).unwrap(); + j.record(&ev(MutationOp::Create, "/etc/a")).unwrap(); + let cursor = j.end_offset().unwrap(); + j.record(&ev(MutationOp::Create, "/etc/b")).unwrap(); + j.record(&ev(MutationOp::Create, "/etc/c")).unwrap(); + let (news, new_cursor) = j.read_from_offset(cursor).unwrap(); + assert_eq!(news.len(), 2); + assert_eq!(news[0].path, PathBuf::from("/etc/b")); + assert_eq!(news[1].path, PathBuf::from("/etc/c")); + assert_eq!(new_cursor, j.end_offset().unwrap()); + // Una segunda llamada desde el nuevo cursor devuelve vacío. + let (more, _) = j.read_from_offset(new_cursor).unwrap(); + assert!(more.is_empty()); + } + + #[test] + fn malformed_line_is_skipped_not_fatal() { + let d = tempfile::tempdir().unwrap(); + let j = Journal::open(d.path()).unwrap(); + j.record(&ev(MutationOp::Create, "/etc/a")).unwrap(); + // Plantamos basura en medio (kernel panic mid-write). + let mut f = std::fs::OpenOptions::new().append(true).open(j.file_path()).unwrap(); + f.write_all(b"{esto no es JSON valido\n").unwrap(); + f.write_all(b"\n").unwrap(); // línea vacía + drop(f); + j.record(&ev(MutationOp::Create, "/etc/b")).unwrap(); + let read = j.read_all().unwrap(); + assert_eq!(read.len(), 2, "líneas válidas se preservan"); + } + + #[test] + fn source_serializes_with_tag() { + let e = ev(MutationOp::Replace, "/bin/grep"); + let j = serde_json::to_string(&e).unwrap(); + assert!(j.contains(r#""kind":"external""#), "got {j}"); + + let mut e2 = e.clone(); + e2.by.source = Source::HammerHydrate { + artifact: "b3:abc".into(), + }; + let j2 = serde_json::to_string(&e2).unwrap(); + assert!(j2.contains(r#""kind":"hammer-hydrate""#), "got {j2}"); + assert!(j2.contains(r#""artifact":"b3:abc""#), "got {j2}"); + } + + #[test] + fn rfc3339_format_basics() { + let ts = now_rfc3339(); + // Verifica forma `YYYY-MM-DDThh:mm:ss.nnnnnnnnnZ` minimal. + assert!(ts.len() >= 29); + assert_eq!(&ts[4..5], "-"); + assert_eq!(&ts[7..8], "-"); + assert_eq!(&ts[10..11], "T"); + assert_eq!(&ts[13..14], ":"); + assert_eq!(&ts[16..17], ":"); + assert!(ts.ends_with('Z')); + // Año razonable (2026 actualmente; este test sólo verifica que no salimos del rango + // 1970..3000). + let year: i32 = ts[..4].parse().unwrap(); + assert!((1970..3000).contains(&year), "year = {year}"); + } +} diff --git a/crates/hammer-overlay/Cargo.toml b/crates/hammer-overlay/Cargo.toml index 2862e6ac..ea5c68fd 100644 --- a/crates/hammer-overlay/Cargo.toml +++ b/crates/hammer-overlay/Cargo.toml @@ -9,11 +9,13 @@ description = "El overlay de experimentación de hammer: try/commit/discard/stat [dependencies] hammer-core.workspace = true +hammer-journal.workspace = true anyhow.workspace = true thiserror.workspace = true tracing.workspace = true serde.workspace = true serde_json.workspace = true +libc = "0.2" [dev-dependencies] tempfile.workspace = true diff --git a/crates/hammer-overlay/src/lib.rs b/crates/hammer-overlay/src/lib.rs index 3541ff93..90e89a77 100644 --- a/crates/hammer-overlay/src/lib.rs +++ b/crates/hammer-overlay/src/lib.rs @@ -218,12 +218,30 @@ pub fn discard(id: &OverlayId, state_root: &Path) -> Result<()> { Ok(()) } -/// Fusiona los `upper`s al FHS y desmonta. Cero diario por ahora — Fase 3 lo añade. +/// Fusiona los `upper`s al FHS y desmonta. Sin diario. /// -/// Estrategia: por cada mount, `rsync -a --delete-after` desde el upperdir al target. -/// Antes del rsync desmontamos el overlay; si no, rsync estaría leyendo de la vista -/// "merged" en lugar del upper puro (y promocionaríamos todo, no sólo los cambios). +/// Para registrar cada promoción en el diario (SDD 04 §2 + SDD 05), usa +/// [`commit_with_journal`]. pub fn commit(id: &OverlayId, state_root: &Path) -> Result { + commit_inner(id, state_root, None) +} + +/// Como [`commit`] pero registra cada archivo promocionado/eliminado como +/// `MutationEvent` con `source = HammerCommit { overlay }`. El SDD insiste en que el ruido +/// del experimento no entra al diario hasta el commit — esto es el momento exacto. +pub fn commit_with_journal( + id: &OverlayId, + state_root: &Path, + journal: &hammer_journal::Journal, +) -> Result { + commit_inner(id, state_root, Some(journal)) +} + +fn commit_inner( + id: &OverlayId, + state_root: &Path, + journal: Option<&hammer_journal::Journal>, +) -> Result { let base = state_root.join(id.as_str()); let state_file = base.join("state.json"); if !state_file.is_file() { @@ -246,10 +264,57 @@ pub fn commit(id: &OverlayId, state_root: &Path) -> Result { for m in &state.mounts { promote_upper_to_lower(&m.upper, &m.target, &mut report)?; } + + if let Some(j) = journal { + record_commit_to_journal(j, id, &report); + } std::fs::remove_dir_all(&base)?; Ok(report) } +fn record_commit_to_journal( + j: &hammer_journal::Journal, + id: &OverlayId, + report: &CommitReport, +) { + use hammer_journal::{Actor, MutationEvent, MutationOp, Source}; + let actor = Actor { + source: Source::HammerCommit { + overlay: id.as_str().to_string(), + artifact: None, + }, + pid: Some(std::process::id()), + uid: Some(unsafe { libc::getuid() }), + }; + let ts = hammer_journal::now_rfc3339(); + for p in &report.copied { + let ev = MutationEvent { + ts: ts.clone(), + op: MutationOp::Replace, + path: p.clone(), + by: actor.clone(), + content_hash: None, + note: None, + }; + if let Err(e) = j.record(&ev) { + tracing::warn!(error = %e, path = %p.display(), "journal: fallo al registrar commit"); + } + } + for p in &report.removed { + let ev = MutationEvent { + ts: ts.clone(), + op: MutationOp::Delete, + path: p.clone(), + by: actor.clone(), + content_hash: None, + note: None, + }; + if let Err(e) = j.record(&ev) { + tracing::warn!(error = %e, path = %p.display(), "journal: fallo al registrar remove"); + } + } +} + fn do_mount(lower: &Path, upper: &Path, work: &Path) -> Result<()> { // overlayfs requiere `,` como separador. Si alguno de los paths contiene una coma, // overlayfs sólo soporta escaparla a través de opciones específicas — rechazamos en @@ -453,4 +518,36 @@ mod tests { let err = discard(&OverlayId("nope".into()), d.path()).unwrap_err().to_string(); assert!(err.contains("no encuentro overlay"), "msg = {err}"); } + + #[test] + fn journal_hook_records_copied_and_removed() { + // No tocamos mount real: ejercitamos directamente la función privada que el commit + // delega para anotar al diario. Eso valida el contrato sin necesidad de overlayfs. + let d = tempfile::tempdir().unwrap(); + let journal = hammer_journal::Journal::open(d.path().join("journal")).unwrap(); + let report = CommitReport { + copied: vec![PathBuf::from("/etc/hosts"), PathBuf::from("/bin/grep")], + removed: vec![PathBuf::from("/bin/obsoleto")], + }; + super::record_commit_to_journal(&journal, &OverlayId("1234-1".into()), &report); + let events = journal.read_all().unwrap(); + assert_eq!(events.len(), 3, "2 copied + 1 removed = 3 eventos"); + // Los promocionados van como Replace, los removidos como Delete. + let ops_paths: Vec<_> = events + .iter() + .map(|e| (e.op, e.path.display().to_string())) + .collect(); + assert!(ops_paths.contains(&(hammer_journal::MutationOp::Replace, "/etc/hosts".into()))); + assert!(ops_paths.contains(&(hammer_journal::MutationOp::Replace, "/bin/grep".into()))); + assert!(ops_paths.contains(&(hammer_journal::MutationOp::Delete, "/bin/obsoleto".into()))); + // Todos comparten el mismo source con overlay id. + for e in &events { + match &e.by.source { + hammer_journal::Source::HammerCommit { overlay, .. } => { + assert_eq!(overlay, "1234-1"); + } + other => panic!("source inesperado: {other:?}"), + } + } + } } diff --git a/crates/hammerd/Cargo.toml b/crates/hammerd/Cargo.toml index eb8a5e22..45f7bea3 100644 --- a/crates/hammerd/Cargo.toml +++ b/crates/hammerd/Cargo.toml @@ -13,7 +13,15 @@ path = "src/main.rs" [dependencies] hammer-core.workspace = true +hammer-journal.workspace = true +hammer-overlay.workspace = true anyhow.workspace = true clap.workspace = true +thiserror.workspace = true tracing.workspace = true tracing-subscriber.workspace = true +nix.workspace = true +libc = "0.2" + +[dev-dependencies] +tempfile.workspace = true diff --git a/crates/hammerd/src/main.rs b/crates/hammerd/src/main.rs index 5f2ce63f..487443e3 100644 --- a/crates/hammerd/src/main.rs +++ b/crates/hammerd/src/main.rs @@ -1,12 +1,18 @@ //! `hammerd` — daemon de hammer. Dos responsabilidades: -//! 1. Bus de agente: /run/agent.sock (JSON-líneas, SO_PEERCRED). Ver `docs/07-agent-bus.md`. -//! 2. Diario de mutaciones: fanotify sobre /bin,/sbin,/lib,/etc. Ver `docs/05-journal.md`. +//! 1. Diario de mutaciones: fanotify sobre /bin,/sbin,/lib,/etc → `hammer-journal`. +//! 2. Bus de agente: /run/agent.sock (JSON-líneas, SO_PEERCRED). Ver `docs/07-agent-bus.md`. //! -//! Esqueleto de arranque. Las dos subsistemas (Fases 3 y 5) se implementarán por separado; -//! aquí queda el binario navegable y el cableado básico de logging/argumentos. +//! Fase 3 implementa (1). Fase 5 añadirá (2). Si el daemon arranca sin CAP_SYS_ADMIN, el +//! watcher falla en init y el daemon sigue corriendo sin él (warn claro). Esto es +//! deliberado para que puedas correr `hammerd` en dev sin sudo y aún así probar la parte +//! del bus. + +use std::path::PathBuf; use clap::Parser; +mod watcher; + #[derive(Parser)] #[command(name = "hammerd", version, about = "Daemon de hammer: bus de agente + diario.")] struct Args { @@ -19,6 +25,13 @@ struct Args { /// Directorio del diario de mutaciones. #[arg(long, default_value = "/var/lib/hammer/journal")] journal: String, + /// Directorio raíz de overlays (para que el watcher pueda filtrar eventos bajo overlays + /// activos). Default coincide con `hammer try`. + #[arg(long, default_value = hammer_overlay::DEFAULT_STATE_ROOT)] + overlay_state_root: String, + /// Directorios extra a vigilar; si vacío, usa los defaults del FHS (SDD 05 §2). + #[arg(long = "watch")] + extra_watch: Vec, } fn main() -> anyhow::Result<()> { @@ -35,11 +48,28 @@ fn main() -> anyhow::Result<()> { agent_sock = %args.agent_sock, init_control = %args.init_control, journal = %args.journal, - "hammerd: arranque (esqueleto)" + "hammerd: arranque" ); - // TODO(fase-3): iniciar el watcher fanotify → diario. - // TODO(fase-5): servir el bus de agente en agent_sock (JSON-líneas, SO_PEERCRED). - tracing::warn!("subsistemas pendientes: diario (Fase 3) y bus de agente (Fase 5)"); + let journal = hammer_journal::Journal::open(&args.journal)?; + let mut dirs = watcher::default_watch_dirs(); + dirs.extend(args.extra_watch); + + let overlay_root = PathBuf::from(&args.overlay_state_root); + match watcher::Watcher::start(&dirs, journal, Some(overlay_root)) { + Ok(w) => { + tracing::info!("watcher fanotify activo"); + w.run_forever()?; + } + Err(e) => { + tracing::warn!( + error = %e, + "watcher fanotify NO arrancó; el daemon sigue (sin diario en este run)" + ); + // TODO(fase-5): aquí entra el bus de agente. De momento sólo dormimos para que + // el daemon no salga. + std::thread::park(); + } + } Ok(()) } diff --git a/crates/hammerd/src/watcher.rs b/crates/hammerd/src/watcher.rs new file mode 100644 index 00000000..a0334ec3 --- /dev/null +++ b/crates/hammerd/src/watcher.rs @@ -0,0 +1,229 @@ +//! Watcher de mutaciones del FHS. Usa `fanotify` para observar `CLOSE_WRITE` (alguien +//! escribió en un archivo y lo cerró) sobre los directorios del SDD 05 §2. +//! +//! Diseño Fase 3 v1: +//! +//! * Marcamos como `FAN_MARK_ADD | FAN_MARK_ONLYDIR` cada directorio vigilado. La cobertura +//! por defecto (ver `default_watch_dirs`) cubre el FHS clásico de userland. +//! * Filtro mínimo: ignoramos eventos cuya ruta resuelta cae dentro de un overlay activo +//! (consultamos `hammer_overlay::status` en cada batch — para Fase 3 el coste es +//! despreciable; cuando los overlays se vuelvan numerosos cachearemos los paths). +//! * Cada evento se materializa como `MutationEvent` con `op = Replace` y `source = External`. +//! El refinamiento (Create/Edit por contenido, hashing de contenido, atribución a un +//! `hydrate` reciente) llega después. +//! +//! Privilegios: `fanotify_init` requiere `CAP_SYS_ADMIN`. Si no lo tenemos, fallamos rápido +//! con un mensaje claro y el daemon sigue arrancando los otros subsistemas. + +use std::ffi::OsString; +use std::os::fd::AsFd; +use std::os::unix::ffi::OsStringExt; +use std::path::{Path, PathBuf}; + +use hammer_journal::{Actor, Journal, MutationEvent, MutationOp, Source}; +use nix::fcntl::AT_FDCWD; +use nix::sys::fanotify::{EventFFlags, Fanotify, InitFlags, MarkFlags, MaskFlags}; + +/// Directorios del FHS que vigilamos por defecto (SDD 05 §2). +pub fn default_watch_dirs() -> Vec { + ["/bin", "/sbin", "/usr/bin", "/usr/sbin", "/lib", "/usr/lib", "/etc"] + .into_iter() + .map(PathBuf::from) + .collect() +} + +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("fanotify_init falló (¿CAP_SYS_ADMIN?): {0}")] + Init(nix::errno::Errno), + #[error("fanotify_mark falló sobre {path}: {err}")] + Mark { path: PathBuf, err: nix::errno::Errno }, + #[error("read_events: {0}")] + Read(nix::errno::Errno), + #[error("journal: {0}")] + Journal(#[from] hammer_journal::Error), + #[error("io: {0}")] + Io(#[from] std::io::Error), +} + +pub type Result = std::result::Result; + +pub struct Watcher { + fan: Fanotify, + journal: Journal, + watched: Vec, + overlay_state_root: Option, +} + +impl Watcher { + /// Inicializa el watcher y registra los `dirs` como vigilados. + /// + /// `overlay_state_root`, si está presente, se usa para filtrar eventos que ocurren + /// dentro de un overlay activo (no entran al diario hasta el `commit` del overlay). + pub fn start( + dirs: &[PathBuf], + journal: Journal, + overlay_state_root: Option, + ) -> Result { + let fan = Fanotify::init( + InitFlags::FAN_CLASS_NOTIF | InitFlags::FAN_CLOEXEC, + EventFFlags::O_RDONLY | EventFFlags::O_CLOEXEC, + ) + .map_err(Error::Init)?; + + for dir in dirs { + if !dir.is_dir() { + tracing::warn!(path = %dir.display(), "watcher: skip — no es directorio"); + continue; + } + fan.mark( + MarkFlags::FAN_MARK_ADD | MarkFlags::FAN_MARK_ONLYDIR, + MaskFlags::FAN_CLOSE_WRITE | MaskFlags::FAN_EVENT_ON_CHILD, + AT_FDCWD, + Some(dir.as_path()), + ) + .map_err(|err| Error::Mark { path: dir.clone(), err })?; + tracing::info!(path = %dir.display(), "watcher: vigilando"); + } + + Ok(Self { + fan, + journal, + watched: dirs.to_vec(), + overlay_state_root, + }) + } + + /// Devuelve un fd que se puede usar con `poll`/`epoll` para esperar eventos sin bloquear. + /// Útil cuando integremos esto con el bus de agente (Fase 5). + #[allow(dead_code)] + pub fn fd(&self) -> std::os::fd::BorrowedFd<'_> { + self.fan.as_fd() + } + + /// Loop bloqueante. Vuelve sólo si `read_events` falla irrecuperablemente. Se espera + /// invocarlo en su propio thread. + pub fn run_forever(&self) -> Result<()> { + loop { + self.poll_once()?; + } + } + + /// Una iteración del loop. Útil para tests. + pub fn poll_once(&self) -> Result { + let events = self.fan.read_events().map_err(Error::Read)?; + let overlay_targets = self.overlay_targets(); + let mut recorded = 0; + for ev in events { + let pid = ev.pid() as u32; + let Some(fd) = ev.fd() else { continue }; + let path = match path_of_fd(fd) { + Ok(p) => p, + Err(e) => { + tracing::warn!(error = %e, "watcher: no pude resolver path del fd, saltando"); + continue; + } + }; + if !path_is_under_any(&path, &self.watched) { + continue; + } + if path_is_under_any(&path, &overlay_targets) { + tracing::debug!(path = %path.display(), "watcher: ignoro evento bajo overlay activo"); + continue; + } + let me = MutationEvent { + ts: hammer_journal::now_rfc3339(), + op: MutationOp::Replace, + path, + by: Actor { + source: Source::External, + pid: Some(pid), + uid: uid_of_pid(pid), + }, + content_hash: None, + note: None, + }; + self.journal.record(&me)?; + recorded += 1; + } + Ok(recorded) + } + + fn overlay_targets(&self) -> Vec { + let Some(root) = &self.overlay_state_root else { + return Vec::new(); + }; + match hammer_overlay::status(root) { + Ok(overlays) => overlays + .into_iter() + .flat_map(|o| o.mounts.into_iter().map(|m| m.target)) + .collect(), + Err(e) => { + tracing::warn!(error = %e, "watcher: status() falló, no filtro overlays"); + Vec::new() + } + } + } +} + +/// Resuelve el path real apuntado por un fd procediendo por `/proc/self/fd/`. Devuelve el +/// path destino del symlink. +fn path_of_fd(fd: std::os::fd::BorrowedFd<'_>) -> std::io::Result { + use std::os::fd::AsRawFd; + let link = format!("/proc/self/fd/{}", fd.as_raw_fd()); + let target = std::fs::read_link(&link)?; + // readlink puede devolver paths suffix " (deleted)" si el archivo fue eliminado entre el + // evento y nuestra resolución. Los preservamos como están — el journal verá el evento + // pero el caller puede filtrarlos si quiere. + Ok(target) +} + +fn path_is_under_any(p: &Path, roots: &[PathBuf]) -> bool { + roots.iter().any(|r| p.starts_with(r)) +} + +/// Lee `/proc//status` y extrae el UID real. Devuelve `None` si el proceso desapareció. +fn uid_of_pid(pid: u32) -> Option { + let bytes = std::fs::read(format!("/proc/{pid}/status")).ok()?; + let text = OsString::from_vec(bytes); + let s = text.to_string_lossy(); + for line in s.lines() { + if let Some(rest) = line.strip_prefix("Uid:") { + let first = rest.split_whitespace().next()?; + return first.parse().ok(); + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_cover_fhs_userland() { + let d = default_watch_dirs(); + assert!(d.iter().any(|p| p == Path::new("/bin"))); + assert!(d.iter().any(|p| p == Path::new("/etc"))); + assert!(d.iter().any(|p| p == Path::new("/usr/lib"))); + assert!(!d.iter().any(|p| p == Path::new("/tmp"))); + } + + #[test] + fn path_filter_under_any() { + let roots = vec![PathBuf::from("/bin"), PathBuf::from("/etc")]; + assert!(path_is_under_any(Path::new("/bin/grep"), &roots)); + assert!(path_is_under_any(Path::new("/etc/passwd"), &roots)); + assert!(!path_is_under_any(Path::new("/tmp/x"), &roots)); + assert!(!path_is_under_any(Path::new("/binx/y"), &roots)); + } + + #[test] + fn uid_of_self_matches() { + let me = std::process::id(); + let uid = uid_of_pid(me).expect("self uid"); + // En entornos normales, comparable con getuid del proceso. + let expected = unsafe { libc::getuid() }; + assert_eq!(uid, expected); + } +} diff --git a/docs/10-roadmap.md b/docs/10-roadmap.md index 8173e4c2..f0d46ab8 100644 --- a/docs/10-roadmap.md +++ b/docs/10-roadmap.md @@ -37,19 +37,25 @@ Alpine vía `bwrap`** — confirma el corazón "fábrica funcional → FHS mutab substrato Alpine. La VM/LXC dedicada queda como ejercicio de empaque, no como pre-requisito de validación. -### Fase 2 — Overlay de experimentación ▶ *en progreso* +### Fase 2 — Overlay de experimentación ✅ - [x] Crate `hammer-overlay`: `try` / `commit` / `discard` / `status` sobre `overlayfs`. - [x] Manifiesto persistente en `//state.json`; `status` enumera. - [x] `commit` promociona archivos y procesa whiteouts (char dev 0/0) como remove. - [x] Subcomandos del CLI: `hammer try [targets…]`, `commit `, `discard `, `status`. - [x] Tests E2E con bwrap+user-ns, gateados en `HAMMER_OVERLAY_TESTS=1` (kernel-dependiente). -- [ ] Hook al diario en `commit` (queda para Fase 3). +- [x] Hook al diario en `commit` (Fase 3 lo añadió vía `commit_with_journal`). - [ ] Anidación real de overlays (un solo overlay activo por target hoy). -### Fase 3 — Diario de mutaciones -- [ ] `hammerd` con `fanotify` sobre `/bin`,`/sbin`,`/lib`,`/etc`. -- [ ] Log JSON-líneas append-only; `hammer journal`. -- **Hecho cuando:** toda mutación trazable/opaca queda registrada y consultable. +### Fase 3 — Diario de mutaciones ▶ *en progreso* +- [x] Crate `hammer-journal`: `MutationEvent`, `Source` (HammerHydrate/HammerCommit/External), + append/read/tail/follow JSON-líneas. RFC 3339 sin chrono. +- [x] `hammerd` con `fanotify` (FAN_CLOSE_WRITE) sobre `/bin`,`/sbin`,`/usr/bin`,`/usr/sbin`, + `/lib`,`/usr/lib`,`/etc`. Filtra eventos bajo overlays activos. Fallback graceful sin + CAP_SYS_ADMIN. +- [x] `hammer journal [--tail N] [--follow] [--format pretty|json]`. +- [x] `hammer commit` registra cada archivo promocionado vía `commit_with_journal`. +- [ ] Refinar op del watcher (Create vs Replace vs Edit con FAN_REPORT_DFID_NAME). +- [ ] Hash del contenido tras la mutación (`content_hash`) y de-dup idempotente. ### Fase 4 — Formato y flujo `.swm` - [ ] `Swm` (de/serialización YAML), `hammer export`, `hammer apply` (con overlay), `verify`.