caboodle/
pendant_sync.rs

1//! Repository pendant discovery. Trust first; composition owns verification and CURRENT.
2use crate::pack_trust::{self, Repo, TrustConfig};
3use anyhow::{bail, Context, Result};
4use fs2::FileExt;
5use serde::{Deserialize, Serialize};
6use sha2::{Digest, Sha256};
7use std::{
8    collections::BTreeMap,
9    fs,
10    io::{Read, Write},
11    path::{Path, PathBuf},
12    process::Command,
13};
14
15const MAX_ARCHIVES: usize = 64;
16const MAX_BYTES: u64 = 256 * 1024 * 1024;
17const AUTHORITY_KEY: &str = "_caboodle-authority";
18
19pub struct Options<'a> {
20    pub repo: &'a Path,
21    pub db: Option<&'a Path>,
22    pub shapes_from: Option<&'a Path>,
23    pub trust_file: Option<&'a Path>,
24    pub monorepo: bool,
25}
26
27#[derive(Debug, Serialize)]
28pub struct Report {
29    pub outcome: &'static str,
30    pub packs: usize,
31    pub current: Option<serde_json::Value>,
32}
33
34#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
35#[serde(deny_unknown_fields)]
36struct Stamp {
37    bytes: u64,
38    modified: String,
39}
40
41#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
42#[serde(deny_unknown_fields)]
43struct Receipt {
44    version: u8,
45    #[serde(default, skip_serializing_if = "is_false")]
46    monorepo: bool,
47    #[serde(default, skip_serializing_if = "Vec::is_empty")]
48    selected: Vec<String>,
49    repo: String,
50    db: PathBuf,
51    hashes: BTreeMap<String, String>,
52    authority: String,
53    database: Option<Stamp>,
54    wal: Option<Stamp>,
55}
56
57fn is_false(value: &bool) -> bool {
58    !value
59}
60
61pub(crate) fn git_command(root: &Path) -> Command {
62    let mut command = Command::new("git");
63    command.arg("-C").arg(root);
64    // A hook may export these for its own repository; --repo remains authoritative.
65    for key in [
66        "GIT_DIR",
67        "GIT_WORK_TREE",
68        "GIT_COMMON_DIR",
69        "GIT_INDEX_FILE",
70    ] {
71        command.env_remove(key);
72    }
73    command
74}
75
76pub(crate) fn git(root: &Path, args: &[&str]) -> Result<String> {
77    let output = git_command(root).args(args).output()?;
78    if !output.status.success() {
79        bail!(
80            "git {} failed: {}",
81            args.join(" "),
82            String::from_utf8_lossy(&output.stderr)
83        );
84    }
85    Ok(String::from_utf8(output.stdout)?
86        .trim_end_matches(['\r', '\n'])
87        .into())
88}
89
90fn sidecar(db: &Path, suffix: &str) -> Result<PathBuf> {
91    let mut name = db
92        .file_name()
93        .context("database needs a file name")?
94        .to_os_string();
95    name.push(suffix);
96    Ok(db.with_file_name(name))
97}
98
99fn stamp(path: &Path) -> Result<Option<Stamp>> {
100    match fs::symlink_metadata(path) {
101        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
102        Err(e) => Err(e.into()),
103        Ok(m) => {
104            if !m.is_file() || m.file_type().is_symlink() {
105                bail!("expected a regular file: {}", path.display());
106            }
107            Ok(Some(Stamp {
108                bytes: m.len(),
109                modified: format!("{:?}", m.modified()?),
110            }))
111        }
112    }
113}
114
115fn regular_directory(path: &Path) -> Result<bool> {
116    match fs::symlink_metadata(path) {
117        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
118        Err(e) => Err(e.into()),
119        Ok(m) if m.is_dir() && !m.file_type().is_symlink() => Ok(true),
120        Ok(_) => bail!(
121            "pendant directory must not be a symlink: {}",
122            path.display()
123        ),
124    }
125}
126
127fn discover(root: &Path) -> Result<Vec<PathBuf>> {
128    let quipu = root.join(".quipu");
129    let packs = quipu.join("packs");
130    if !regular_directory(&quipu)? || !regular_directory(&packs)? {
131        return Ok(Vec::new());
132    }
133    let mut found = Vec::new();
134    for entry in fs::read_dir(&packs)? {
135        let entry = entry?;
136        let name = entry.file_name();
137        let name = name.to_str().context("pendant file name must be UTF-8")?;
138        if !name.ends_with(".pendant.tar.gz") {
139            continue;
140        }
141        if name.len() > 200
142            || !name
143                .chars()
144                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
145        {
146            bail!("invalid pendant file name: {name}");
147        }
148        if !entry.file_type()?.is_file() {
149            bail!("pendant must be a regular file, not a symlink: {name}");
150        }
151        found.push(entry.path());
152        if found.len() > MAX_ARCHIVES {
153            bail!("more than {MAX_ARCHIVES} pendants in {}", packs.display());
154        }
155    }
156    found.sort();
157    Ok(found)
158}
159
160/// Freeze the bytes that trust and compose both consume. The caller owns the authority path.
161fn freeze(source: &Path, target: &Path, budget: &mut u64) -> Result<String> {
162    if !fs::symlink_metadata(source)?.is_file() {
163        bail!("not a regular pendant: {}", source.display());
164    }
165    let mut input = fs::File::open(source)?;
166    let mut output = fs::OpenOptions::new()
167        .write(true)
168        .create_new(true)
169        .open(target)?;
170    let mut hash = Sha256::new();
171    let mut buffer = [0u8; 65536];
172    loop {
173        let n = input.read(&mut buffer)?;
174        if n == 0 {
175            break;
176        }
177        *budget = budget
178            .checked_sub(n as u64)
179            .context("pendant bytes exceed the 256 MiB sync limit")?;
180        output.write_all(&buffer[..n])?;
181        hash.update(&buffer[..n]);
182    }
183    output.sync_all()?;
184    Ok(format!("{:x}", hash.finalize()))
185}
186
187pub fn sync(options: &Options<'_>) -> Result<Report> {
188    let root = PathBuf::from(git(options.repo, &["rev-parse", "--show-toplevel"])?);
189    let directory = PathBuf::from(git(&root, &["rev-parse", "--absolute-git-dir"])?);
190    let default_db = directory.join("caboodle-pendants/store.db");
191    let db = options.db.unwrap_or(&default_db);
192    let db = if db.is_absolute() {
193        db.to_path_buf()
194    } else {
195        std::env::current_dir()?.join(db)
196    };
197    let receipt_path = sidecar(&db, ".caboodle-packs.json")?;
198    let packs = discover(&root)?;
199    if packs.is_empty() && !receipt_path.exists() {
200        return Ok(Report {
201            outcome: "no_packs",
202            packs: 0,
203            current: None,
204        });
205    }
206    let remote = git(&root, &["remote", "get-url", "origin"])?;
207    let repo = Repo::parse(&remote)?.key();
208    let authority = options.shapes_from.context(
209        "packs sync requires --shapes-from pointing to an operator-controlled authority pendant",
210    )?;
211    let trust_path = match options.trust_file {
212        Some(p) => p.to_path_buf(),
213        None => pack_trust::default_path()?,
214    };
215    let trust = TrustConfig::load(&trust_path)?;
216    let frozen = tempfile::tempdir()?;
217    let mut budget = MAX_BYTES;
218    let mut hashes = BTreeMap::new();
219    let mut inputs = Vec::new();
220    for path in &packs {
221        let name = path.file_name().unwrap().to_str().unwrap();
222        let target = frozen.path().join(name);
223        let hash = freeze(path, &target, &mut budget)?;
224        match trust.decide_pack(&remote, name, &hash) {
225            pack_trust::Decision::Allow { .. } => (),
226            pack_trust::Decision::Refuse { message } => bail!("{message}"),
227        }
228        hashes.insert(name.into(), hash);
229        inputs.push((name.to_owned(), target));
230    }
231    let all_inputs = inputs.clone();
232    if options.monorepo {
233        inputs = crate::pendant_projects::select(&root, inputs)?;
234    }
235    let loaded_packs = inputs.len();
236    fs::create_dir(frozen.path().join("operator"))?;
237    let authority_copy = frozen.path().join("operator/authority.pendant.tar.gz");
238    let authority_hash = freeze(authority, &authority_copy, &mut budget)?;
239    let parent = db.parent().context("database needs a parent directory")?;
240    fs::create_dir_all(parent)?;
241    let lock_path = sidecar(&db, ".caboodle-packs.lock")?;
242    if fs::symlink_metadata(&lock_path).is_ok_and(|m| !m.is_file() || m.file_type().is_symlink()) {
243        bail!("invalid pendant lock file");
244    }
245    let lock = fs::OpenOptions::new()
246        .create(true)
247        .truncate(false)
248        .read(true)
249        .write(true)
250        .open(lock_path)?;
251    lock.try_lock_exclusive()
252        .context("another pendant sync owns this database; no compose attempted")?;
253    let wal = sidecar(&db, "-wal")?;
254    let mut receipt = Receipt {
255        version: if options.monorepo { 2 } else { 1 },
256        monorepo: options.monorepo,
257        selected: if options.monorepo {
258            inputs.iter().map(|(key, _)| key.clone()).collect()
259        } else {
260            Vec::new()
261        },
262        repo: repo.clone(),
263        db: db.clone(),
264        hashes,
265        authority: authority_hash,
266        database: stamp(&db)?,
267        wal: stamp(&wal)?,
268    };
269    if receipt_path.exists() {
270        let prior: Receipt = serde_json::from_slice(&fs::read(&receipt_path)?)
271            .context("invalid pendant receipt; no compose attempted")?;
272        if !matches!(prior.version, 1 | 2) {
273            bail!("unsupported pendant receipt version; no compose attempted");
274        }
275        if receipt.database.is_some() && prior == receipt {
276            return Ok(Report {
277                outcome: "unchanged",
278                packs: loaded_packs,
279                current: None,
280            });
281        }
282    }
283    let program = std::env::var_os("QUIPU_BIN").unwrap_or_else(|| "quipu".into());
284    if options.monorepo {
285        // Even excluded parents and absent projects must pass native integrity,
286        // vocabulary and authority gates. Verification touches only a throwaway store.
287        let mut verify = Command::new(&program);
288        verify
289            .arg("compose")
290            .arg("--db")
291            .arg(frozen.path().join("verification.db"))
292            .arg("--shapes-from")
293            .arg(&authority_copy)
294            .arg(&authority_copy);
295        for (_, path) in &all_inputs {
296            verify.arg(path);
297        }
298        let checked = verify.output().context("verify all monorepo shares")?;
299        let report: serde_json::Value = serde_json::from_slice(&checked.stdout).unwrap_or_default();
300        if !checked.status.success() || report["outcome"] != "composed" {
301            bail!(
302                "monorepo share verification refused; CURRENT and receipt unchanged: {}",
303                String::from_utf8_lossy(&checked.stderr)
304            );
305        }
306    }
307    let mut command = Command::new(program);
308    command
309        .args(["compose", "--repo", &repo])
310        .arg("--db")
311        .arg(&db)
312        .args(["--shapes-from", AUTHORITY_KEY]);
313    command.arg(format!("{AUTHORITY_KEY}={}", authority_copy.display()));
314    for (key, path) in inputs {
315        command.arg(format!("{key}={}", path.display()));
316    }
317    let output = command
318        .output()
319        .context("run Quipu repository composition")?;
320    if !output.status.success() {
321        bail!(
322            "pendant composition refused; receipt unchanged: {}",
323            String::from_utf8_lossy(&output.stderr)
324        );
325    }
326    let result: serde_json::Value = serde_json::from_slice(&output.stdout)
327        .context("Quipu returned no verifiable repository composition receipt")?;
328    let expected = format!("urn:quipu:repo:{repo}:current");
329    if result["repo"] != repo
330        || result["composition"]["outcome"] != "composed"
331        || result["current"]["dataset"] != expected
332    {
333        bail!(
334            "Quipu did not attest the expected repository CURRENT composition; receipt unchanged"
335        );
336    }
337    receipt.database = stamp(&db)?;
338    receipt.wal = stamp(&wal)?;
339    if receipt.database.is_none() {
340        bail!("Quipu reported composition but the database is absent; receipt unchanged");
341    }
342    let mut file = tempfile::NamedTempFile::new_in(parent)?;
343    serde_json::to_writer_pretty(&mut file, &receipt)?;
344    file.as_file().sync_all()?;
345    file.persist(&receipt_path).context(
346        "composition succeeded but saving its receipt failed; inspect CURRENT before retrying",
347    )?;
348    let saved: Receipt = serde_json::from_slice(&fs::read(&receipt_path)?)?;
349    if saved != receipt {
350        bail!("pendant receipt read-back differs");
351    }
352    Ok(Report {
353        outcome: "composed",
354        packs: loaded_packs,
355        current: Some(result["current"].clone()),
356    })
357}
caboodle/pendant_sync.rs

caboodle/
pendant_sync.rs
