codex_sync/sync/
core.rs

1//! Transport-independent snapshot creation, validation, and incremental import.
2
3use crate::config::Config;
4use anyhow::{Context, Result, bail};
5use serde::{Deserialize, Serialize};
6use std::{
7    collections::{HashMap, HashSet},
8    fs,
9    io::Write,
10    path::{Component, Path, PathBuf},
11    time::{SystemTime, UNIX_EPOCH},
12};
13use walkdir::WalkDir;
14
15pub const MANIFEST_FILE: &str = "manifest.json";
16
17#[derive(Debug, Clone, Serialize, Deserialize)]
18pub struct FileInfo {
19    pub path: String,
20    pub size: u64,
21    pub modified: u64,
22    pub hash: String,
23}
24
25#[derive(Debug, Clone, Serialize, Deserialize)]
26pub struct Manifest {
27    pub node_id: String,
28    pub node_name: String,
29    pub files: Vec<FileInfo>,
30}
31
32#[derive(Debug, Default, Serialize)]
33pub struct PullReport {
34    pub downloaded: usize,
35    pub skipped: usize,
36    pub conflicts: usize,
37}
38
39#[allow(async_fn_in_trait)]
40/// Trait implemented by HTTP, directory, or future object-storage sources.
41pub trait SnapshotSource {
42    async fn manifest(&self) -> Result<Manifest>;
43    async fn read(&self, file: &FileInfo) -> Result<Vec<u8>>;
44}
45
46pub struct DirectorySource {
47    files_root: PathBuf,
48    manifest: Manifest,
49}
50
51impl DirectorySource {
52    pub fn open(snapshot: &Path) -> Result<Self> {
53        let manifest = read_manifest(snapshot)?;
54        Ok(Self {
55            files_root: snapshot.join("files"),
56            manifest,
57        })
58    }
59}
60
61impl SnapshotSource for DirectorySource {
62    async fn manifest(&self) -> Result<Manifest> {
63        Ok(self.manifest.clone())
64    }
65
66    async fn read(&self, file: &FileInfo) -> Result<Vec<u8>> {
67        let path = safe_path(&self.files_root, &file.path)?;
68        fs::read(&path).with_context(|| format!("快照文件缺失：{}", path.display()))
69    }
70}
71
72pub fn manifest(cfg: &Config) -> Result<Manifest> {
73    if !cfg.sync_dir.exists() {
74        fs::create_dir_all(&cfg.sync_dir)?;
75    }
76    manifest_tree(
77        &cfg.sync_dir,
78        cfg.node_id(),
79        cfg.node_name.clone(),
80        cfg.max_file_size_bytes,
81        cfg.max_files,
82        |relative| !excluded(cfg, relative),
83    )
84}
85
86pub fn manifest_tree(
87    root: &Path,
88    node_id: String,
89    node_name: String,
90    max_file_size: u64,
91    max_files: usize,
92    include: impl Fn(&Path) -> bool,
93) -> Result<Manifest> {
94    let mut files = Vec::new();
95    for entry in WalkDir::new(root)
96        .follow_links(false)
97        .into_iter()
98        .filter_map(Result::ok)
99    {
100        if !entry.file_type().is_file() {
101            continue;
102        }
103        let relative = entry.path().strip_prefix(root)?;
104        if !include(relative) {
105            continue;
106        }
107        if files.len() >= max_files {
108            bail!("同步文件数量超过上限 {max_files}");
109        }
110        let metadata = entry.metadata()?;
111        if metadata.len() > max_file_size {
112            bail!(
113                "同步文件超过大小上限 {}：{}",
114                max_file_size,
115                entry.path().display()
116            );
117        }
118        let bytes = fs::read(entry.path())?;
119        files.push(FileInfo {
120            path: relative.to_string_lossy().replace('\\', "/"),
121            size: metadata.len(),
122            modified: metadata
123                .modified()?
124                .duration_since(UNIX_EPOCH)
125                .unwrap_or_default()
126                .as_secs(),
127            hash: blake3::hash(&bytes).to_hex().to_string(),
128        });
129    }
130    files.sort_by(|left, right| left.path.cmp(&right.path));
131    Ok(Manifest {
132        node_id,
133        node_name,
134        files,
135    })
136}
137
138/// Imports a source using one validation, incremental, and conflict policy.
139pub async fn import_from<S: SnapshotSource>(
140    cfg: &Config,
141    source: &S,
142    overwrite: bool,
143) -> Result<PullReport> {
144    let remote = source.manifest().await?;
145    validate_manifest(
146        &cfg.sync_dir,
147        &remote,
148        cfg.max_file_size_bytes,
149        cfg.max_files,
150    )?;
151    if remote.node_id == cfg.node_id() {
152        bail!("这是本机导出的快照，无需导入");
153    }
154    let local_map: HashMap<_, _> = manifest(cfg)?
155        .files
156        .into_iter()
157        .map(|file| (file.path.clone(), file))
158        .collect();
159    let mut report = PullReport::default();
160
161    for file in &remote.files {
162        if local_map
163            .get(&file.path)
164            .is_some_and(|local| local.hash == file.hash)
165        {
166            report.skipped += 1;
167            continue;
168        }
169        let bytes = source.read(file).await?;
170        verify_bytes(file, &bytes, cfg.max_file_size_bytes)?;
171        let target = safe_path(&cfg.sync_dir, &file.path)?;
172        let target = if target.exists() && !overwrite {
173            report.conflicts += 1;
174            conflict_path(&target, &remote.node_name)
175        } else {
176            report.downloaded += 1;
177            target
178        };
179        atomic_write(&target, &bytes)?;
180    }
181    Ok(report)
182}
183
184pub fn publish_snapshot(cfg: &Config, snapshot: &Path) -> Result<Manifest> {
185    let manifest = manifest(cfg)?;
186    let files_root = snapshot.join("files");
187    fs::create_dir_all(&files_root)?;
188    let published_manifest = snapshot.join(MANIFEST_FILE);
189    if published_manifest.exists() {
190        fs::remove_file(&published_manifest)?;
191    }
192    for file in &manifest.files {
193        let source = safe_path(&cfg.sync_dir, &file.path)?;
194        let target = safe_path(&files_root, &file.path)?;
195        atomic_copy(&source, &target)?;
196    }
197    atomic_write(&published_manifest, &serde_json::to_vec_pretty(&manifest)?)?;
198    Ok(manifest)
199}
200
201pub fn read_manifest(snapshot: &Path) -> Result<Manifest> {
202    Ok(serde_json::from_slice(&fs::read(
203        snapshot.join(MANIFEST_FILE),
204    )?)?)
205}
206
207pub fn validate_manifest(
208    target_root: &Path,
209    manifest: &Manifest,
210    max_file_size: u64,
211    max_files: usize,
212) -> Result<()> {
213    if manifest.files.len() > max_files {
214        bail!("远端同步文件数量超过上限 {max_files}");
215    }
216    let mut paths = HashSet::new();
217    for file in &manifest.files {
218        safe_path(target_root, &file.path)?;
219        if file.size > max_file_size {
220            bail!("远端文件超过大小上限：{}", file.path);
221        }
222        if file.hash.len() != 64 || !file.hash.bytes().all(|byte| byte.is_ascii_hexdigit()) {
223            bail!("远端文件哈希格式无效：{}", file.path);
224        }
225        if !paths.insert(&file.path) {
226            bail!("远端清单包含重复路径：{}", file.path);
227        }
228    }
229    Ok(())
230}
231
232pub fn verify_snapshot(snapshot: &Path, manifest: &Manifest, max_file_size: u64) -> Result<()> {
233    let files_root = snapshot.join("files");
234    for file in &manifest.files {
235        let bytes = fs::read(safe_path(&files_root, &file.path)?)?;
236        verify_bytes(file, &bytes, max_file_size)?;
237    }
238    Ok(())
239}
240
241pub fn verify_bytes(file: &FileInfo, bytes: &[u8], max_file_size: u64) -> Result<()> {
242    if bytes.len() as u64 > max_file_size
243        || bytes.len() as u64 != file.size
244        || blake3::hash(bytes).to_hex().as_str() != file.hash
245    {
246        bail!("同步文件校验失败：{}", file.path);
247    }
248    Ok(())
249}
250
251pub fn read_local_file(cfg: &Config, relative: &str) -> Result<Vec<u8>> {
252    let file = safe_path(&cfg.sync_dir, relative)?;
253    let metadata = fs::symlink_metadata(&file)?;
254    if !metadata.file_type().is_file() || metadata.file_type().is_symlink() {
255        bail!("拒绝非普通文件");
256    }
257    if metadata.len() > cfg.max_file_size_bytes {
258        bail!("文件超过大小上限");
259    }
260    Ok(fs::read(file)?)
261}
262
263pub fn conflict_path(target: &Path, node_name: &str) -> PathBuf {
264    target.with_extension(format!(
265        "{}.conflict-{}",
266        target
267            .extension()
268            .and_then(|value| value.to_str())
269            .unwrap_or("file"),
270        sanitize_name(node_name)
271    ))
272}
273
274pub fn safe_path(root: &Path, relative: &str) -> Result<PathBuf> {
275    let path = Path::new(relative);
276    if path.is_absolute()
277        || path
278            .components()
279            .any(|component| !matches!(component, Component::Normal(_)))
280    {
281        bail!("非法路径");
282    }
283    Ok(root.join(path))
284}
285
286pub fn atomic_copy(source: &Path, target: &Path) -> Result<()> {
287    let bytes = fs::read(source)?;
288    atomic_write(target, &bytes)
289}
290
291pub fn atomic_write(target: &Path, bytes: &[u8]) -> Result<()> {
292    if let Some(parent) = target.parent() {
293        fs::create_dir_all(parent)?;
294    }
295    if target
296        .symlink_metadata()
297        .is_ok_and(|metadata| metadata.file_type().is_symlink())
298    {
299        bail!("拒绝覆盖符号链接：{}", target.display());
300    }
301    let temporary = temporary_path(target);
302    let mut output = fs::OpenOptions::new()
303        .create_new(true)
304        .write(true)
305        .open(&temporary)
306        .with_context(|| format!("写入 {}", temporary.display()))?;
307    output.write_all(bytes)?;
308    output.sync_all()?;
309    fs::rename(temporary, target)?;
310    Ok(())
311}
312
313fn excluded(cfg: &Config, relative: &Path) -> bool {
314    relative.components().any(|component| {
315        cfg.exclude
316            .iter()
317            .any(|item| component.as_os_str() == item.as_str())
318    })
319}
320
321fn sanitize_name(name: &str) -> String {
322    name.chars()
323        .map(|character| {
324            if character.is_ascii_alphanumeric() || matches!(character, '-' | '_') {
325                character
326            } else {
327                '_'
328            }
329        })
330        .collect()
331}
332
333fn temporary_path(target: &Path) -> PathBuf {
334    let nonce = SystemTime::now()
335        .duration_since(UNIX_EPOCH)
336        .unwrap_or_default()
337        .as_nanos();
338    target.with_extension(format!("codex-sync-{nonce}-{}-tmp", std::process::id()))
339}
340
341#[cfg(test)]
342mod tests {
343    use super::*;
344
345    struct MemorySource {
346        manifest: Manifest,
347        bytes: Vec<u8>,
348    }
349
350    impl SnapshotSource for MemorySource {
351        async fn manifest(&self) -> Result<Manifest> {
352            Ok(self.manifest.clone())
353        }
354
355        async fn read(&self, _file: &FileInfo) -> Result<Vec<u8>> {
356            Ok(self.bytes.clone())
357        }
358    }
359
360    fn temp_dir() -> PathBuf {
361        std::env::temp_dir().join(format!(
362            "codex-sync-core-test-{}-{}",
363            std::process::id(),
364            SystemTime::now()
365                .duration_since(UNIX_EPOCH)
366                .unwrap_or_default()
367                .as_nanos()
368        ))
369    }
370
371    #[test]
372    fn safe_path_rejects_traversal_and_absolute_paths() {
373        let root = Path::new("/tmp/root");
374        assert!(safe_path(root, "sessions/chat.jsonl").is_ok());
375        assert!(safe_path(root, "../auth.json").is_err());
376        assert!(safe_path(root, "/etc/passwd").is_err());
377        assert!(safe_path(root, "sessions/./chat.jsonl").is_ok());
378    }
379
380    #[test]
381    fn manifest_validation_rejects_untrusted_entries() {
382        let file = FileInfo {
383            path: "sessions/a.jsonl".into(),
384            size: 1,
385            modified: 0,
386            hash: "0".repeat(64),
387        };
388        let valid = Manifest {
389            node_id: "node".into(),
390            node_name: "node".into(),
391            files: vec![file.clone()],
392        };
393        assert!(validate_manifest(Path::new("/tmp/root"), &valid, 10, 1).is_ok());
394
395        let mut traversal = valid.clone();
396        traversal.files[0].path = "../auth.json".into();
397        assert!(validate_manifest(Path::new("/tmp/root"), &traversal, 10, 1).is_err());
398
399        let mut duplicate = valid.clone();
400        duplicate.files.push(file);
401        assert!(validate_manifest(Path::new("/tmp/root"), &duplicate, 10, 2).is_err());
402
403        let mut oversized = valid;
404        oversized.files[0].size = 11;
405        assert!(validate_manifest(Path::new("/tmp/root"), &oversized, 10, 1).is_err());
406    }
407
408    #[tokio::test]
409    async fn any_snapshot_source_gets_the_same_conflict_policy() -> Result<()> {
410        let root = temp_dir();
411        fs::create_dir_all(root.join("sessions"))?;
412        fs::write(root.join("sessions/chat.jsonl"), b"local")?;
413        let cfg = Config {
414            node_name: "local".into(),
415            shared_key: "shared-key".into(),
416            sync_dir: root.clone(),
417            ..Config::default()
418        };
419        let bytes = b"remote".to_vec();
420        let source = MemorySource {
421            manifest: Manifest {
422                node_id: "remote-node".into(),
423                node_name: "memory-adapter".into(),
424                files: vec![FileInfo {
425                    path: "sessions/chat.jsonl".into(),
426                    size: bytes.len() as u64,
427                    modified: 0,
428                    hash: blake3::hash(&bytes).to_hex().to_string(),
429                }],
430            },
431            bytes,
432        };
433
434        let report = import_from(&cfg, &source, false).await?;
435        assert_eq!(report.conflicts, 1);
436        assert_eq!(fs::read(root.join("sessions/chat.jsonl"))?, b"local");
437        assert_eq!(
438            fs::read(root.join("sessions/chat.jsonl.conflict-memory-adapter"))?,
439            b"remote"
440        );
441        fs::remove_dir_all(root)?;
442        Ok(())
443    }
444}
codex_sync/sync/core.rs

codex_sync/sync/
core.rs
