fuchsia_repo/repository/
file_system.rs

1// Copyright 2021 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::range::{ContentRange, Range};
6use crate::repository::{Error, RepoProvider, RepoStorage, Resource};
7use crate::util::file_stream;
8use anyhow::{Context as _, Result, anyhow};
9use camino::{Utf8Component, Utf8Path, Utf8PathBuf};
10use delivery_blob::DeliveryBlobType;
11use fuchsia_async as fasync;
12use fuchsia_merkle::Hash;
13use futures::future::BoxFuture;
14use futures::{AsyncRead, FutureExt as _};
15use log::warn;
16use std::collections::BTreeSet;
17use std::fs::{self, DirBuilder};
18use std::io::{Seek as _, SeekFrom};
19use std::os::unix::fs::MetadataExt;
20use std::time::SystemTime;
21use tempfile::{NamedTempFile, TempPath};
22use tuf::metadata::{MetadataPath, MetadataVersion, TargetPath};
23use tuf::pouf::Pouf1;
24use tuf::repository::{
25    FileSystemRepository as TufFileSystemRepository,
26    FileSystemRepositoryBuilder as TufFileSystemRepositoryBuilder,
27    RepositoryProvider as TufRepositoryProvider, RepositoryStorage as TufRepositoryStorage,
28};
29
30#[cfg(not(target_os = "fuchsia"))]
31use {
32    crate::repository::RepositorySpec,
33    futures::{Stream, StreamExt as _, stream::BoxStream},
34    notify::{RecursiveMode, Watcher as _, recommended_watcher},
35    std::{
36        ffi::OsStr,
37        pin::Pin,
38        task::{Context, Poll},
39    },
40};
41
42/// Describes how package blobs should be copied into the repository.
43#[derive(Copy, Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
44pub enum CopyMode {
45    /// Copy package blobs into the repository. This will skip copying the blob if it already exists
46    /// in the repository.
47    ///
48    /// This will create a Copy-on-Write (reflink) on file systems that support it.
49    #[default]
50    Copy,
51
52    /// Copy package blobs into the repository. This will overwrite a blob if it already exists in
53    /// the repository.
54    ///
55    /// This will create a Copy-on-Write (reflink) on file systems that support it.
56    CopyOverwrite,
57
58    /// Create hard links from the package blobs into the repository.
59    HardLink,
60}
61
62/// A builder to create a repository contained on the local file system.
63pub struct FileSystemRepositoryBuilder {
64    metadata_repo_path: Utf8PathBuf,
65    blob_repo_path: Utf8PathBuf,
66    copy_mode: CopyMode,
67    aliases: BTreeSet<String>,
68    delivery_blob_type: DeliveryBlobType,
69}
70
71impl FileSystemRepositoryBuilder {
72    /// Creates a [FileSystemRepositoryBuilder] where the TUF metadata is stored in
73    /// `metadata_repo_path`, and the blobs are stored in `blob_repo_path`.
74    pub fn new(metadata_repo_path: Utf8PathBuf, blob_repo_path: Utf8PathBuf) -> Self {
75        FileSystemRepositoryBuilder {
76            metadata_repo_path,
77            blob_repo_path,
78            copy_mode: CopyMode::Copy,
79            aliases: BTreeSet::new(),
80            delivery_blob_type: DeliveryBlobType::Type1,
81        }
82    }
83
84    /// Select which [CopyMode] to use when copying files into the repository.
85    pub fn copy_mode(mut self, copy_mode: CopyMode) -> Self {
86        self.copy_mode = copy_mode;
87        self
88    }
89
90    /// alias this repository to this name when this repository is registered on a target.
91    pub fn alias(mut self, alias: String) -> Self {
92        self.aliases.insert(alias);
93        self
94    }
95
96    /// alias this repository to these names when this repository is registered on a target.
97    pub fn aliases(mut self, aliases: impl IntoIterator<Item = String>) -> Self {
98        for alias in aliases {
99            self = self.alias(alias);
100        }
101        self
102    }
103
104    /// Set the type of delivery blob to generate when copying blobs into the repository.
105    pub fn delivery_blob_type(mut self, delivery_blob_type: DeliveryBlobType) -> Self {
106        self.delivery_blob_type = delivery_blob_type;
107        self
108    }
109
110    /// Set the path to the blob repo.
111    pub fn blob_repo_path(mut self, blob_repo_path: Utf8PathBuf) -> Self {
112        self.blob_repo_path = blob_repo_path;
113        self
114    }
115
116    /// Build a [FileSystemRepository].
117    pub fn build(self) -> FileSystemRepository {
118        FileSystemRepository {
119            metadata_repo_path: self.metadata_repo_path.clone(),
120            blob_repo_path: self.blob_repo_path,
121            copy_mode: self.copy_mode,
122            aliases: self.aliases,
123            delivery_blob_type: self.delivery_blob_type,
124            tuf_repo: TufFileSystemRepositoryBuilder::new(self.metadata_repo_path)
125                .targets_prefix("targets")
126                .build(),
127        }
128    }
129}
130
131/// Serve a repository from the file system.
132#[derive(Debug)]
133pub struct FileSystemRepository {
134    metadata_repo_path: Utf8PathBuf,
135    blob_repo_path: Utf8PathBuf,
136    copy_mode: CopyMode,
137    aliases: BTreeSet<String>,
138    delivery_blob_type: DeliveryBlobType,
139    tuf_repo: TufFileSystemRepository<Pouf1>,
140}
141
142impl FileSystemRepository {
143    /// Construct a [FileSystemRepositoryBuilder].
144    pub fn builder(
145        metadata_repo_path: Utf8PathBuf,
146        blob_repo_path: Utf8PathBuf,
147    ) -> FileSystemRepositoryBuilder {
148        FileSystemRepositoryBuilder::new(metadata_repo_path, blob_repo_path)
149    }
150
151    /// Construct a [FileSystemRepository].
152    pub fn new(metadata_repo_path: Utf8PathBuf, blob_repo_path: Utf8PathBuf) -> Self {
153        Self::builder(metadata_repo_path, blob_repo_path).build()
154    }
155
156    pub fn blob_repo_path(&self) -> &Utf8PathBuf {
157        &self.blob_repo_path
158    }
159
160    fn fetch<'a>(
161        &'a self,
162        repo_path: &Utf8Path,
163        resource_path: &str,
164        range: Range,
165    ) -> BoxFuture<'a, Result<Resource, Error>> {
166        let file_path = sanitize_path(repo_path, resource_path);
167        async move {
168            let file_path = file_path?;
169            let mut file = std::fs::File::open(&file_path).map_err(|err| {
170                if err.kind() == std::io::ErrorKind::NotFound {
171                    Error::NotFound
172                } else {
173                    Error::Io(err)
174                }
175            })?;
176
177            let total_len = file.metadata().map_err(Error::Io)?.len();
178
179            let content_range = match range {
180                Range::Full => ContentRange::Full { complete_len: total_len },
181                Range::Inclusive { first_byte_pos, last_byte_pos } => {
182                    if first_byte_pos > last_byte_pos
183                        || first_byte_pos >= total_len
184                        || last_byte_pos >= total_len
185                    {
186                        return Err(Error::RangeNotSatisfiable);
187                    }
188
189                    file.seek(SeekFrom::Start(first_byte_pos)).map_err(Error::Io)?;
190
191                    ContentRange::Inclusive {
192                        first_byte_pos,
193                        last_byte_pos,
194                        complete_len: total_len,
195                    }
196                }
197                Range::From { first_byte_pos } => {
198                    if first_byte_pos >= total_len {
199                        return Err(Error::RangeNotSatisfiable);
200                    }
201
202                    file.seek(SeekFrom::Start(first_byte_pos)).map_err(Error::Io)?;
203
204                    ContentRange::Inclusive {
205                        first_byte_pos,
206                        last_byte_pos: total_len - 1,
207                        complete_len: total_len,
208                    }
209                }
210                Range::Suffix { len } => {
211                    if len > total_len {
212                        return Err(Error::RangeNotSatisfiable);
213                    }
214                    let start = total_len - len;
215                    file.seek(SeekFrom::Start(start)).map_err(Error::Io)?;
216
217                    ContentRange::Inclusive {
218                        first_byte_pos: start,
219                        last_byte_pos: total_len - 1,
220                        complete_len: total_len,
221                    }
222                }
223            };
224
225            let content_len = content_range.content_len();
226
227            Ok(Resource {
228                content_range,
229                stream: Box::pin(file_stream(content_len, file, Some(file_path))),
230            })
231        }
232        .boxed()
233    }
234}
235
236impl RepoProvider for FileSystemRepository {
237    #[cfg(not(target_os = "fuchsia"))]
238    fn spec(&self) -> RepositorySpec {
239        RepositorySpec::FileSystem {
240            metadata_repo_path: self.metadata_repo_path.clone(),
241            blob_repo_path: self.blob_repo_path.clone(),
242            aliases: self.aliases.clone(),
243        }
244    }
245
246    fn aliases(&self) -> &BTreeSet<String> {
247        &self.aliases
248    }
249
250    fn fetch_metadata_range<'a>(
251        &'a self,
252        resource_path: &str,
253        range: Range,
254    ) -> BoxFuture<'a, Result<Resource, Error>> {
255        self.fetch(&self.metadata_repo_path, resource_path, range)
256    }
257
258    fn fetch_blob_range<'a>(
259        &'a self,
260        resource_path: &str,
261        range: Range,
262    ) -> BoxFuture<'a, Result<Resource, Error>> {
263        self.fetch(&self.blob_repo_path, resource_path, range)
264    }
265
266    #[cfg(not(target_os = "fuchsia"))]
267    fn supports_watch(&self) -> bool {
268        true
269    }
270
271    #[cfg(not(target_os = "fuchsia"))]
272    fn watch(&self) -> Result<BoxStream<'static, ()>> {
273        // Since all we are doing is signaling that the timestamp file is changed, it's it's fine
274        // if the channel is full, since that just means we haven't consumed our notice yet.
275        let (mut sender, receiver) = futures::channel::mpsc::channel(1);
276
277        let mut watcher = recommended_watcher(move |event: notify::Result<notify::Event>| {
278            let event = match event {
279                Ok(event) => event,
280                Err(err) => {
281                    warn!("error receving notify event: {}", err);
282                    return;
283                }
284            };
285
286            if !matches!(
287                event.kind,
288                notify::EventKind::Create(_)
289                    | notify::EventKind::Modify(_)
290                    | notify::EventKind::Remove(_)
291            ) {
292                return;
293            }
294
295            // Send an event if any applied to timestamp.json.
296            let timestamp_name = OsStr::new("timestamp.json");
297            if event.paths.iter().any(|p| p.file_name() == Some(timestamp_name))
298                && let Err(e) = sender.try_send(())
299            {
300                if e.is_full() {
301                    // It's okay to ignore a full channel, since that just means that the other
302                    // side of the channel still has an outstanding notice, which should be the
303                    // same effect if we re-sent the event.
304                } else if !e.is_disconnected() {
305                    warn!("Error sending event: {:?}", e);
306                }
307            }
308        })?;
309
310        // Watch the repo path instead of directly watching timestamp.json to avoid
311        // https://github.com/notify-rs/notify/issues/165.
312        watcher.watch(self.metadata_repo_path.as_std_path(), RecursiveMode::NonRecursive)?;
313
314        Ok(WatchStream { _watcher: watcher, receiver }.boxed())
315    }
316
317    fn blob_modification_time<'a>(
318        &'a self,
319        path: &str,
320    ) -> BoxFuture<'a, Result<Option<SystemTime>>> {
321        let file_path = sanitize_path(&self.blob_repo_path, path);
322        async move {
323            let file_path = file_path?;
324            Ok(Some(fs::metadata(&file_path)?.modified()?))
325        }
326        .boxed()
327    }
328
329    fn blob_type(&self) -> DeliveryBlobType {
330        self.delivery_blob_type
331    }
332}
333
334impl TufRepositoryProvider<Pouf1> for FileSystemRepository {
335    fn fetch_metadata<'a>(
336        &'a self,
337        meta_path: &MetadataPath,
338        version: MetadataVersion,
339    ) -> BoxFuture<'a, tuf::Result<Box<dyn AsyncRead + Send + Unpin + 'a>>> {
340        self.tuf_repo.fetch_metadata(meta_path, version)
341    }
342
343    fn fetch_target<'a>(
344        &'a self,
345        target_path: &TargetPath,
346    ) -> BoxFuture<'a, tuf::Result<Box<dyn AsyncRead + Send + Unpin + 'a>>> {
347        self.tuf_repo.fetch_target(target_path)
348    }
349}
350
351impl TufRepositoryStorage<Pouf1> for FileSystemRepository {
352    fn store_metadata<'a>(
353        &'a self,
354        meta_path: &MetadataPath,
355        version: MetadataVersion,
356        metadata: &'a mut (dyn AsyncRead + Send + Unpin + 'a),
357    ) -> BoxFuture<'a, tuf::Result<()>> {
358        self.tuf_repo.store_metadata(meta_path, version, metadata)
359    }
360
361    fn store_target<'a>(
362        &'a self,
363        target_path: &TargetPath,
364        target: &'a mut (dyn AsyncRead + Send + Unpin + 'a),
365    ) -> BoxFuture<'a, tuf::Result<()>> {
366        self.tuf_repo.store_target(target_path, target)
367    }
368}
369
370impl RepoStorage for FileSystemRepository {
371    fn store_blob<'a>(
372        &'a self,
373        hash: &Hash,
374        len: u64,
375        src: &Utf8Path,
376    ) -> BoxFuture<'a, Result<()>> {
377        let src = src.to_path_buf();
378        let hash_str = hash.to_string();
379        let hash = *hash;
380
381        async move {
382            let src_metadata = fs::metadata(&src)?;
383            if src_metadata.len() != len {
384                return Err(anyhow!(BlobSizeMismatchError {
385                    hash,
386                    path: src.clone(),
387                    manifest_size: len,
388                    file_size: src_metadata.len(),
389                }));
390            }
391
392            let dst = sanitize_path(
393                &self.blob_repo_path,
394                &format!("{}/{hash_str}", u32::from(self.delivery_blob_type)),
395            )?;
396            let existing_len = match fs::File::open(&dst) {
397                Ok(file) => {
398                    if let Ok(len) = delivery_blob::decompressed_size_from_reader(file) {
399                        Some(len)
400                    } else {
401                        // In the event that the delivery blob is corrupt, log a warning and
402                        // return None to signify that it needs to be written.
403                        warn!("corrupt delivery blob found at {dst}, overwriting");
404                        None
405                    }
406                }
407                Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
408                Err(e) => return Err(anyhow!(e)),
409            };
410
411            if self.copy_mode == CopyMode::CopyOverwrite || existing_len != Some(len) {
412                generate_delivery_blob(&src, &dst, self.delivery_blob_type).await?;
413            }
414
415            Ok(())
416        }
417        .boxed()
418    }
419
420    fn store_delivery_blob<'a>(
421        &'a self,
422        hash: &Hash,
423        src: &Utf8Path,
424        delivery_blob_type: DeliveryBlobType,
425    ) -> BoxFuture<'a, Result<()>> {
426        let src = src.to_path_buf();
427        let hash = *hash;
428
429        async move {
430            if delivery_blob_type != self.delivery_blob_type {
431                warn!(
432                    "storing delivery blob type {:?} in repository with delivery blob type {:?}",
433                    delivery_blob_type, self.delivery_blob_type,
434                );
435                // TODO: convert the delivery blob to the expected type?
436            }
437            let dst = sanitize_path(
438                &self.blob_repo_path,
439                &format!("{}/{hash}", u32::from(delivery_blob_type)),
440            )?;
441
442            let src_metadata = fs::metadata(&src)?;
443            let dst_metadata = match fs::metadata(&dst) {
444                Ok(metadata) => Some(metadata),
445                Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
446                Err(e) => return Err(anyhow!(e)),
447            };
448
449            let dst_is_hardlink = if let Some(dst_metadata) = &dst_metadata {
450                dst_metadata.nlink() > 1
451            } else {
452                false
453            };
454
455            let dst_exists = dst_metadata.is_some();
456            let dst_dirty = !dst_exists;
457
458            match self.copy_mode {
459                CopyMode::Copy => {
460                    if dst_dirty || dst_is_hardlink {
461                        copy_blob(&src, &dst).await?
462                    }
463                }
464                CopyMode::CopyOverwrite => copy_blob(&src, &dst).await?,
465                CopyMode::HardLink => {
466                    let is_hardlink = if let Some(dst_metadata) = &dst_metadata {
467                        src_metadata.dev() == dst_metadata.dev()
468                            && src_metadata.ino() == dst_metadata.ino()
469                    } else {
470                        false
471                    };
472
473                    if is_hardlink {
474                        // No work to do if src and dest are already hardlinks.
475                    } else {
476                        // Create the parent directory if it doesn't yet exist.
477                        if let Some(parent) = dst.parent() {
478                            std::fs::create_dir_all(parent)?;
479                        }
480                        match fs::hard_link(&src, &dst) {
481                            Ok(()) => {
482                                // FIXME(b/271694204): Workaround an unknown issue where hardlinks
483                                // aren't readable immediately after creation in some environments.
484                                if fs::metadata(&dst).is_err() {
485                                    fuchsia_async::Timer::new(std::time::Duration::from_secs(1))
486                                        .await;
487                                    if fs::metadata(&dst).is_err() {
488                                        copy_blob(&src, &dst).await?
489                                    }
490                                }
491                            }
492                            Err(_) if dst_dirty => copy_blob(&src, &dst).await?,
493                            Err(_) => {
494                                // The dest file exists and has the right size,
495                                // but we failed to make it a hardlink.
496                            }
497                        }
498                    }
499                }
500            }
501            Ok(())
502        }
503        .boxed()
504    }
505}
506
507async fn create_temp_file(path: &Utf8Path) -> Result<TempPath> {
508    let temp_file = if let Some(parent) = path.parent() {
509        DirBuilder::new().recursive(true).create(parent)?;
510
511        NamedTempFile::new_in(parent)?
512    } else {
513        NamedTempFile::new_in(".")?
514    };
515
516    Ok(temp_file.into_temp_path())
517}
518
519// Set the blob at `path` to be read-only.
520async fn set_blob_read_only(path: &Utf8Path) -> Result<()> {
521    let file = fs::File::open(path)?;
522    let mut permissions = file.metadata()?.permissions();
523    permissions.set_readonly(true);
524    file.set_permissions(permissions)?;
525
526    Ok(())
527}
528
529// Performs a Copy-on-Write (reflink) of the file at `src_path` to `dst_path`.
530#[cfg(target_os = "linux")]
531async fn reflink(src_path: &Utf8Path, dst_path: &Utf8Path) -> Result<(), std::io::Error> {
532    use std::os::fd::AsRawFd;
533
534    let src = fs::File::open(src_path)?;
535    let dst = fs::File::create(dst_path)?;
536
537    // Safe because this is a synchronous syscall and the raw fds don't outlive the call.
538    let res = unsafe { libc::ioctl(dst.as_raw_fd(), libc::FICLONE, src.as_raw_fd()) };
539
540    match res {
541        -1 => {
542            let err = std::io::Error::last_os_error();
543
544            drop(dst);
545            let _ = fs::remove_file(dst_path);
546
547            match err.raw_os_error().unwrap() {
548                // The filesystem does not support reflinks.
549                libc::EOPNOTSUPP |
550                // src_path and dst_path are different filesystems.
551                libc::EXDEV |
552                // An invalid ioctl number was specified in an ioctl system call.
553                libc::ENOTTY => {
554                    Err(std::io::Error::new(std::io::ErrorKind::Unsupported, err))
555                }
556                _ => Err(err),
557            }
558        }
559        _ => Ok(()),
560    }
561}
562
563#[cfg(not(target_os = "linux"))]
564async fn reflink(_src_path: &Utf8Path, _dst_path: &Utf8Path) -> Result<(), std::io::Error> {
565    use libc as _;
566    Err(std::io::Error::from(std::io::ErrorKind::Unsupported))
567}
568
569async fn copy_blob(src: &Utf8Path, dst: &Utf8Path) -> Result<()> {
570    let temp_path = create_temp_file(dst).await?;
571    match reflink(src, (*temp_path).try_into()?).await {
572        Ok(()) => {}
573        Err(e) if e.kind() == std::io::ErrorKind::Unsupported => {
574            let src = src.to_owned();
575            let temp_path = temp_path.to_path_buf();
576            fasync::unblock(move || fs::copy(src, &temp_path)).await?;
577        }
578        Err(e) => return Err(anyhow!(e)),
579    }
580    temp_path.persist(dst)?;
581
582    set_blob_read_only(dst).await
583}
584
585pub(crate) async fn generate_delivery_blob(
586    src: &Utf8Path,
587    dst: &Utf8Path,
588    blob_type: DeliveryBlobType,
589) -> Result<()> {
590    let src_blob = fs::read(src).with_context(|| format!("reading {src}"))?;
591
592    let temp_path = create_temp_file(dst).await?;
593    let file = std::fs::File::create(&temp_path)?;
594    fasync::unblock(move || {
595        delivery_blob::generate_to(blob_type, &src_blob, std::io::BufWriter::new(file))
596    })
597    .await
598    .context("generate delivery blob")?;
599
600    temp_path.persist(dst)?;
601
602    set_blob_read_only(dst).await
603}
604
605#[cfg(not(target_os = "fuchsia"))]
606#[pin_project::pin_project]
607struct WatchStream {
608    _watcher: notify::RecommendedWatcher,
609    #[pin]
610    receiver: futures::channel::mpsc::Receiver<()>,
611}
612
613#[cfg(not(target_os = "fuchsia"))]
614impl Stream for WatchStream {
615    type Item = ();
616    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
617        self.project().receiver.poll_next(cx)
618    }
619}
620
621/// Make sure the resource is inside the repo_path.
622fn sanitize_path(repo_path: &Utf8Path, resource_path: &str) -> Result<Utf8PathBuf, Error> {
623    let resource_path = Utf8Path::new(resource_path);
624
625    let mut parts = vec![];
626    for component in resource_path.components() {
627        match component {
628            Utf8Component::Normal(part) => {
629                parts.push(part);
630            }
631            _ => {
632                warn!("invalid resource_path: {}", resource_path);
633                return Err(Error::InvalidPath(resource_path.into()));
634            }
635        }
636    }
637
638    let path = parts.into_iter().collect::<Utf8PathBuf>();
639    Ok(repo_path.join(path))
640}
641
642#[derive(Debug, thiserror::Error)]
643#[error(
644    "blob {hash} at {path:?} is {file_size} bytes in size, \
645     but the package manifest indicates it should be {manifest_size} bytes in size"
646)]
647struct BlobSizeMismatchError {
648    hash: Hash,
649    path: Utf8PathBuf,
650    manifest_size: u64,
651    file_size: u64,
652}
653
654#[cfg(test)]
655mod tests {
656    use super::*;
657    use crate::repository::repo_tests::{self, TestEnv as _};
658    use crate::util::CHUNK_SIZE;
659    use assert_matches::assert_matches;
660    use fuchsia_async as fasync;
661    use futures::{FutureExt, StreamExt};
662    use std::fs::File;
663    use std::io::Write as _;
664    use std::time::Duration;
665    struct TestEnv {
666        _tmp: tempfile::TempDir,
667        metadata_path: Utf8PathBuf,
668        blob_path: Utf8PathBuf,
669        repo: FileSystemRepository,
670    }
671
672    impl TestEnv {
673        fn new() -> Self {
674            let tmp = tempfile::tempdir().unwrap();
675            let dir = Utf8Path::from_path(tmp.path()).unwrap();
676            let metadata_path = dir.join("metadata");
677            let blob_path = dir.join("blobs");
678            std::fs::create_dir(&metadata_path).unwrap();
679            std::fs::create_dir(&blob_path).unwrap();
680
681            Self {
682                _tmp: tmp,
683                metadata_path: metadata_path.clone(),
684                blob_path: blob_path.clone(),
685                repo: FileSystemRepository::new(metadata_path, blob_path),
686            }
687        }
688    }
689
690    impl repo_tests::TestEnv for TestEnv {
691        fn supports_range(&self) -> bool {
692            true
693        }
694
695        fn write_metadata(&self, path: &str, bytes: &[u8]) {
696            let file_path = self.metadata_path.join(path);
697            let mut f = File::create(file_path).unwrap();
698            f.write_all(bytes).unwrap();
699        }
700
701        fn write_blob(&self, path: &str, bytes: &[u8]) {
702            let file_path = self.blob_path.join(path);
703            let mut f = File::create(file_path).unwrap();
704            f.write_all(bytes).unwrap();
705        }
706
707        fn repo(&self) -> &dyn RepoProvider {
708            &self.repo
709        }
710    }
711
712    repo_tests::repo_test_suite! {
713        env = TestEnv::new();
714        chunk_size = CHUNK_SIZE;
715    }
716
717    #[fuchsia::test]
718    async fn test_blob_modification_time() {
719        let env = TestEnv::new();
720
721        let f = File::create(env.blob_path.join("empty-blob")).unwrap();
722        let blob_mtime = f.metadata().unwrap().modified().unwrap();
723        drop(f);
724
725        assert_matches!(
726            env.repo.blob_modification_time("empty-blob").await,
727            Ok(Some(t)) if t == blob_mtime
728        );
729    }
730
731    #[fuchsia::test]
732    async fn test_reject_invalid_paths() {
733        let env = TestEnv::new();
734        env.write_metadata("empty", b"");
735
736        assert_matches!(repo_tests::read_metadata(&env, "empty", Range::Full).await, Ok(body) if body == b"");
737        assert_matches!(repo_tests::read_metadata(&env, "subdir/../empty", Range::Full).await,
738            Err(Error::InvalidPath(path)) if path == Utf8Path::new("subdir/../empty")
739        );
740    }
741
742    #[fuchsia::test]
743    async fn test_watch() {
744        let env = TestEnv::new();
745
746        // We support watch.
747        assert!(env.repo.supports_watch());
748
749        let mut watch_stream = env.repo.watch().unwrap().fuse();
750
751        // Try to read from the stream. This should not return anything since we haven't created a
752        // file yet.
753        futures::select! {
754            _ = watch_stream.next() => panic!("should not have received an event"),
755            _ = fasync::Timer::new(Duration::from_millis(10)).fuse() => (),
756        };
757
758        // Next, write to the file and make sure we observe an event.
759        env.write_metadata("timestamp.json", br#"{"version":1}"#);
760
761        futures::select! {
762            result = watch_stream.next() => {
763                assert_eq!(result, Some(()));
764            },
765            _ = fasync::Timer::new(Duration::from_secs(10)).fuse() => {
766                panic!("wrote to timestamp.json, but did not get an event");
767            },
768        };
769
770        // Write to the file again and make sure we receive another event.
771        env.write_metadata("timestamp.json", br#"{"version":2}"#);
772
773        futures::select! {
774            result = watch_stream.next() => {
775                assert_eq!(result, Some(()));
776            },
777            _ = fasync::Timer::new(Duration::from_secs(10)).fuse() => {
778                panic!("wrote to timestamp.json, but did not get an event");
779            },
780        };
781
782        // FIXME(https://github.com/notify-rs/notify/pull/337): On OSX, notify uses a
783        // crossbeam-channel in `Drop` to shut down the interior thread. Unfortunately this can
784        // trip over an issue where OSX will tear down the thread local storage before shutting
785        // down the thread, which can trigger a panic. To avoid this issue, sleep a little bit
786        // after shutting down our stream.
787        drop(watch_stream);
788        fasync::Timer::new(Duration::from_millis(100)).await;
789    }
790
791    #[fuchsia::test]
792    async fn test_watch_ignores_access() {
793        let env = TestEnv::new();
794
795        // Write an initial timestamp.json file.
796        env.write_metadata("timestamp.json", br#"{"version":1}"#);
797
798        let mut watch_stream = env.repo.watch().unwrap().fuse();
799
800        // Opening and reading timestamp.json should not trigger a watch event.
801        let _ = std::fs::read(env.metadata_path.join("timestamp.json")).unwrap();
802
803        futures::select! {
804            _ = watch_stream.next() => {
805                panic!("reading timestamp.json should not trigger a watch event")
806            }
807            _ = fasync::Timer::new(Duration::from_millis(200)).fuse() => (),
808        };
809
810        drop(watch_stream);
811        fasync::Timer::new(Duration::from_millis(100)).await;
812    }
813
814    #[fuchsia::test]
815    async fn test_store_blob_verifies_src_length() {
816        let tmp = tempfile::tempdir().unwrap();
817        let dir = Utf8Path::from_path(tmp.path()).unwrap();
818
819        let metadata_repo_path = dir.join("metadata");
820        let blob_repo_path = dir.join("blobs");
821        std::fs::create_dir(&metadata_repo_path).unwrap();
822        std::fs::create_dir(&blob_repo_path).unwrap();
823
824        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
825            .copy_mode(CopyMode::Copy)
826            .build();
827
828        // Store the blob.
829        let contents = b"hello world";
830        let path = dir.join("my-blob");
831        std::fs::write(&path, contents).unwrap();
832
833        let hash = fuchsia_merkle::root_from_slice(contents);
834        let err = repo.store_blob(&hash, contents.len() as u64 + 1, &path).await.unwrap_err();
835        assert_matches!(err.downcast_ref::<BlobSizeMismatchError>(), Some(_));
836    }
837
838    #[fuchsia::test]
839    async fn test_store_blob_copy_detects_length_mismatch() {
840        let tmp = tempfile::tempdir().unwrap();
841        let dir = Utf8Path::from_path(tmp.path()).unwrap();
842
843        let metadata_repo_path = dir.join("metadata");
844        let blob_repo_path = dir.join("blobs");
845        std::fs::create_dir(&metadata_repo_path).unwrap();
846        std::fs::create_dir(&blob_repo_path).unwrap();
847
848        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
849            .copy_mode(CopyMode::Copy)
850            .build();
851
852        // The blob contents and its hash.
853        let contents = b"hello world";
854        let hash = fuchsia_merkle::root_from_slice(contents);
855
856        let path = dir.join("my-blob");
857        std::fs::write(&path, contents).unwrap();
858
859        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
860
861        // Make sure we can read it back.
862        let blob_path = blob_repo_path.join(format!("1/{hash}"));
863        let delivery_blob = std::fs::read(&blob_path).unwrap();
864        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
865        assert_eq!(&actual, &contents[..]);
866
867        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
868
869        // Next, overwrite a blob that already exists.
870        let contents2 = b"another hello world";
871        let path2 = dir.join("my-blob2");
872        std::fs::write(&path2, contents2).unwrap();
873        assert_matches!(repo.store_blob(&hash, contents2.len() as u64, &path2).await, Ok(()));
874
875        // Make sure we get the new contents back.
876        let delivery_blob = std::fs::read(&blob_path).unwrap();
877        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
878        assert_eq!(&actual, &contents2[..]);
879    }
880
881    #[fuchsia::test]
882    async fn test_store_blob_copy_skips_present_blobs_of_correct_length() {
883        let tmp = tempfile::tempdir().unwrap();
884        let dir = Utf8Path::from_path(tmp.path()).unwrap();
885
886        let metadata_repo_path = dir.join("metadata");
887        let blob_repo_path = dir.join("blobs");
888        std::fs::create_dir(&metadata_repo_path).unwrap();
889        std::fs::create_dir(&blob_repo_path).unwrap();
890
891        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
892            .copy_mode(CopyMode::Copy)
893            .build();
894
895        // Store the blob.
896        let contents = b"hello world.";
897        let path = dir.join("my-blob");
898        std::fs::write(&path, contents).unwrap();
899
900        let hash = fuchsia_merkle::root_from_slice(contents);
901        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
902
903        // Make sure we can read it back.
904        let blob_path = blob_repo_path.join(format!("1/{hash}"));
905        let delivery_blob = std::fs::read(&blob_path).unwrap();
906        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
907        assert_eq!(&actual, &contents[..]);
908
909        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
910
911        // Next, we won't overwrite a blob that already exists.
912        let contents2 = b"Hello World!";
913        let path2 = dir.join("my-blob2");
914        std::fs::write(&path2, contents2).unwrap();
915        assert_matches!(repo.store_blob(&hash, contents2.len() as u64, &path2).await, Ok(()));
916
917        // Make sure we get the original contents back.
918        let delivery_blob = std::fs::read(&blob_path).unwrap();
919        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
920        assert_eq!(&actual, &contents[..]);
921    }
922
923    #[fuchsia::test]
924    async fn test_store_blob_copy_overwrite() {
925        let tmp = tempfile::tempdir().unwrap();
926        let dir = Utf8Path::from_path(tmp.path()).unwrap();
927
928        let metadata_repo_path = dir.join("metadata");
929        let blob_repo_path = dir.join("blobs");
930        std::fs::create_dir(&metadata_repo_path).unwrap();
931        std::fs::create_dir(&blob_repo_path).unwrap();
932
933        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
934            .copy_mode(CopyMode::CopyOverwrite)
935            .build();
936
937        // Store the blob.
938        let contents = b"hello world";
939        let path = dir.join("my-blob");
940        std::fs::write(&path, contents).unwrap();
941
942        let hash = fuchsia_merkle::root_from_slice(contents);
943        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
944
945        // Make sure we can read it back.
946        let blob_path = blob_repo_path.join(format!("1/{hash}"));
947        let delivery_blob = std::fs::read(&blob_path).unwrap();
948        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
949        assert_eq!(&actual, &contents[..]);
950
951        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
952
953        // Next, overwrite a blob that already exists.
954        let contents2 = b"another blob";
955        let path2 = dir.join("my-blob2");
956        std::fs::write(&path2, contents2).unwrap();
957        assert_matches!(repo.store_blob(&hash, contents2.len() as u64, &path2).await, Ok(()));
958
959        // Make sure we get the new contents back.
960        let delivery_blob = std::fs::read(&blob_path).unwrap();
961        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
962        assert_eq!(&actual, &contents2[..]);
963    }
964
965    #[fuchsia::test]
966    async fn test_store_delivery_blob_hard_link() {
967        let tmp = tempfile::tempdir().unwrap();
968        let dir = Utf8Path::from_path(tmp.path()).unwrap();
969
970        let metadata_repo_path = dir.join("metadata");
971        let blob_repo_path = dir.join("blobs");
972        std::fs::create_dir(&metadata_repo_path).unwrap();
973        std::fs::create_dir(&blob_repo_path).unwrap();
974
975        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
976            .copy_mode(CopyMode::HardLink)
977            .build();
978
979        // Store the blob.
980        let contents = b"hello world";
981        let hash = fuchsia_merkle::root_from_slice(contents);
982
983        let uncompressed_path = dir.join("my-blob");
984        std::fs::write(&uncompressed_path, contents).unwrap();
985        let path = dir.join("my-delivery-blob");
986        generate_delivery_blob(&uncompressed_path, &path, DeliveryBlobType::Type1).await.unwrap();
987
988        assert_matches!(
989            repo.store_delivery_blob(&hash, &path, DeliveryBlobType::Type1).await,
990            Ok(())
991        );
992
993        // Make sure we can read it back.
994        let blob_path = blob_repo_path.join(format!("1/{hash}"));
995        let delivery_blob = std::fs::read(&blob_path).unwrap();
996        let actual: Vec<u8> = delivery_blob::decompress(&delivery_blob).unwrap();
997        assert_eq!(&actual, &contents[..]);
998
999        #[cfg(target_family = "unix")]
1000        async fn check_links(blob_path: &Utf8Path) {
1001            use std::os::unix::fs::MetadataExt as _;
1002
1003            assert_eq!(std::fs::metadata(blob_path).unwrap().nlink(), 2);
1004        }
1005
1006        #[cfg(not(target_family = "unix"))]
1007        async fn check_links(_blob_path: &Utf8Path) {}
1008
1009        // Make sure the hard link count was incremented.
1010        check_links(&blob_path).await;
1011    }
1012
1013    #[fuchsia::test]
1014    async fn test_store_blob_generates_delivery_blob() {
1015        let tmp = tempfile::tempdir().unwrap();
1016        let dir = Utf8Path::from_path(tmp.path()).unwrap();
1017
1018        let metadata_repo_path = dir.join("metadata");
1019        let blob_repo_path = dir.join("blobs");
1020        std::fs::create_dir(&metadata_repo_path).unwrap();
1021        std::fs::create_dir(&blob_repo_path).unwrap();
1022
1023        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
1024            .delivery_blob_type(DeliveryBlobType::Type1)
1025            .build();
1026
1027        // Store the blob.
1028        let contents = b"hello world";
1029        let hash = fuchsia_merkle::root_from_slice(contents);
1030
1031        let path = dir.join("my-blob");
1032        std::fs::write(&path, contents).unwrap();
1033
1034        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
1035
1036        // Make sure we can read the delivery blob.
1037        let blob_path = blob_repo_path.join("1").join(hash.to_string());
1038        let delivery_blob = std::fs::read(&blob_path).unwrap();
1039        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
1040        assert_eq!(&actual, &contents[..]);
1041
1042        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
1043    }
1044
1045    #[fuchsia::test]
1046    async fn test_store_delivery_blob() {
1047        let tmp = tempfile::tempdir().unwrap();
1048        let dir = Utf8Path::from_path(tmp.path()).unwrap();
1049
1050        let metadata_repo_path = dir.join("metadata");
1051        let blob_repo_path = dir.join("blobs");
1052        std::fs::create_dir(&metadata_repo_path).unwrap();
1053        std::fs::create_dir(&blob_repo_path).unwrap();
1054
1055        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
1056            .delivery_blob_type(DeliveryBlobType::Type1)
1057            .build();
1058
1059        // Store the blob.
1060        let contents = b"hello world";
1061        let uncompressed_path = dir.join("my-blob");
1062        std::fs::write(&uncompressed_path, contents).unwrap();
1063        let path = dir.join("my-delivery-blob");
1064        generate_delivery_blob(&uncompressed_path, &path, DeliveryBlobType::Type1).await.unwrap();
1065        let delivery_blob = std::fs::read(&path).unwrap();
1066
1067        let hash = fuchsia_merkle::root_from_slice(contents);
1068        assert_matches!(
1069            repo.store_delivery_blob(&hash, &path, DeliveryBlobType::Type1).await,
1070            Ok(())
1071        );
1072
1073        // Make sure we can read the delivery blob.
1074        let blob_path = blob_repo_path.join("1").join(hash.to_string());
1075        let stored_delivery_blob = std::fs::read(&blob_path).unwrap();
1076        assert_eq!(stored_delivery_blob, delivery_blob);
1077
1078        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
1079    }
1080}
fuchsia_repo/repository/file_system.rs

fuchsia_repo/repository/
file_system.rs
