dvs/backends/
local.rs

1use std::collections::HashSet;
2use std::io::{BufReader, Write};
3use std::path::{Path, PathBuf};
4use std::sync::Mutex;
5
6#[cfg(unix)]
7use anyhow::anyhow;
8use anyhow::{Result, bail};
9use fs_err as fs;
10use serde::{Deserialize, Serialize};
11use uuid::Uuid;
12
13use crate::Hashes;
14use crate::audit::{AuditEntry, parse_audit_log};
15use crate::backends::Backend;
16use crate::config::Compression;
17
18const AUDIT_LOG_FILENAME: &str = "audit.log.jsonl";
19/// Only protects the current dvs process, not concurrent dvs processes
20static AUDIT_LOG_LOCK: Mutex<()> = Mutex::new(());
21
22/// Resolve a group name to its GID.
23#[cfg(unix)]
24fn resolve_group(group_name: &str) -> Result<nix::unistd::Gid> {
25    use nix::unistd::Group;
26    let group =
27        Group::from_name(group_name)?.ok_or_else(|| anyhow!("Group '{}' not found", group_name))?;
28    Ok(nix::unistd::Gid::from_raw(group.gid.as_raw()))
29}
30
31#[cfg(not(unix))]
32fn resolve_group(_: &str) -> Result<()> {
33    Ok(())
34}
35
36/// Detect the current user's primary group name.
37#[cfg(unix)]
38fn detect_primary_group() -> Result<String> {
39    use nix::unistd::Group;
40    let gid = nix::unistd::getegid();
41    let group = Group::from_gid(gid)?
42        .ok_or_else(|| anyhow!("Could not resolve primary group for GID {}", gid))?;
43    Ok(group.name)
44}
45
46fn make_readonly(path: impl AsRef<Path>) -> Result<()> {
47    let mut perms = fs::metadata(path.as_ref())?.permissions();
48    perms.set_readonly(true);
49    fs::set_permissions(path.as_ref(), perms)?;
50    Ok(())
51}
52
53const SHARED_DIRECTORY_MODE: u32 = 0o2770;
54const SHARED_BLOB_MODE: u32 = 0o0440;
55const SHARED_AUDIT_LOG_MODE: u32 = 0o0660;
56
57const OPEN_DIRECTORY_MODE: u32 = 0o2777;
58const OPEN_AUDIT_LOG_MODE: u32 = 0o0666;
59
60#[cfg(unix)]
61fn ensure_mode(path: impl AsRef<Path>, mode: u32) -> Result<()> {
62    use std::os::unix::fs::PermissionsExt;
63
64    let path = path.as_ref();
65    let mut perms = fs::metadata(path)?.permissions();
66    if perms.mode() & 0o7777 != mode {
67        perms.set_mode(mode);
68        fs::set_permissions(path, perms)?;
69    }
70    Ok(())
71}
72
73#[cfg(not(unix))]
74fn ensure_mode(_path: impl AsRef<Path>, _mode: u32) -> Result<()> {
75    Ok(())
76}
77
78#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)]
79pub struct LocalBackend {
80    pub path: PathBuf,
81    group: Option<String>,
82    #[serde(default, skip_serializing_if = "core::ops::Not::not")]
83    open: bool,
84}
85
86impl LocalBackend {
87    pub fn new(path: impl AsRef<Path>, group: Option<String>) -> Result<Self> {
88        let group = match group {
89            Some(g) => {
90                resolve_group(&g)?;
91                Some(g)
92            }
93            None => {
94                #[cfg(unix)]
95                {
96                    match detect_primary_group() {
97                        Ok(g) => {
98                            log::debug!("Auto-detected primary group: {}", g);
99                            Some(g)
100                        }
101                        Err(e) => {
102                            log::warn!("Could not detect primary group: {e}");
103                            None
104                        }
105                    }
106                }
107                #[cfg(not(unix))]
108                {
109                    None
110                }
111            }
112        };
113
114        Ok(Self {
115            path: path.as_ref().to_path_buf(),
116            group,
117            open: false,
118        })
119    }
120
121    /// Apply configured group ownership to a path.
122    /// No-op on non-Unix or if no group is set.
123    #[cfg(unix)]
124    fn apply_group(&self, path: impl AsRef<Path>) -> Result<()> {
125        use std::os::unix::fs::MetadataExt;
126
127        use nix::unistd::chown;
128
129        if let Some(group_name) = &self.group {
130            let path = path.as_ref();
131            let gid = resolve_group(group_name)?;
132            if fs::metadata(path)?.gid() == gid.as_raw() {
133                return Ok(());
134            }
135
136            log::debug!("Setting group {} on {}", group_name, path.display());
137            chown(path, None, Some(gid))?;
138        }
139
140        Ok(())
141    }
142
143    #[cfg(not(unix))]
144    fn apply_group(&self, _path: impl AsRef<Path>) -> Result<()> {
145        Ok(())
146    }
147
148    fn ensure_group_and_mode(&self, path: impl AsRef<Path>, mode: u32) -> Result<()> {
149        if self.group.is_some() || self.open {
150            let path = path.as_ref();
151            self.apply_group(path)?;
152            ensure_mode(path, mode)?;
153        }
154        Ok(())
155    }
156
157    fn dir_mode(&self) -> u32 {
158        if self.open {
159            OPEN_DIRECTORY_MODE
160        } else {
161            SHARED_DIRECTORY_MODE
162        }
163    }
164
165    fn audit_mode(&self) -> u32 {
166        if self.open {
167            OPEN_AUDIT_LOG_MODE
168        } else {
169            SHARED_AUDIT_LOG_MODE
170        }
171    }
172
173    // On non-Unix, a configured group is intentionally a no-op. We still need
174    // blobs to become read-only eg on Windows, so this cannot be just `self.group.is_some()`.
175    fn use_shared_blob_mode(&self) -> bool {
176        #[cfg(unix)]
177        {
178            self.group.is_some()
179        }
180        #[cfg(not(unix))]
181        {
182            false
183        }
184    }
185
186    fn hash_to_path(&self, hashes: &Hashes) -> Result<PathBuf> {
187        let hash = hashes.get_blake3();
188        if hash.len() < 3 || !hash.chars().all(|c| c.is_ascii_hexdigit()) {
189            bail!("Invalid hash: {hash}");
190        }
191        let (prefix, suffix) = hash.split_at(2);
192        Ok(self.path.join(prefix).join(suffix))
193    }
194}
195
196impl Backend for LocalBackend {
197    fn is_initialized(&self) -> Result<bool> {
198        Ok(self.path.join(AUDIT_LOG_FILENAME).exists())
199    }
200
201    fn local_path(&self) -> Option<&Path> {
202        Some(&self.path)
203    }
204
205    fn init(&self) -> Result<()> {
206        log::debug!("Creating storage directory: {}", self.path.display());
207        fs::create_dir_all(&self.path)?;
208        self.ensure_group_and_mode(&self.path, self.dir_mode())?;
209        log::info!("Initialized local storage at {}", self.path.display());
210        Ok(())
211    }
212
213    fn store(
214        &self,
215        hash: &Hashes,
216        source: &Path,
217        compression: Compression,
218        on_bytes: Option<&(dyn Fn(u64) + Send + Sync)>,
219    ) -> Result<u64> {
220        let path = self.hash_to_path(hash)?;
221        if let Some(parent) = path.parent() {
222            fs::create_dir_all(parent)?;
223            self.ensure_group_and_mode(&self.path, self.dir_mode())?;
224            self.ensure_group_and_mode(parent, self.dir_mode())?;
225        }
226
227        // Unique tmp name
228        let tmp_path = path.with_extension(format!("tmp.{}", Uuid::new_v4()));
229
230        // Always clean up the tmp file even if an error occured
231        let result = (|| {
232            let stored_size = compression.compress(source, &tmp_path, on_bytes)?;
233            if self.use_shared_blob_mode() {
234                self.ensure_group_and_mode(&tmp_path, SHARED_BLOB_MODE)?;
235            } else {
236                make_readonly(&tmp_path)?;
237            }
238            fs::rename(&tmp_path, &path)?;
239            Ok(stored_size)
240        })();
241        if result.is_err() {
242            let _ = fs::remove_file(&tmp_path);
243        }
244        result
245    }
246
247    fn retrieve(
248        &self,
249        hash: &Hashes,
250        target: &Path,
251        compression: Compression,
252        on_bytes: Option<&(dyn Fn(u64) + Send + Sync)>,
253    ) -> Result<bool> {
254        let path = self.hash_to_path(hash)?;
255        if path.is_file() {
256            if let Some(parent) = target.parent() {
257                fs::create_dir_all(parent)?;
258            }
259            compression.decompress(&path, target, on_bytes)?;
260            Ok(true)
261        } else {
262            Ok(false)
263        }
264    }
265
266    fn exists(&self, hash: &Hashes) -> Result<bool> {
267        Ok(self.hash_to_path(hash)?.is_file())
268    }
269
270    fn remove(&self, hash: &Hashes) -> Result<()> {
271        let path = self.hash_to_path(hash)?;
272        if path.is_file() {
273            log::debug!("Removing {path:?} from storage");
274            fs::remove_file(path)?;
275        }
276        Ok(())
277    }
278
279    fn log_audit(&self, entry: &AuditEntry) -> Result<()> {
280        let _guard = AUDIT_LOG_LOCK
281            .lock()
282            .expect("audit log lock should not be poisoned");
283        log::debug!("Appending {entry:?} to audit log");
284
285        fs::create_dir_all(&self.path)?;
286        self.ensure_group_and_mode(&self.path, self.dir_mode())?;
287        let audit_path = self.path.join(AUDIT_LOG_FILENAME);
288        let mut file = fs::OpenOptions::new()
289            .create(true)
290            .append(true)
291            .open(&audit_path)?;
292        self.ensure_group_and_mode(&audit_path, self.audit_mode())?;
293        let json = serde_json::to_string(entry)?;
294        writeln!(file, "{}", json)?;
295        Ok(())
296    }
297
298    /// An empty `files` slice returns the full audit log. See
299    /// [`parse_audit_log`] for the filtering rules.
300    fn read_audit_file(&self, files: &[PathBuf]) -> Result<Vec<AuditEntry>> {
301        let files_to_include: HashSet<_> = HashSet::from_iter(files.iter().cloned());
302        let audit_path = self.path.join(AUDIT_LOG_FILENAME);
303        let f = fs::File::open(&audit_path)?;
304        let entries = parse_audit_log(BufReader::new(f), &files_to_include)?;
305        Ok(entries)
306    }
307}
308
309#[cfg(test)]
310mod tests {
311    use super::*;
312    use crate::audit::{Action, AuditEntry, AuditFile, parse_audit_log};
313    use crate::config::Compression;
314    use crate::hashes::Hashes;
315    use std::io::Cursor;
316
317    fn test_hash(hash: &str) -> Hashes {
318        Hashes {
319            blake3: hash.to_string(),
320            md5: None,
321        }
322    }
323
324    #[test]
325    fn hash_to_path_rejects_bad_hash() {
326        let backend = LocalBackend::new("/tmp/storage", None).unwrap();
327
328        // These should error or be sanitized
329        assert!(
330            backend
331                .hash_to_path(&test_hash("../../etc/passwd"))
332                .is_err()
333        );
334        assert!(backend.hash_to_path(&test_hash("../escape")).is_err());
335        assert!(
336            backend
337                .hash_to_path(&test_hash("d41d8cd98f00b204e9800998ecf8427e"))
338                .is_ok()
339        );
340    }
341
342    #[test]
343    fn init_creates_storage_directory() {
344        let tmp = tempfile::tempdir().unwrap();
345        let storage_path = tmp.path().join("storage");
346
347        let backend = LocalBackend::new(&storage_path, None).unwrap();
348        assert!(!storage_path.exists());
349
350        backend.init().unwrap();
351        assert!(storage_path.is_dir());
352    }
353
354    #[test]
355    fn store_creates_hash_prefixed_path() {
356        let tmp = tempfile::tempdir().unwrap();
357        let storage = tmp.path().join("storage");
358        let backend = LocalBackend::new(&storage, None).unwrap();
359        backend.init().unwrap();
360
361        // Create source file
362        let source = tmp.path().join("source.txt");
363        fs::write(&source, b"test content").unwrap();
364
365        let hash = test_hash("d41d8cd98f00b204e9800998ecf8427e");
366        backend
367            .store(&hash, &source, Compression::None, None)
368            .unwrap();
369
370        let stored = storage.join("d4").join("1d8cd98f00b204e9800998ecf8427e");
371        assert!(stored.is_file());
372        assert_eq!(fs::read(&stored).unwrap(), b"test content");
373    }
374
375    #[test]
376    fn retrieve_copies_to_target() {
377        let tmp = tempfile::tempdir().unwrap();
378        let storage = tmp.path().join("storage");
379        let backend = LocalBackend::new(&storage, None).unwrap();
380        backend.init().unwrap();
381
382        // Store content via store()
383        let hash = test_hash("abc123def456789012345678901234ab");
384        let source = tmp.path().join("source.txt");
385        fs::write(&source, b"stored content").unwrap();
386        backend
387            .store(&hash, &source, Compression::None, None)
388            .unwrap();
389
390        // Retrieve to new location
391        let target = tmp.path().join("retrieved.txt");
392        let result = backend
393            .retrieve(&hash, &target, Compression::None, None)
394            .unwrap();
395
396        // file was copied if result == true
397        assert!(result);
398        assert_eq!(fs::read(&target).unwrap(), b"stored content");
399    }
400
401    #[test]
402    fn retrieve_returns_false_when_missing() {
403        let tmp = tempfile::tempdir().unwrap();
404        let storage = tmp.path().join("storage");
405        let backend = LocalBackend::new(&storage, None).unwrap();
406        backend.init().unwrap();
407
408        let target = tmp.path().join("target.txt");
409        let result = backend
410            .retrieve(
411                &test_hash("1234567890123456789012"),
412                &target,
413                Compression::None,
414                None,
415            )
416            .unwrap();
417
418        assert!(!result);
419        assert!(!target.exists());
420    }
421
422    #[test]
423    fn exists_returns_true_for_stored() {
424        let tmp = tempfile::tempdir().unwrap();
425        let storage = tmp.path().join("storage");
426        let backend = LocalBackend::new(&storage, None).unwrap();
427        backend.init().unwrap();
428
429        let hash = test_hash("abc123def456789012345678901234ab");
430        assert!(!backend.exists(&hash).unwrap());
431        let source = tmp.path().join("source.txt");
432        fs::write(&source, b"content").unwrap();
433        backend
434            .store(&hash, &source, Compression::None, None)
435            .unwrap();
436        assert!(backend.exists(&hash).unwrap());
437    }
438
439    #[test]
440    fn remove_deletes_stored_file() {
441        let tmp = tempfile::tempdir().unwrap();
442        let storage = tmp.path().join("storage");
443        let backend = LocalBackend::new(&storage, None).unwrap();
444        backend.init().unwrap();
445
446        let hash = test_hash("abc123def456789012345678901234ab");
447        let source = tmp.path().join("source.txt");
448        fs::write(&source, b"content").unwrap();
449        backend
450            .store(&hash, &source, Compression::None, None)
451            .unwrap();
452        assert!(backend.exists(&hash).unwrap());
453
454        backend.remove(&hash).unwrap();
455        assert!(!backend.exists(&hash).unwrap());
456        // removing something that doesn't exist is a noop
457        backend.remove(&hash).unwrap();
458    }
459
460    #[test]
461    fn stored_files_are_readonly() {
462        let tmp = tempfile::tempdir().unwrap();
463        let storage = tmp.path().join("storage");
464        let backend = LocalBackend::new(&storage, None).unwrap();
465        backend.init().unwrap();
466
467        let hash = test_hash("abc123def456789012345678901234ab");
468        let source = tmp.path().join("source.txt");
469        fs::write(&source, b"content").unwrap();
470        backend
471            .store(&hash, &source, Compression::None, None)
472            .unwrap();
473
474        let stored = storage.join("ab").join("c123def456789012345678901234ab");
475        let perms = fs::metadata(&stored).unwrap().permissions();
476        assert!(perms.readonly());
477    }
478
479    #[test]
480    fn log_audit_appends_to_jsonl() {
481        let tmp = tempfile::tempdir().unwrap();
482        let storage = tmp.path().join("storage");
483        let backend = LocalBackend::new(&storage, None).unwrap();
484        backend.init().unwrap();
485
486        let hash = test_hash("abc123def456789012345678901234ab");
487
488        let entry1 = AuditEntry {
489            operation_id: "op-1".to_string(),
490            timestamp: jiff::Timestamp::from_second(1000000000).unwrap(),
491            user: "alice".to_string(),
492            action: Action::Add {
493                file: AuditFile {
494                    path: PathBuf::from("file1.txt"),
495                    hashes: hash.clone(),
496                },
497                compression: Compression::Zstd,
498            },
499        };
500
501        let entry2 = AuditEntry {
502            operation_id: "op-2".to_string(),
503            timestamp: jiff::Timestamp::from_second(2000000000).unwrap(),
504            user: "bob".to_string(),
505            action: Action::Add {
506                file: AuditFile {
507                    path: PathBuf::from("file2.txt"),
508                    hashes: hash.clone(),
509                },
510                compression: Compression::Zstd,
511            },
512        };
513
514        backend.log_audit(&entry1).unwrap();
515        backend.log_audit(&entry2).unwrap();
516
517        let audit_path = storage.join("audit.log.jsonl");
518        assert!(audit_path.is_file());
519
520        let content = fs::read(&audit_path).unwrap();
521        let entries = parse_audit_log(Cursor::new(content), &HashSet::new()).unwrap();
522        assert_eq!(entries.len(), 2);
523
524        assert_eq!(entries[0].operation_id, "op-1");
525        // i64-seconds wire format still deserializes into a typed Timestamp
526        assert_eq!(entries[0].timestamp.as_second(), 1000000000);
527        assert_eq!(entries[0].user, "alice");
528
529        assert_eq!(entries[1].operation_id, "op-2");
530        assert_eq!(entries[1].timestamp.as_second(), 2000000000);
531        assert_eq!(entries[1].user, "bob");
532    }
533
534    #[test]
535    fn log_audit_is_valid_jsonl_under_concurrency() {
536        let tmp = tempfile::tempdir().unwrap();
537        let storage = tmp.path().join("storage");
538        let backend = LocalBackend::new(&storage, None).unwrap();
539        backend.init().unwrap();
540
541        let hash = test_hash("abc123def456789012345678901234ab");
542        let workers = 4;
543        let entries_per_worker = 64;
544
545        std::thread::scope(|scope| {
546            let backend = &backend;
547            for worker in 0..workers {
548                let hash = hash.clone();
549                scope.spawn(move || {
550                    for idx in 0..entries_per_worker {
551                        let entry = AuditEntry {
552                            operation_id: format!("op-{worker}-{idx}"),
553                            timestamp: jiff::Timestamp::from_second(
554                                (worker * entries_per_worker + idx) as i64,
555                            )
556                            .unwrap(),
557                            user: format!("user-{worker}"),
558                            action: Action::Add {
559                                file: AuditFile {
560                                    path: PathBuf::from(format!("file-{worker}-{idx}.txt")),
561                                    hashes: hash.clone(),
562                                },
563                                compression: Compression::Zstd,
564                            },
565                        };
566                        backend.log_audit(&entry).unwrap();
567                    }
568                });
569            }
570        });
571
572        let audit_path = storage.join("audit.log.jsonl");
573        let content = fs::read(&audit_path).unwrap();
574        let entries = parse_audit_log(Cursor::new(content), &HashSet::new()).unwrap();
575        assert_eq!(entries.len(), workers * entries_per_worker);
576    }
577
578    #[cfg(unix)]
579    fn test_audit_entry() -> AuditEntry {
580        AuditEntry {
581            operation_id: "op-1".to_string(),
582            timestamp: jiff::Timestamp::from_second(1000000000).unwrap(),
583            user: "alice".to_string(),
584            action: Action::Add {
585                file: AuditFile {
586                    path: PathBuf::from("file1.txt"),
587                    hashes: test_hash("abc123def456789012345678901234ab"),
588                },
589                compression: Compression::Zstd,
590            },
591        }
592    }
593
594    #[cfg(unix)]
595    #[test]
596    fn permissions_with_group_use_shared_modes() {
597        use nix::unistd::{Group, getegid};
598        use std::os::unix::fs::PermissionsExt;
599
600        let current_group_name = Group::from_gid(getegid()).unwrap().unwrap().name;
601
602        let tmp = tempfile::tempdir().unwrap();
603        let storage = tmp.path().join("storage");
604        let backend = LocalBackend::new(&storage, Some(current_group_name)).unwrap();
605
606        backend.init().unwrap();
607        let storage_mode = fs::metadata(&storage).unwrap().permissions().mode();
608        assert_eq!(storage_mode & 0o7777, SHARED_DIRECTORY_MODE);
609
610        let hash = test_hash("abc123def456789012345678901234ab");
611        let source = tmp.path().join("source.txt");
612        fs::write(&source, b"content").unwrap();
613        backend
614            .store(&hash, &source, Compression::None, None)
615            .unwrap();
616
617        let prefix_dir = storage.join("ab");
618        let prefix_mode = fs::metadata(&prefix_dir).unwrap().permissions().mode();
619        assert_eq!(prefix_mode & 0o7777, SHARED_DIRECTORY_MODE);
620
621        let stored = prefix_dir.join("c123def456789012345678901234ab");
622        let file_mode = fs::metadata(&stored).unwrap().permissions().mode();
623        assert_eq!(file_mode & 0o777, SHARED_BLOB_MODE);
624
625        backend.log_audit(&test_audit_entry()).unwrap();
626        let audit_path = storage.join("audit.log.jsonl");
627        let audit_mode = fs::metadata(&audit_path).unwrap().permissions().mode();
628        assert_eq!(audit_mode & 0o777, SHARED_AUDIT_LOG_MODE);
629    }
630}
dvs/backends/local.rs

dvs/backends/
local.rs
