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    #[async_trait::async_trait]
691    impl repo_tests::TestEnv for TestEnv {
692        fn supports_range(&self) -> bool {
693            true
694        }
695
696        fn write_metadata(&self, path: &str, bytes: &[u8]) {
697            let file_path = self.metadata_path.join(path);
698            let mut f = File::create(file_path).unwrap();
699            f.write_all(bytes).unwrap();
700        }
701
702        fn write_blob(&self, path: &str, bytes: &[u8]) {
703            let file_path = self.blob_path.join(path);
704            let mut f = File::create(file_path).unwrap();
705            f.write_all(bytes).unwrap();
706        }
707
708        fn repo(&self) -> &dyn RepoProvider {
709            &self.repo
710        }
711    }
712
713    repo_tests::repo_test_suite! {
714        env = TestEnv::new();
715        chunk_size = CHUNK_SIZE;
716    }
717
718    #[fuchsia::test]
719    async fn test_blob_modification_time() {
720        let env = TestEnv::new();
721
722        let f = File::create(env.blob_path.join("empty-blob")).unwrap();
723        let blob_mtime = f.metadata().unwrap().modified().unwrap();
724        drop(f);
725
726        assert_matches!(
727            env.repo.blob_modification_time("empty-blob").await,
728            Ok(Some(t)) if t == blob_mtime
729        );
730    }
731
732    #[fuchsia::test]
733    async fn test_reject_invalid_paths() {
734        let env = TestEnv::new();
735        env.write_metadata("empty", b"");
736
737        assert_matches!(repo_tests::read_metadata(&env, "empty", Range::Full).await, Ok(body) if body == b"");
738        assert_matches!(repo_tests::read_metadata(&env, "subdir/../empty", Range::Full).await,
739            Err(Error::InvalidPath(path)) if path == Utf8Path::new("subdir/../empty")
740        );
741    }
742
743    #[fuchsia::test]
744    async fn test_watch() {
745        let env = TestEnv::new();
746
747        // We support watch.
748        assert!(env.repo.supports_watch());
749
750        let mut watch_stream = env.repo.watch().unwrap().fuse();
751
752        // Try to read from the stream. This should not return anything since we haven't created a
753        // file yet.
754        futures::select! {
755            _ = watch_stream.next() => panic!("should not have received an event"),
756            _ = fasync::Timer::new(Duration::from_millis(10)).fuse() => (),
757        };
758
759        // Next, write to the file and make sure we observe an event.
760        env.write_metadata("timestamp.json", br#"{"version":1}"#);
761
762        futures::select! {
763            result = watch_stream.next() => {
764                assert_eq!(result, Some(()));
765            },
766            _ = fasync::Timer::new(Duration::from_secs(10)).fuse() => {
767                panic!("wrote to timestamp.json, but did not get an event");
768            },
769        };
770
771        // Write to the file again and make sure we receive another event.
772        env.write_metadata("timestamp.json", br#"{"version":2}"#);
773
774        futures::select! {
775            result = watch_stream.next() => {
776                assert_eq!(result, Some(()));
777            },
778            _ = fasync::Timer::new(Duration::from_secs(10)).fuse() => {
779                panic!("wrote to timestamp.json, but did not get an event");
780            },
781        };
782
783        // FIXME(https://github.com/notify-rs/notify/pull/337): On OSX, notify uses a
784        // crossbeam-channel in `Drop` to shut down the interior thread. Unfortunately this can
785        // trip over an issue where OSX will tear down the thread local storage before shutting
786        // down the thread, which can trigger a panic. To avoid this issue, sleep a little bit
787        // after shutting down our stream.
788        drop(watch_stream);
789        fasync::Timer::new(Duration::from_millis(100)).await;
790    }
791
792    #[fuchsia::test]
793    async fn test_watch_ignores_access() {
794        let env = TestEnv::new();
795
796        // Write an initial timestamp.json file.
797        env.write_metadata("timestamp.json", br#"{"version":1}"#);
798
799        let mut watch_stream = env.repo.watch().unwrap().fuse();
800
801        // Opening and reading timestamp.json should not trigger a watch event.
802        let _ = std::fs::read(env.metadata_path.join("timestamp.json")).unwrap();
803
804        futures::select! {
805            _ = watch_stream.next() => {
806                panic!("reading timestamp.json should not trigger a watch event")
807            }
808            _ = fasync::Timer::new(Duration::from_millis(200)).fuse() => (),
809        };
810
811        drop(watch_stream);
812        fasync::Timer::new(Duration::from_millis(100)).await;
813    }
814
815    #[fuchsia::test]
816    async fn test_store_blob_verifies_src_length() {
817        let tmp = tempfile::tempdir().unwrap();
818        let dir = Utf8Path::from_path(tmp.path()).unwrap();
819
820        let metadata_repo_path = dir.join("metadata");
821        let blob_repo_path = dir.join("blobs");
822        std::fs::create_dir(&metadata_repo_path).unwrap();
823        std::fs::create_dir(&blob_repo_path).unwrap();
824
825        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
826            .copy_mode(CopyMode::Copy)
827            .build();
828
829        // Store the blob.
830        let contents = b"hello world";
831        let path = dir.join("my-blob");
832        std::fs::write(&path, contents).unwrap();
833
834        let hash = fuchsia_merkle::root_from_slice(contents);
835        let err = repo.store_blob(&hash, contents.len() as u64 + 1, &path).await.unwrap_err();
836        assert_matches!(err.downcast_ref::<BlobSizeMismatchError>(), Some(_));
837    }
838
839    #[fuchsia::test]
840    async fn test_store_blob_copy_detects_length_mismatch() {
841        let tmp = tempfile::tempdir().unwrap();
842        let dir = Utf8Path::from_path(tmp.path()).unwrap();
843
844        let metadata_repo_path = dir.join("metadata");
845        let blob_repo_path = dir.join("blobs");
846        std::fs::create_dir(&metadata_repo_path).unwrap();
847        std::fs::create_dir(&blob_repo_path).unwrap();
848
849        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
850            .copy_mode(CopyMode::Copy)
851            .build();
852
853        // The blob contents and its hash.
854        let contents = b"hello world";
855        let hash = fuchsia_merkle::root_from_slice(contents);
856
857        let path = dir.join("my-blob");
858        std::fs::write(&path, contents).unwrap();
859
860        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
861
862        // Make sure we can read it back.
863        let blob_path = blob_repo_path.join(format!("1/{hash}"));
864        let delivery_blob = std::fs::read(&blob_path).unwrap();
865        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
866        assert_eq!(&actual, &contents[..]);
867
868        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
869
870        // Next, overwrite a blob that already exists.
871        let contents2 = b"another hello world";
872        let path2 = dir.join("my-blob2");
873        std::fs::write(&path2, contents2).unwrap();
874        assert_matches!(repo.store_blob(&hash, contents2.len() as u64, &path2).await, Ok(()));
875
876        // Make sure we get the new contents back.
877        let delivery_blob = std::fs::read(&blob_path).unwrap();
878        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
879        assert_eq!(&actual, &contents2[..]);
880    }
881
882    #[fuchsia::test]
883    async fn test_store_blob_copy_skips_present_blobs_of_correct_length() {
884        let tmp = tempfile::tempdir().unwrap();
885        let dir = Utf8Path::from_path(tmp.path()).unwrap();
886
887        let metadata_repo_path = dir.join("metadata");
888        let blob_repo_path = dir.join("blobs");
889        std::fs::create_dir(&metadata_repo_path).unwrap();
890        std::fs::create_dir(&blob_repo_path).unwrap();
891
892        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
893            .copy_mode(CopyMode::Copy)
894            .build();
895
896        // Store the blob.
897        let contents = b"hello world.";
898        let path = dir.join("my-blob");
899        std::fs::write(&path, contents).unwrap();
900
901        let hash = fuchsia_merkle::root_from_slice(contents);
902        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
903
904        // Make sure we can read it back.
905        let blob_path = blob_repo_path.join(format!("1/{hash}"));
906        let delivery_blob = std::fs::read(&blob_path).unwrap();
907        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
908        assert_eq!(&actual, &contents[..]);
909
910        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
911
912        // Next, we won't overwrite a blob that already exists.
913        let contents2 = b"Hello World!";
914        let path2 = dir.join("my-blob2");
915        std::fs::write(&path2, contents2).unwrap();
916        assert_matches!(repo.store_blob(&hash, contents2.len() as u64, &path2).await, Ok(()));
917
918        // Make sure we get the original contents back.
919        let delivery_blob = std::fs::read(&blob_path).unwrap();
920        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
921        assert_eq!(&actual, &contents[..]);
922    }
923
924    #[fuchsia::test]
925    async fn test_store_blob_copy_overwrite() {
926        let tmp = tempfile::tempdir().unwrap();
927        let dir = Utf8Path::from_path(tmp.path()).unwrap();
928
929        let metadata_repo_path = dir.join("metadata");
930        let blob_repo_path = dir.join("blobs");
931        std::fs::create_dir(&metadata_repo_path).unwrap();
932        std::fs::create_dir(&blob_repo_path).unwrap();
933
934        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
935            .copy_mode(CopyMode::CopyOverwrite)
936            .build();
937
938        // Store the blob.
939        let contents = b"hello world";
940        let path = dir.join("my-blob");
941        std::fs::write(&path, contents).unwrap();
942
943        let hash = fuchsia_merkle::root_from_slice(contents);
944        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
945
946        // Make sure we can read it back.
947        let blob_path = blob_repo_path.join(format!("1/{hash}"));
948        let delivery_blob = std::fs::read(&blob_path).unwrap();
949        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
950        assert_eq!(&actual, &contents[..]);
951
952        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
953
954        // Next, overwrite a blob that already exists.
955        let contents2 = b"another blob";
956        let path2 = dir.join("my-blob2");
957        std::fs::write(&path2, contents2).unwrap();
958        assert_matches!(repo.store_blob(&hash, contents2.len() as u64, &path2).await, Ok(()));
959
960        // Make sure we get the new contents back.
961        let delivery_blob = std::fs::read(&blob_path).unwrap();
962        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
963        assert_eq!(&actual, &contents2[..]);
964    }
965
966    #[fuchsia::test]
967    async fn test_store_delivery_blob_hard_link() {
968        let tmp = tempfile::tempdir().unwrap();
969        let dir = Utf8Path::from_path(tmp.path()).unwrap();
970
971        let metadata_repo_path = dir.join("metadata");
972        let blob_repo_path = dir.join("blobs");
973        std::fs::create_dir(&metadata_repo_path).unwrap();
974        std::fs::create_dir(&blob_repo_path).unwrap();
975
976        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
977            .copy_mode(CopyMode::HardLink)
978            .build();
979
980        // Store the blob.
981        let contents = b"hello world";
982        let hash = fuchsia_merkle::root_from_slice(contents);
983
984        let uncompressed_path = dir.join("my-blob");
985        std::fs::write(&uncompressed_path, contents).unwrap();
986        let path = dir.join("my-delivery-blob");
987        generate_delivery_blob(&uncompressed_path, &path, DeliveryBlobType::Type1).await.unwrap();
988
989        assert_matches!(
990            repo.store_delivery_blob(&hash, &path, DeliveryBlobType::Type1).await,
991            Ok(())
992        );
993
994        // Make sure we can read it back.
995        let blob_path = blob_repo_path.join(format!("1/{hash}"));
996        let delivery_blob = std::fs::read(&blob_path).unwrap();
997        let actual: Vec<u8> = delivery_blob::decompress(&delivery_blob).unwrap();
998        assert_eq!(&actual, &contents[..]);
999
1000        #[cfg(target_family = "unix")]
1001        async fn check_links(blob_path: &Utf8Path) {
1002            use std::os::unix::fs::MetadataExt as _;
1003
1004            assert_eq!(std::fs::metadata(blob_path).unwrap().nlink(), 2);
1005        }
1006
1007        #[cfg(not(target_family = "unix"))]
1008        async fn check_links(_blob_path: &Utf8Path) {}
1009
1010        // Make sure the hard link count was incremented.
1011        check_links(&blob_path).await;
1012    }
1013
1014    #[fuchsia::test]
1015    async fn test_store_blob_generates_delivery_blob() {
1016        let tmp = tempfile::tempdir().unwrap();
1017        let dir = Utf8Path::from_path(tmp.path()).unwrap();
1018
1019        let metadata_repo_path = dir.join("metadata");
1020        let blob_repo_path = dir.join("blobs");
1021        std::fs::create_dir(&metadata_repo_path).unwrap();
1022        std::fs::create_dir(&blob_repo_path).unwrap();
1023
1024        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
1025            .delivery_blob_type(DeliveryBlobType::Type1)
1026            .build();
1027
1028        // Store the blob.
1029        let contents = b"hello world";
1030        let hash = fuchsia_merkle::root_from_slice(contents);
1031
1032        let path = dir.join("my-blob");
1033        std::fs::write(&path, contents).unwrap();
1034
1035        assert_matches!(repo.store_blob(&hash, contents.len() as u64, &path).await, Ok(()));
1036
1037        // Make sure we can read the delivery blob.
1038        let blob_path = blob_repo_path.join("1").join(hash.to_string());
1039        let delivery_blob = std::fs::read(&blob_path).unwrap();
1040        let actual = delivery_blob::decompress(&delivery_blob).unwrap();
1041        assert_eq!(&actual, &contents[..]);
1042
1043        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
1044    }
1045
1046    #[fuchsia::test]
1047    async fn test_store_delivery_blob() {
1048        let tmp = tempfile::tempdir().unwrap();
1049        let dir = Utf8Path::from_path(tmp.path()).unwrap();
1050
1051        let metadata_repo_path = dir.join("metadata");
1052        let blob_repo_path = dir.join("blobs");
1053        std::fs::create_dir(&metadata_repo_path).unwrap();
1054        std::fs::create_dir(&blob_repo_path).unwrap();
1055
1056        let repo = FileSystemRepository::builder(metadata_repo_path, blob_repo_path.clone())
1057            .delivery_blob_type(DeliveryBlobType::Type1)
1058            .build();
1059
1060        // Store the blob.
1061        let contents = b"hello world";
1062        let uncompressed_path = dir.join("my-blob");
1063        std::fs::write(&uncompressed_path, contents).unwrap();
1064        let path = dir.join("my-delivery-blob");
1065        generate_delivery_blob(&uncompressed_path, &path, DeliveryBlobType::Type1).await.unwrap();
1066        let delivery_blob = std::fs::read(&path).unwrap();
1067
1068        let hash = fuchsia_merkle::root_from_slice(contents);
1069        assert_matches!(
1070            repo.store_delivery_blob(&hash, &path, DeliveryBlobType::Type1).await,
1071            Ok(())
1072        );
1073
1074        // Make sure we can read the delivery blob.
1075        let blob_path = blob_repo_path.join("1").join(hash.to_string());
1076        let stored_delivery_blob = std::fs::read(&blob_path).unwrap();
1077        assert_eq!(stored_delivery_blob, delivery_blob);
1078
1079        assert!(std::fs::metadata(&blob_path).unwrap().permissions().readonly());
1080    }
1081}
fuchsia_repo/repository/file_system.rs

fuchsia_repo/repository/
file_system.rs
