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 `op = Replace` y `source = External`.
12//!   El refinamiento (Create/Edit por contenido, hashing de contenido, atribución a un
13//!   `hydrate` reciente) llega después.
14//!
15//! Privilegios: `fanotify_init` requiere `CAP_SYS_ADMIN`. Si no lo tenemos, fallamos rápido
16//! con un mensaje claro y el daemon sigue arrancando los otros subsistemas.
17
18use std::ffi::OsString;
19use std::os::fd::AsFd;
20use std::os::unix::ffi::OsStringExt;
21use std::path::{Path, PathBuf};
22
23use hammer_journal::{Actor, Journal, MutationEvent, MutationOp, Source};
24use nix::fcntl::AT_FDCWD;
25use nix::sys::fanotify::{EventFFlags, Fanotify, InitFlags, MarkFlags, MaskFlags};
26
27/// Directorios del FHS que vigilamos por defecto (SDD 05 §2).
28pub fn default_watch_dirs() -> Vec<PathBuf> {
29    ["/bin", "/sbin", "/usr/bin", "/usr/sbin", "/lib", "/usr/lib", "/etc"]
30        .into_iter()
31        .map(PathBuf::from)
32        .collect()
33}
34
35#[derive(Debug, thiserror::Error)]
36pub enum Error {
37    #[error("fanotify_init falló (¿CAP_SYS_ADMIN?): {0}")]
38    Init(nix::errno::Errno),
39    #[error("fanotify_mark falló sobre {path}: {err}")]
40    Mark { path: PathBuf, err: nix::errno::Errno },
41    #[error("read_events: {0}")]
42    Read(nix::errno::Errno),
43    #[error("journal: {0}")]
44    Journal(#[from] hammer_journal::Error),
45    #[error("io: {0}")]
46    Io(#[from] std::io::Error),
47}
48
49pub type Result<T> = std::result::Result<T, Error>;
50
51pub struct Watcher {
52    fan: Fanotify,
53    journal: Journal,
54    watched: Vec<PathBuf>,
55    overlay_state_root: Option<PathBuf>,
56    /// Si está, cada `MutationEvent` registrado se re-emite al bus de agente como
57    /// `Event::Modified` (SDD 07 §3). Sin bus, el watcher actúa como en Fase 3.
58    events: Option<crate::events::EventBus>,
59}
60
61impl Watcher {
62    /// Inicializa el watcher y registra los `dirs` como vigilados.
63    ///
64    /// `overlay_state_root`, si está presente, se usa para filtrar eventos que ocurren
65    /// dentro de un overlay activo (no entran al diario hasta el `commit` del overlay).
66    #[allow(dead_code)]
67    pub fn start(
68        dirs: &[PathBuf],
69        journal: Journal,
70        overlay_state_root: Option<PathBuf>,
71    ) -> Result<Self> {
72        Self::start_with_events(dirs, journal, overlay_state_root, None)
73    }
74
75    /// Igual que `start`, pero conecta el watcher al bus de eventos del daemon. Cada
76    /// mutación registrada se re-emite como `Event::Modified` a las conexiones abiertas.
77    pub fn start_with_events(
78        dirs: &[PathBuf],
79        journal: Journal,
80        overlay_state_root: Option<PathBuf>,
81        events: Option<crate::events::EventBus>,
82    ) -> Result<Self> {
83        let fan = Fanotify::init(
84            InitFlags::FAN_CLASS_NOTIF | InitFlags::FAN_CLOEXEC,
85            EventFFlags::O_RDONLY | EventFFlags::O_CLOEXEC,
86        )
87        .map_err(Error::Init)?;
88
89        for dir in dirs {
90            if !dir.is_dir() {
91                tracing::warn!(path = %dir.display(), "watcher: skip — no es directorio");
92                continue;
93            }
94            fan.mark(
95                MarkFlags::FAN_MARK_ADD | MarkFlags::FAN_MARK_ONLYDIR,
96                MaskFlags::FAN_CLOSE_WRITE | MaskFlags::FAN_EVENT_ON_CHILD,
97                AT_FDCWD,
98                Some(dir.as_path()),
99            )
100            .map_err(|err| Error::Mark { path: dir.clone(), err })?;
101            tracing::info!(path = %dir.display(), "watcher: vigilando");
102        }
103
104        Ok(Self {
105            fan,
106            journal,
107            watched: dirs.to_vec(),
108            overlay_state_root,
109            events,
110        })
111    }
112
113    /// Devuelve un fd que se puede usar con `poll`/`epoll` para esperar eventos sin bloquear.
114    /// Útil cuando integremos esto con el bus de agente (Fase 5).
115    #[allow(dead_code)]
116    pub fn fd(&self) -> std::os::fd::BorrowedFd<'_> {
117        self.fan.as_fd()
118    }
119
120    /// Loop bloqueante. Vuelve sólo si `read_events` falla irrecuperablemente. Se espera
121    /// invocarlo en su propio thread.
122    pub fn run_forever(&self) -> Result<()> {
123        loop {
124            self.poll_once()?;
125        }
126    }
127
128    /// Una iteración del loop. Útil para tests.
129    pub fn poll_once(&self) -> Result<usize> {
130        let events = self.fan.read_events().map_err(Error::Read)?;
131        let overlay_targets = self.overlay_targets();
132        let mut recorded = 0;
133        for ev in events {
134            let pid = ev.pid() as u32;
135            let Some(fd) = ev.fd() else { continue };
136            let path = match path_of_fd(fd) {
137                Ok(p) => p,
138                Err(e) => {
139                    tracing::warn!(error = %e, "watcher: no pude resolver path del fd, saltando");
140                    continue;
141                }
142            };
143            if !path_is_under_any(&path, &self.watched) {
144                continue;
145            }
146            if path_is_under_any(&path, &overlay_targets) {
147                tracing::debug!(path = %path.display(), "watcher: ignoro evento bajo overlay activo");
148                continue;
149            }
150            let me = MutationEvent {
151                ts: hammer_journal::now_rfc3339(),
152                op: MutationOp::Replace,
153                path,
154                by: Actor {
155                    source: Source::External,
156                    pid: Some(pid),
157                    uid: uid_of_pid(pid),
158                },
159                content_hash: None,
160                note: None,
161            };
162            self.journal.record(&me)?;
163            recorded += 1;
164            if let Some(bus) = &self.events {
165                bus.publish(&crate::proto::Event::Modified {
166                    path: me.path.display().to_string(),
167                    op: me.op.as_str().to_string(),
168                    ts: me.ts.clone(),
169                });
170            }
171        }
172        Ok(recorded)
173    }
174
175    fn overlay_targets(&self) -> Vec<PathBuf> {
176        let Some(root) = &self.overlay_state_root else {
177            return Vec::new();
178        };
179        match hammer_overlay::status(root) {
180            Ok(overlays) => overlays
181                .into_iter()
182                .flat_map(|o| o.mounts.into_iter().map(|m| m.target))
183                .collect(),
184            Err(e) => {
185                tracing::warn!(error = %e, "watcher: status() falló, no filtro overlays");
186                Vec::new()
187            }
188        }
189    }
190}
191
192/// Resuelve el path real apuntado por un fd procediendo por `/proc/self/fd/<n>`. Devuelve el
193/// path destino del symlink.
194fn path_of_fd(fd: std::os::fd::BorrowedFd<'_>) -> std::io::Result<PathBuf> {
195    use std::os::fd::AsRawFd;
196    let link = format!("/proc/self/fd/{}", fd.as_raw_fd());
197    let target = std::fs::read_link(&link)?;
198    // readlink puede devolver paths suffix " (deleted)" si el archivo fue eliminado entre el
199    // evento y nuestra resolución. Los preservamos como están — el journal verá el evento
200    // pero el caller puede filtrarlos si quiere.
201    Ok(target)
202}
203
204fn path_is_under_any(p: &Path, roots: &[PathBuf]) -> bool {
205    roots.iter().any(|r| p.starts_with(r))
206}
207
208/// Lee `/proc/<pid>/status` y extrae el UID real. Devuelve `None` si el proceso desapareció.
209fn uid_of_pid(pid: u32) -> Option<u32> {
210    let bytes = std::fs::read(format!("/proc/{pid}/status")).ok()?;
211    let text = OsString::from_vec(bytes);
212    let s = text.to_string_lossy();
213    for line in s.lines() {
214        if let Some(rest) = line.strip_prefix("Uid:") {
215            let first = rest.split_whitespace().next()?;
216            return first.parse().ok();
217        }
218    }
219    None
220}
221
222#[cfg(test)]
223mod tests {
224    use super::*;
225
226    #[test]
227    fn defaults_cover_fhs_userland() {
228        let d = default_watch_dirs();
229        assert!(d.iter().any(|p| p == Path::new("/bin")));
230        assert!(d.iter().any(|p| p == Path::new("/etc")));
231        assert!(d.iter().any(|p| p == Path::new("/usr/lib")));
232        assert!(!d.iter().any(|p| p == Path::new("/tmp")));
233    }
234
235    #[test]
236    fn path_filter_under_any() {
237        let roots = vec![PathBuf::from("/bin"), PathBuf::from("/etc")];
238        assert!(path_is_under_any(Path::new("/bin/grep"), &roots));
239        assert!(path_is_under_any(Path::new("/etc/passwd"), &roots));
240        assert!(!path_is_under_any(Path::new("/tmp/x"), &roots));
241        assert!(!path_is_under_any(Path::new("/binx/y"), &roots));
242    }
243
244    #[test]
245    fn uid_of_self_matches() {
246        let me = std::process::id();
247        let uid = uid_of_pid(me).expect("self uid");
248        // En entornos normales, comparable con getuid del proceso.
249        let expected = unsafe { libc::getuid() };
250        assert_eq!(uid, expected);
251    }
252}