sergio/takana

1//! Watcher de mutaciones del FHS. Usa `fanotify` para observar `CLOSE_WRITE` (alguien
2//! escribió en un archivo y lo cerró) sobre los directorios del SDD 05 §2.
3//!
4//! Diseño Fase 3 v1:
5//!
6//! * Marcamos como `FAN_MARK_ADD | FAN_MARK_ONLYDIR` cada directorio vigilado. La cobertura
7//!   por defecto (ver `default_watch_dirs`) cubre el FHS clásico de userland.
8//! * Filtro mínimo: ignoramos eventos cuya ruta resuelta cae dentro de un overlay activo
9//!   (consultamos `hammer_overlay::status` en cada batch — para Fase 3 el coste es
10//!   despreciable; cuando los overlays se vuelvan numerosos cachearemos los paths).
11//! * Cada evento se materializa como `MutationEvent` con `source = External` y la `op`
12//!   se deriva (`classify_op`): si el fd resuelve a un path borrado (`" (deleted)"`) es
13//!   `Delete`; si el diario ya conocía ese path (con un evento previo no-`Delete`) es
14//!   `Edit`; en otro caso es `Create` (primera mutación que hammer observa sobre el path).
15//!   Nota: `FAN_CLOSE_WRITE` no distingue un `rename`-sobre-destino de una reescritura in
16//!   situ, ni reporta el nombre en eventos de directorio — la atribución plena de
17//!   `Create`/`Move`/`Delete` vía `FAN_REPORT_DFID_NAME` (con `open_by_handle_at` y
18//!   `CAP_DAC_READ_SEARCH`) queda para el track posterior; `nix` 0.30 aún no parsea los
19//!   info-records de FID, así que exigiría libc crudo y perdería el fd que hoy hashea el
20//!   contenido.
21//!
22//! Privilegios: `fanotify_init` requiere `CAP_SYS_ADMIN`. Si no lo tenemos, fallamos rápido
23//! con un mensaje claro y el daemon sigue arrancando los otros subsistemas.
24
25use std::ffi::OsString;
26use std::os::fd::AsFd;
27use std::os::unix::ffi::OsStringExt;
28use std::path::{Path, PathBuf};
29
30use hammer_journal::{Actor, Journal, MutationEvent, MutationOp, Source};
31use nix::fcntl::AT_FDCWD;
32use nix::sys::fanotify::{EventFFlags, Fanotify, InitFlags, MarkFlags, MaskFlags};
33
34/// Directorios del FHS que vigilamos por defecto (SDD 05 §2).
35pub fn default_watch_dirs() -> Vec<PathBuf> {
36    ["/bin", "/sbin", "/usr/bin", "/usr/sbin", "/lib", "/usr/lib", "/etc"]
37        .into_iter()
38        .map(PathBuf::from)
39        .collect()
40}
41
42#[derive(Debug, thiserror::Error)]
43pub enum Error {
44    #[error("fanotify_init falló (¿CAP_SYS_ADMIN?): {0}")]
45    Init(nix::errno::Errno),
46    #[error("fanotify_mark falló sobre {path}: {err}")]
47    Mark { path: PathBuf, err: nix::errno::Errno },
48    #[error("read_events: {0}")]
49    Read(nix::errno::Errno),
50    #[error("journal: {0}")]
51    Journal(#[from] hammer_journal::Error),
52    #[error("io: {0}")]
53    Io(#[from] std::io::Error),
54}
55
56pub type Result<T> = std::result::Result<T, Error>;
57
58pub struct Watcher {
59    fan: Fanotify,
60    journal: Journal,
61    watched: Vec<PathBuf>,
62    overlay_state_root: Option<PathBuf>,
63    /// Si está, cada `MutationEvent` registrado se re-emite al bus de agente como
64    /// `Event::Modified` (SDD 07 §3). Sin bus, el watcher actúa como en Fase 3.
65    events: Option<crate::events::EventBus>,
66}
67
68impl Watcher {
69    /// Inicializa el watcher y registra los `dirs` como vigilados.
70    ///
71    /// `overlay_state_root`, si está presente, se usa para filtrar eventos que ocurren
72    /// dentro de un overlay activo (no entran al diario hasta el `commit` del overlay).
73    #[allow(dead_code)]
74    pub fn start(
75        dirs: &[PathBuf],
76        journal: Journal,
77        overlay_state_root: Option<PathBuf>,
78    ) -> Result<Self> {
79        Self::start_with_events(dirs, journal, overlay_state_root, None)
80    }
81
82    /// Igual que `start`, pero conecta el watcher al bus de eventos del daemon. Cada
83    /// mutación registrada se re-emite como `Event::Modified` a las conexiones abiertas.
84    pub fn start_with_events(
85        dirs: &[PathBuf],
86        journal: Journal,
87        overlay_state_root: Option<PathBuf>,
88        events: Option<crate::events::EventBus>,
89    ) -> Result<Self> {
90        let fan = Fanotify::init(
91            InitFlags::FAN_CLASS_NOTIF | InitFlags::FAN_CLOEXEC,
92            EventFFlags::O_RDONLY | EventFFlags::O_CLOEXEC,
93        )
94        .map_err(Error::Init)?;
95
96        for dir in dirs {
97            if !dir.is_dir() {
98                tracing::warn!(path = %dir.display(), "watcher: skip — no es directorio");
99                continue;
100            }
101            fan.mark(
102                MarkFlags::FAN_MARK_ADD | MarkFlags::FAN_MARK_ONLYDIR,
103                MaskFlags::FAN_CLOSE_WRITE | MaskFlags::FAN_EVENT_ON_CHILD,
104                AT_FDCWD,
105                Some(dir.as_path()),
106            )
107            .map_err(|err| Error::Mark { path: dir.clone(), err })?;
108            tracing::info!(path = %dir.display(), "watcher: vigilando");
109        }
110
111        Ok(Self {
112            fan,
113            journal,
114            watched: dirs.to_vec(),
115            overlay_state_root,
116            events,
117        })
118    }
119
120    /// Devuelve un fd que se puede usar con `poll`/`epoll` para esperar eventos sin bloquear.
121    /// Útil cuando integremos esto con el bus de agente (Fase 5).
122    #[allow(dead_code)]
123    pub fn fd(&self) -> std::os::fd::BorrowedFd<'_> {
124        self.fan.as_fd()
125    }
126
127    /// Loop bloqueante. Vuelve sólo si `read_events` falla irrecuperablemente. Se espera
128    /// invocarlo en su propio thread.
129    pub fn run_forever(&self) -> Result<()> {
130        loop {
131            self.poll_once()?;
132        }
133    }
134
135    /// Una iteración del loop. Útil para tests.
136    pub fn poll_once(&self) -> Result<usize> {
137        let events = self.fan.read_events().map_err(Error::Read)?;
138        let overlay_targets = self.overlay_targets();
139        let mut recorded = 0;
140        for ev in events {
141            let pid = ev.pid() as u32;
142            let Some(fd) = ev.fd() else { continue };
143            let (path, deleted) = match path_of_fd(fd) {
144                Ok(p) => p,
145                Err(e) => {
146                    tracing::warn!(error = %e, "watcher: no pude resolver path del fd, saltando");
147                    continue;
148                }
149            };
150            if !path_is_under_any(&path, &self.watched) {
151                continue;
152            }
153            if path_is_under_any(&path, &overlay_targets) {
154                tracing::debug!(path = %path.display(), "watcher: ignoro evento bajo overlay activo");
155                continue;
156            }
157            // Hash del contenido tras la mutación: alimenta el de-dup idempotente y, más
158            // tarde, la correlación en `export`. Si el archivo se borró (o no podemos leerlo
159            // por permiso) seguimos con `None` — el evento sigue siendo información válida.
160            let content_hash = if deleted {
161                None
162            } else {
163                match hammer_journal::hash_file(&path) {
164                    Ok(h) => h,
165                    Err(e) => {
166                        tracing::warn!(path = %path.display(), error = %e, "watcher: no pude hashear contenido");
167                        None
168                    }
169                }
170            };
171            let op = self.classify_op(&path, deleted)?;
172            let me = MutationEvent {
173                ts: hammer_journal::now_rfc3339(),
174                op,
175                path,
176                by: Actor {
177                    source: Source::External,
178                    pid: Some(pid),
179                    uid: uid_of_pid(pid),
180                },
181                content_hash,
182                note: None,
183            };
184            // De-dup: una reescritura que deja el archivo idéntico no ensucia el diario ni
185            // despierta al bus. Sólo emitimos `Modified` cuando algo cambió de verdad.
186            if !self.journal.record_dedup(&me)? {
187                continue;
188            }
189            recorded += 1;
190            if let Some(bus) = &self.events {
191                bus.publish(&hammer_core::proto::Event::Modified {
192                    path: me.path.display().to_string(),
193                    op: me.op.as_str().to_string(),
194                    ts: me.ts.clone(),
195                });
196            }
197        }
198        Ok(recorded)
199    }
200
201    /// Deriva la `op` de un `CLOSE_WRITE` a partir del estado del fd y del diario:
202    ///
203    /// * `deleted` (el fd resuelve a `" (deleted)"`) ⇒ `Delete`.
204    /// * el diario ya tiene un evento previo no-`Delete` para el path ⇒ `Edit` (el archivo
205    ///   ya era conocido y se reescribió).
206    /// * en otro caso ⇒ `Create` — la primera mutación que hammer observa sobre el path, o
207    ///   una resurrección tras un `Delete` previo.
208    ///
209    /// No es una verdad absoluta del FS (un archivo preexistente al daemon, editado por
210    /// primera vez, se ve como `Create`); es la divergencia que el diario puede atestiguar.
211    fn classify_op(&self, path: &Path, deleted: bool) -> Result<MutationOp> {
212        classify_op(&self.journal, path, deleted)
213    }
214
215    fn overlay_targets(&self) -> Vec<PathBuf> {
216        let Some(root) = &self.overlay_state_root else {
217            return Vec::new();
218        };
219        match hammer_overlay::status(root) {
220            Ok(overlays) => overlays
221                .into_iter()
222                .flat_map(|o| o.mounts.into_iter().map(|m| m.target))
223                .collect(),
224            Err(e) => {
225                tracing::warn!(error = %e, "watcher: status() falló, no filtro overlays");
226                Vec::new()
227            }
228        }
229    }
230}
231
232/// Resuelve el path real apuntado por un fd procediendo por `/proc/self/fd/<n>`. Devuelve el
233/// path destino del symlink y un flag `deleted`: `readlink` añade el sufijo `" (deleted)"`
234/// cuando el archivo se desvinculó entre el evento y nuestra resolución. Lo recortamos para
235/// que el path entre limpio al diario y propagamos el flag para clasificar la `op` como
236/// `Delete`.
237fn path_of_fd(fd: std::os::fd::BorrowedFd<'_>) -> std::io::Result<(PathBuf, bool)> {
238    use std::os::fd::AsRawFd;
239    use std::os::unix::ffi::OsStrExt;
240    let link = format!("/proc/self/fd/{}", fd.as_raw_fd());
241    let target = std::fs::read_link(&link)?;
242    const SUFFIX: &[u8] = b" (deleted)";
243    let bytes = target.as_os_str().as_bytes();
244    if let Some(trimmed) = bytes.strip_suffix(SUFFIX) {
245        return Ok((PathBuf::from(std::ffi::OsStr::from_bytes(trimmed)), true));
246    }
247    Ok((target, false))
248}
249
250fn path_is_under_any(p: &Path, roots: &[PathBuf]) -> bool {
251    roots.iter().any(|r| p.starts_with(r))
252}
253
254/// Deriva la `op` de un `CLOSE_WRITE` (ver [`Watcher::classify_op`]). Libre de estado del
255/// fanotify para poder probarla sin `CAP_SYS_ADMIN`.
256fn classify_op(journal: &Journal, path: &Path, deleted: bool) -> Result<MutationOp> {
257    if deleted {
258        return Ok(MutationOp::Delete);
259    }
260    Ok(match journal.last_for_path(path)? {
261        Some(ev) if ev.op != MutationOp::Delete => MutationOp::Edit,
262        _ => MutationOp::Create,
263    })
264}
265
266/// Lee `/proc/<pid>/status` y extrae el UID real. Devuelve `None` si el proceso desapareció.
267fn uid_of_pid(pid: u32) -> Option<u32> {
268    let bytes = std::fs::read(format!("/proc/{pid}/status")).ok()?;
269    let text = OsString::from_vec(bytes);
270    let s = text.to_string_lossy();
271    for line in s.lines() {
272        if let Some(rest) = line.strip_prefix("Uid:") {
273            let first = rest.split_whitespace().next()?;
274            return first.parse().ok();
275        }
276    }
277    None
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283
284    #[test]
285    fn defaults_cover_fhs_userland() {
286        let d = default_watch_dirs();
287        assert!(d.iter().any(|p| p == Path::new("/bin")));
288        assert!(d.iter().any(|p| p == Path::new("/etc")));
289        assert!(d.iter().any(|p| p == Path::new("/usr/lib")));
290        assert!(!d.iter().any(|p| p == Path::new("/tmp")));
291    }
292
293    #[test]
294    fn path_filter_under_any() {
295        let roots = vec![PathBuf::from("/bin"), PathBuf::from("/etc")];
296        assert!(path_is_under_any(Path::new("/bin/grep"), &roots));
297        assert!(path_is_under_any(Path::new("/etc/passwd"), &roots));
298        assert!(!path_is_under_any(Path::new("/tmp/x"), &roots));
299        assert!(!path_is_under_any(Path::new("/binx/y"), &roots));
300    }
301
302    fn me(op: MutationOp, path: &str, hash: Option<&str>) -> MutationEvent {
303        MutationEvent {
304            ts: hammer_journal::now_rfc3339(),
305            op,
306            path: PathBuf::from(path),
307            by: Actor { source: Source::External, pid: Some(1), uid: Some(0) },
308            content_hash: hash.map(str::to_string),
309            note: None,
310        }
311    }
312
313    #[test]
314    fn classify_op_deleted_is_delete() {
315        let d = tempfile::tempdir().unwrap();
316        let j = Journal::open(d.path()).unwrap();
317        assert_eq!(classify_op(&j, Path::new("/etc/x"), true).unwrap(), MutationOp::Delete);
318    }
319
320    #[test]
321    fn classify_op_unknown_path_is_create() {
322        let d = tempfile::tempdir().unwrap();
323        let j = Journal::open(d.path()).unwrap();
324        assert_eq!(classify_op(&j, Path::new("/bin/new"), false).unwrap(), MutationOp::Create);
325    }
326
327    #[test]
328    fn classify_op_known_path_is_edit() {
329        let d = tempfile::tempdir().unwrap();
330        let j = Journal::open(d.path()).unwrap();
331        j.record(&me(MutationOp::Create, "/etc/hosts", Some("b3:aaa"))).unwrap();
332        assert_eq!(classify_op(&j, Path::new("/etc/hosts"), false).unwrap(), MutationOp::Edit);
333    }
334
335    #[test]
336    fn classify_op_resurrection_after_delete_is_create() {
337        let d = tempfile::tempdir().unwrap();
338        let j = Journal::open(d.path()).unwrap();
339        j.record(&me(MutationOp::Create, "/etc/svc", Some("b3:aaa"))).unwrap();
340        j.record(&me(MutationOp::Delete, "/etc/svc", None)).unwrap();
341        assert_eq!(classify_op(&j, Path::new("/etc/svc"), false).unwrap(), MutationOp::Create);
342    }
343
344    #[test]
345    fn uid_of_self_matches() {
346        let me = std::process::id();
347        let uid = uid_of_pid(me).expect("self uid");
348        // En entornos normales, comparable con getuid del proceso.
349        let expected = unsafe { libc::getuid() };
350        assert_eq!(uid, expected);
351    }
352}