sergio/takana

1//! **E3 — mirror del store content-addressed** ([SDD 13 §E3](../../docs/13-release-engineering.md)).
2//!
3//! El `/store` de takana es un CAS: cada artefacto vive en un directorio `<hash>-<name>` y *el hash es
4//! la dirección*. Un mirror, entonces, es ese CAS **replicado** entre dos máquinas + **resolución por
5//! hash**. Esta capa replica los artefactos que le faltan al destino y **ancla la integridad en
6//! `of_tree`** (el content-hash BLAKE3 del árbol, [`ArtifactHash::of_tree`]): el receptor **recomputa**
7//! `of_tree` sobre lo que copió y **exige** que case el del índice **antes** de sellar. Una transferencia
8//! corrupta (o un origen que sirvió bytes que no casan el hash anunciado) se **rechaza**, no se instala.
9//!
10//! El transporte de esta primera entrega es **sistema de ficheros**: el "remoto" es una ruta a otro
11//! store (que puede ser un montaje de red, sshfs, etc.). El transporte sobre SSH/red —que el producto ya
12//! expone vía sshd— es un envoltorio ortogonal posterior; la lógica de CAS (índice, diff, copia
13//! verificada, sellado atómico) vive aquí y es agnóstica del medio.
14//!
15//! **Modelo de confianza:** la verificación `of_tree` detecta corrupción de transporte y que el origen
16//! sirva un árbol que no case el hash que anuncia. Anclar ese hash a una raíz de confianza firmada
17//! (el log de transparencia `bootstrap.json` / la atestación de Etapa D) es la capa de endurecimiento
18//! posterior: hoy el invariante CAS que se *fuerza* es "el contenido recibido HASHEA a lo que el índice
19//! dice".
20
21use std::path::Path;
22
23use takana_core::{ArtifactHash, Store};
24use serde::{Deserialize, Serialize};
25
26#[derive(Debug, thiserror::Error)]
27pub enum Error {
28    #[error("io: {0}")]
29    Io(#[from] std::io::Error),
30    #[error("core: {0}")]
31    Core(#[from] takana_core::Error),
32    #[error(
33        "integridad: el artefacto '{dir}' recibido hashea a {got} pero el índice anuncia {want} — \
34         transferencia corrupta o manipulada; NO se selló"
35    )]
36    Verify { dir: String, want: String, got: String },
37    #[error("mirror: {0}")]
38    Other(String),
39}
40
41pub type Result<T> = std::result::Result<T, Error>;
42
43/// Una entrada del índice del store: el directorio del artefacto (`<hash>-<name>`) y su **content-hash**
44/// (`of_tree`, `b3:…`). El `dir` es la dirección (input-addressed) y `content` la huella de los bytes:
45/// dos artefactos con el mismo `dir` deben tener el mismo `content` (coherencia del CAS).
46#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
47pub struct ArtifactRef {
48    pub dir: String,
49    pub content: String,
50}
51
52/// Índice de un store: la lista de sus artefactos con su content-hash, ordenada por `dir` (estable).
53#[derive(Debug, Clone, Default, Serialize, Deserialize)]
54pub struct StoreIndex {
55    pub artifacts: Vec<ArtifactRef>,
56}
57
58impl StoreIndex {
59    /// Busca un artefacto por su `dir`.
60    pub fn get(&self, dir: &str) -> Option<&ArtifactRef> {
61        self.artifacts.iter().find(|a| a.dir == dir)
62    }
63}
64
65/// ¿El nombre de directorio tiene forma de artefacto del store? = 64 hex + `-` + `<name>` no vacío. Así
66/// se ignoran `bootstrap.json`, `.bootstrap-tmp`, índices y otros ficheros del root del store.
67fn is_artifact_dir(name: &str) -> bool {
68    let bytes = name.as_bytes();
69    bytes.len() > 65
70        && bytes[64] == b'-'
71        && bytes[..64].iter().all(|b| b.is_ascii_hexdigit())
72}
73
74/// Parte `<64hex>-<name>` en `(ArtifactHash, name)`. El `name` puede contener `-` (p.ej.
75/// `product-attested-rootfs`); sólo el primer `-`, en la posición 64, separa el hash.
76fn parse_dir(dir: &str) -> Option<(ArtifactHash, String)> {
77    if !is_artifact_dir(dir) {
78        return None;
79    }
80    Some((ArtifactHash::from_hex(&dir[..64]), dir[65..].to_string()))
81}
82
83/// Construye el índice de un store: para cada directorio-artefacto, su `of_tree`. Un store inexistente
84/// se trata como vacío (el destino aún no creado en un primer `pull`). Determinista (ordenado por `dir`).
85pub fn index(store_root: &Path) -> Result<StoreIndex> {
86    let mut artifacts = Vec::new();
87    if store_root.is_dir() {
88        for entry in std::fs::read_dir(store_root)? {
89            let entry = entry?;
90            let name = entry.file_name().to_string_lossy().into_owned();
91            if !is_artifact_dir(&name) || !entry.path().is_dir() {
92                continue;
93            }
94            let content = ArtifactHash::of_tree(&entry.path())?.as_str().to_string();
95            artifacts.push(ArtifactRef { dir: name, content });
96        }
97    }
98    artifacts.sort_by(|a, b| a.dir.cmp(&b.dir));
99    Ok(StoreIndex { artifacts })
100}
101
102/// Diferencia entre dos stores: `(faltan_en_dst, faltan_en_src, conflictos)`. Un **conflicto** es un
103/// artefacto presente en ambos con el MISMO `dir` pero DISTINTO `content` (misma dirección, otros bytes:
104/// rotura de la coherencia del CAS — no-reproducibilidad o corrupción). No copia nada.
105pub fn diff(
106    src_root: &Path,
107    dst_root: &Path,
108) -> Result<(Vec<ArtifactRef>, Vec<ArtifactRef>, Vec<String>)> {
109    let si = index(src_root)?;
110    let di = index(dst_root)?;
111    let mut missing_in_dst = Vec::new();
112    let mut conflicts = Vec::new();
113    for a in &si.artifacts {
114        match di.get(&a.dir) {
115            None => missing_in_dst.push(a.clone()),
116            Some(d) if d.content != a.content => conflicts.push(a.dir.clone()),
117            Some(_) => {}
118        }
119    }
120    let missing_in_src = di
121        .artifacts
122        .iter()
123        .filter(|d| si.get(&d.dir).is_none())
124        .cloned()
125        .collect();
126    Ok((missing_in_dst, missing_in_src, conflicts))
127}
128
129/// Reporte de un [`sync`]: qué se copió, cuántos ya estaban, conflictos y fallos de verificación.
130#[derive(Debug, Clone, Default)]
131pub struct SyncReport {
132    /// `dir`s copiados+verificados+sellados en el destino.
133    pub copied: Vec<String>,
134    /// Ya presentes con el mismo content-hash (idempotencia).
135    pub skipped_present: usize,
136    /// Presentes en ambos con content-hash distinto (CAS incoherente; NO se sobrescriben).
137    pub conflicts: Vec<String>,
138    /// `(dir, motivo)` de los que fallaron la verificación de integridad (no se sellaron).
139    pub failed: Vec<(String, String)>,
140}
141
142impl SyncReport {
143    /// Éxito = nada falló la verificación y no hubo conflictos.
144    pub fn ok(&self) -> bool {
145        self.failed.is_empty() && self.conflicts.is_empty()
146    }
147}
148
149/// Copia el árbol de un artefacto a un staging bajo el destino, **recomputa `of_tree`**, exige que case
150/// `expected_content` y sólo entonces lo **sella** en `dst_store` (rename atómico + read-only, vía
151/// [`Store::seal`]). El staging vive bajo el root del destino ⇒ mismo filesystem ⇒ el rename del seal es
152/// atómico; ante un hash que no casa, se borra el staging y NO se instala nada.
153fn install_verified(
154    src_dir: &Path,
155    dst_store: &Store,
156    h: &ArtifactHash,
157    name: &str,
158    expected_content: &str,
159) -> Result<()> {
160    let dir_label = format!("{}-{name}", h.as_str().trim_start_matches("b3:"));
161    let staging = dst_store
162        .root()
163        .join(".mirror-tmp")
164        .join(format!("{dir_label}.incoming"));
165    let _ = std::fs::remove_dir_all(&staging);
166    if let Some(parent) = staging.parent() {
167        std::fs::create_dir_all(parent)?;
168    }
169
170    let result = (|| -> Result<()> {
171        copy_tree(src_dir, &staging)?;
172        let got = ArtifactHash::of_tree(&staging)?;
173        if got.as_str() != expected_content {
174            return Err(Error::Verify {
175                dir: dir_label.clone(),
176                want: expected_content.to_string(),
177                got: got.as_str().to_string(),
178            });
179        }
180        dst_store.seal(&staging, h, name)?;
181        Ok(())
182    })();
183    let _ = std::fs::remove_dir_all(&staging);
184    result
185}
186
187/// **Sincroniza** `src`→`dst`: copia a `dst` cada artefacto que le falta (verificando `of_tree` antes de
188/// sellar), salta los ya presentes con el mismo contenido y reporta conflictos (mismo `dir`, otro
189/// contenido). Idempotente: re-correr sobre un destino ya al día copia cero. Es el motor de `push`
190/// (`sync(local, remoto)`) y `pull` (`sync(remoto, local)`).
191pub fn sync(src_root: &Path, dst_root: &Path) -> Result<SyncReport> {
192    let si = index(src_root)?;
193    let di = index(dst_root)?;
194    let dst_store = Store::open(dst_root)?;
195    let mut report = SyncReport::default();
196
197    for a in &si.artifacts {
198        match di.get(&a.dir) {
199            Some(d) if d.content == a.content => {
200                report.skipped_present += 1;
201                continue;
202            }
203            Some(_) => {
204                report.conflicts.push(a.dir.clone());
205                continue;
206            }
207            None => {}
208        }
209        let (h, name) = match parse_dir(&a.dir) {
210            Some(p) => p,
211            None => continue, // index() ya filtra, defensivo
212        };
213        match install_verified(&src_root.join(&a.dir), &dst_store, &h, &name, &a.content) {
214            Ok(()) => {
215                tracing::info!(dir = %a.dir, "mirror: artefacto replicado y verificado");
216                report.copied.push(a.dir.clone());
217            }
218            Err(e) => {
219                tracing::warn!(dir = %a.dir, error = %e, "mirror: fallo de replicación");
220                report.failed.push((a.dir.clone(), e.to_string()));
221            }
222        }
223    }
224    Ok(report)
225}
226
227/// Copia recursiva `src`→`dst` preservando symlinks (target literal) y modos (incl. bit de ejecución,
228/// que `of_tree` mira). El destino se asume inexistente (staging recién limpiado).
229fn copy_tree(src: &Path, dst: &Path) -> Result<()> {
230    let meta = std::fs::symlink_metadata(src)?;
231    let ft = meta.file_type();
232    if ft.is_symlink() {
233        let target = std::fs::read_link(src)?;
234        std::os::unix::fs::symlink(target, dst)?;
235    } else if ft.is_dir() {
236        std::fs::create_dir_all(dst)?;
237        for entry in std::fs::read_dir(src)? {
238            let entry = entry?;
239            copy_tree(&entry.path(), &dst.join(entry.file_name()))?;
240        }
241    } else {
242        std::fs::copy(src, dst)?;
243    }
244    Ok(())
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250
251    /// Sella un artefacto sintético `<hex>-<name>` en `root` con los ficheros que pinte `build`.
252    fn seal(root: &Path, hex: &str, name: &str, build: impl Fn(&Path)) -> String {
253        let store = Store::open(root).unwrap();
254        let h = ArtifactHash::from_hex(hex);
255        let staging = root.join(".bootstrap-tmp").join(h.store_dir_name(name));
256        let _ = std::fs::remove_dir_all(&staging);
257        std::fs::create_dir_all(&staging).unwrap();
258        build(&staging);
259        store.seal(&staging, &h, name).unwrap();
260        h.store_dir_name(name)
261    }
262
263    fn h64(prefix: &str) -> String {
264        format!("{prefix}{}", "0".repeat(64 - prefix.len()))
265    }
266
267    #[test]
268    fn artifact_dir_shape_filters_non_artifacts() {
269        assert!(is_artifact_dir(&format!("{}-busybox", h64("ab"))));
270        assert!(is_artifact_dir(&format!("{}-product-attested-rootfs", h64("ff"))));
271        assert!(!is_artifact_dir("bootstrap.json"));
272        assert!(!is_artifact_dir(".bootstrap-tmp"));
273        assert!(!is_artifact_dir(&format!("{}-", h64("ab"))), "sin name");
274        assert!(!is_artifact_dir(&format!("{}xy-z", h64("ab"))), "65 chars no-hex antes del -");
275        // parse_dir respeta los '-' del name.
276        let (hh, name) = parse_dir(&format!("{}-find-utils-xargs", h64("cd"))).unwrap();
277        assert_eq!(name, "find-utils-xargs");
278        assert!(hh.as_str().starts_with("b3:"));
279    }
280
281    #[test]
282    fn index_lists_artifacts_with_content_hash_and_skips_junk() {
283        let tmp = tempfile::tempdir().unwrap();
284        let root = tmp.path();
285        seal(root, &h64("aa"), "busybox", |w| {
286            std::fs::write(w.join("bin"), b"\x7fELF").unwrap();
287        });
288        // ruido del root del store que NO debe entrar al índice.
289        std::fs::write(root.join("bootstrap.json"), b"{}").unwrap();
290        std::fs::create_dir_all(root.join(".bootstrap-tmp")).unwrap();
291
292        let idx = index(root).unwrap();
293        assert_eq!(idx.artifacts.len(), 1, "sólo el artefacto, no bootstrap.json ni .bootstrap-tmp");
294        assert!(idx.artifacts[0].dir.ends_with("-busybox"));
295        assert!(idx.artifacts[0].content.starts_with("b3:"));
296        // content == of_tree real del dir.
297        let ot = ArtifactHash::of_tree(&root.join(&idx.artifacts[0].dir)).unwrap();
298        assert_eq!(idx.artifacts[0].content, ot.as_str());
299    }
300
301    #[test]
302    fn sync_replicates_missing_and_is_idempotent() {
303        let src = tempfile::tempdir().unwrap();
304        let dst = tempfile::tempdir().unwrap();
305        seal(src.path(), &h64("aa"), "busybox", |w| {
306            std::fs::write(w.join("bin"), b"\x7fELF-bb").unwrap();
307            std::os::unix::fs::symlink("bin", w.join("sh")).unwrap();
308        });
309        seal(src.path(), &h64("bb"), "uutils", |w| {
310            std::fs::create_dir_all(w.join("usr/bin")).unwrap();
311            std::fs::write(w.join("usr/bin/coreutils"), b"\x7fELF-uu").unwrap();
312        });
313
314        let r = sync(src.path(), dst.path()).unwrap();
315        assert!(r.ok());
316        assert_eq!(r.copied.len(), 2);
317        // el destino quedó con los mismos artefactos y mismos content-hash.
318        let (md, ms, conf) = diff(src.path(), dst.path()).unwrap();
319        assert!(md.is_empty() && conf.is_empty(), "destino al día");
320        assert!(ms.is_empty(), "el destino no tiene de más");
321        // los bytes (y el symlink) viajaron.
322        let bbdir = dst.path().join(format!("{}-busybox", h64("aa")));
323        assert_eq!(std::fs::read(bbdir.join("bin")).unwrap(), b"\x7fELF-bb");
324        assert_eq!(std::fs::read_link(bbdir.join("sh")).unwrap(), std::path::PathBuf::from("bin"));
325
326        // 2ª corrida: idempotente, copia 0.
327        let r2 = sync(src.path(), dst.path()).unwrap();
328        assert!(r2.copied.is_empty() && r2.skipped_present == 2);
329        assert!(!dst.path().join(".mirror-tmp").join(format!("{}-busybox.incoming", h64("aa"))).exists(),
330            "el staging se limpió");
331    }
332
333    #[test]
334    fn sync_rejects_corrupted_transfer_without_sealing() {
335        // El guardrail de integridad: si lo que llega NO hashea a lo que el índice anuncia, se rechaza.
336        let dst = tempfile::tempdir().unwrap();
337        let src_dir = tempfile::tempdir().unwrap();
338        std::fs::write(src_dir.path().join("bin"), b"bytes-reales").unwrap();
339        let store = Store::open(dst.path()).unwrap();
340        let h = ArtifactHash::from_hex(&h64("cc"));
341        // expected_content MENTIROSO (no es el of_tree real del src) ⇒ debe fallar Verify y no sellar.
342        let err = install_verified(src_dir.path(), &store, &h, "evil", "b3:deadbeef").unwrap_err();
343        assert!(matches!(err, Error::Verify { .. }), "esperaba Verify, vino {err}");
344        assert!(!store.has(&h, "evil"), "un artefacto que no verifica NO se sella");
345        assert!(!dst.path().join(".mirror-tmp").join(format!("{}-evil.incoming", h64("cc"))).exists(),
346            "staging limpiado tras el rechazo");
347    }
348
349    #[test]
350    fn sync_flags_conflicts_and_does_not_overwrite() {
351        // Mismo dir en src y dst pero contenido distinto = conflicto (CAS incoherente). No se pisa.
352        let src = tempfile::tempdir().unwrap();
353        let dst = tempfile::tempdir().unwrap();
354        seal(src.path(), &h64("dd"), "x", |w| std::fs::write(w.join("f"), b"SRC").unwrap());
355        seal(dst.path(), &h64("dd"), "x", |w| std::fs::write(w.join("f"), b"DST-distinto").unwrap());
356
357        let r = sync(src.path(), dst.path()).unwrap();
358        assert!(!r.ok(), "hay conflicto ⇒ no-ok");
359        assert_eq!(r.conflicts.len(), 1);
360        assert!(r.copied.is_empty());
361        // el destino conserva SUS bytes (no se sobrescribió).
362        assert_eq!(std::fs::read(dst.path().join(format!("{}-x", h64("dd"))).join("f")).unwrap(), b"DST-distinto");
363    }
364}