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

dvs/backends/
local.rs
