blob: 80c76517f955bca0bf21a9a48c2c0513498ab44f [file] [log] [blame]
//! Repository implementation backed by a file system.
use futures_io::AsyncRead;
use futures_util::future::{BoxFuture, FutureExt};
use futures_util::io::{copy, AllowStdIo};
use log::debug;
use std::collections::HashMap;
use std::fs::{DirBuilder, File};
use std::marker::PhantomData;
use std::path::{Path, PathBuf};
use tempfile::{NamedTempFile, TempPath};
use crate::interchange::DataInterchange;
use crate::metadata::{MetadataPath, MetadataVersion, TargetPath};
use crate::repository::{RepositoryProvider, RepositoryStorage};
use crate::Result;
/// A builder to create a repository contained on the local file system.
pub struct FileSystemRepositoryBuilder<D> {
local_path: PathBuf,
metadata_prefix: Option<PathBuf>,
targets_prefix: Option<PathBuf>,
_interchange: PhantomData<D>,
}
impl<D> FileSystemRepositoryBuilder<D>
where
D: DataInterchange,
{
/// Create a new repository with the given `local_path` prefix.
pub fn new<P: Into<PathBuf>>(local_path: P) -> Self {
FileSystemRepositoryBuilder {
local_path: local_path.into(),
metadata_prefix: None,
targets_prefix: None,
_interchange: PhantomData,
}
}
/// The argument `metadata_prefix` is used to provide an alternate path where metadata is
/// stored on the repository. If `None`, this defaults to `/`. For example, if there is a TUF
/// repository at `/usr/local/repo/`, but all metadata is stored at `/usr/local/repo/meta/`,
/// then passing the arg `Some("meta".into())` would cause `root.json` to be fetched from
/// `/usr/local/repo/meta/root.json`.
pub fn metadata_prefix<P: Into<PathBuf>>(mut self, metadata_prefix: P) -> Self {
self.metadata_prefix = Some(metadata_prefix.into());
self
}
/// The argument `targets_prefix` is used to provide an alternate path where targets are
/// stored on the repository. If `None`, this defaults to `/`. For example, if there is a TUF
/// repository at `/usr/local/repo/`, but all targets are stored at `/usr/local/repo/targets/`,
/// then passing the arg `Some("targets".into())` would cause `hello-world` to be fetched from
/// `/usr/local/repo/targets/hello-world`.
pub fn targets_prefix<P: Into<PathBuf>>(mut self, targets_prefix: P) -> Self {
self.targets_prefix = Some(targets_prefix.into());
self
}
/// Build a `FileSystemRepository`.
pub fn build(self) -> Result<FileSystemRepository<D>> {
let metadata_path = if let Some(metadata_prefix) = self.metadata_prefix {
self.local_path.join(metadata_prefix)
} else {
self.local_path.clone()
};
DirBuilder::new().recursive(true).create(&metadata_path)?;
let targets_path = if let Some(targets_prefix) = self.targets_prefix {
self.local_path.join(targets_prefix)
} else {
self.local_path.clone()
};
DirBuilder::new().recursive(true).create(&targets_path)?;
Ok(FileSystemRepository {
metadata_path,
targets_path,
_interchange: PhantomData,
})
}
}
/// A repository contained on the local file system.
#[derive(Debug)]
pub struct FileSystemRepository<D>
where
D: DataInterchange,
{
metadata_path: PathBuf,
targets_path: PathBuf,
_interchange: PhantomData<D>,
}
impl<D> FileSystemRepository<D>
where
D: DataInterchange,
{
/// Create a [FileSystemRepositoryBuilder].
pub fn builder<P: Into<PathBuf>>(local_path: P) -> FileSystemRepositoryBuilder<D> {
FileSystemRepositoryBuilder::new(local_path)
}
/// Create a new repository on the local file system.
pub fn new<P: Into<PathBuf>>(local_path: P) -> Result<Self> {
FileSystemRepositoryBuilder::new(local_path)
.metadata_prefix("metadata")
.targets_prefix("targets")
.build()
}
/// Returns a [FileSystemBatchUpdate] for manipulating this repository. This allows callers to
/// stage a number of mutations, and optionally write them all at once.
pub fn batch_update(&mut self) -> FileSystemBatchUpdate<D> {
FileSystemBatchUpdate {
repo: self,
metadata: HashMap::new(),
targets: HashMap::new(),
}
}
fn metadata_path(&self, meta_path: &MetadataPath, version: MetadataVersion) -> PathBuf {
let mut path = self.metadata_path.clone();
path.extend(meta_path.components::<D>(version));
path
}
fn target_path(&self, target_path: &TargetPath) -> PathBuf {
let mut path = self.targets_path.clone();
path.extend(target_path.components());
path
}
fn fetch_path(
&self,
path: &Path,
) -> BoxFuture<'_, Result<Box<dyn AsyncRead + Send + Unpin + '_>>> {
let reader = File::open(&path);
async move {
let reader = reader?;
let reader: Box<dyn AsyncRead + Send + Unpin> = Box::new(AllowStdIo::new(reader));
Ok(reader)
}
.boxed()
}
}
impl<D> RepositoryProvider<D> for FileSystemRepository<D>
where
D: DataInterchange + Sync,
{
fn fetch_metadata<'a>(
&'a self,
meta_path: &MetadataPath,
version: MetadataVersion,
) -> BoxFuture<'a, Result<Box<dyn AsyncRead + Send + Unpin + 'a>>> {
let path = self.metadata_path(meta_path, version);
self.fetch_path(&path)
}
fn fetch_target<'a>(
&'a self,
target_path: &TargetPath,
) -> BoxFuture<'a, Result<Box<dyn AsyncRead + Send + Unpin + 'a>>> {
let path = self.target_path(target_path);
self.fetch_path(&path)
}
}
impl<D> RepositoryStorage<D> for FileSystemRepository<D>
where
D: DataInterchange + Sync + Send,
{
fn store_metadata<'a>(
&'a mut self,
meta_path: &MetadataPath,
version: MetadataVersion,
metadata: &'a mut (dyn AsyncRead + Send + Unpin + 'a),
) -> BoxFuture<'a, Result<()>> {
let path = self.metadata_path(meta_path, version);
async move {
if path.exists() {
debug!("Metadata path exists. Overwriting: {:?}", path);
}
let mut temp_file = AllowStdIo::new(create_temp_file(&path)?);
copy(metadata, &mut temp_file).await?;
temp_file.into_inner().persist(&path)?;
Ok(())
}
.boxed()
}
fn store_target<'a>(
&'a mut self,
target_path: &TargetPath,
read: &'a mut (dyn AsyncRead + Send + Unpin + 'a),
) -> BoxFuture<'a, Result<()>> {
let path = self.target_path(target_path);
async move {
if path.exists() {
debug!("Target path exists. Overwriting: {:?}", path);
}
let mut temp_file = AllowStdIo::new(create_temp_file(&path)?);
copy(read, &mut temp_file).await?;
temp_file.into_inner().persist(&path)?;
Ok(())
}
.boxed()
}
}
/// [FileSystemBatchUpdate] is a special repository that is designed to write the metadata and
/// targets to an [FileSystemRepository] in a single batch.
///
/// Note: `FileSystemBatchUpdate::commit()` must be called in order to write the metadata and
/// targets to the [FileSystemRepository]. Otherwise any queued changes will be lost on drop.
#[derive(Debug)]
pub struct FileSystemBatchUpdate<'a, D: DataInterchange> {
repo: &'a mut FileSystemRepository<D>,
metadata: HashMap<PathBuf, TempPath>,
targets: HashMap<PathBuf, TempPath>,
}
impl<'a, D> FileSystemBatchUpdate<'a, D>
where
D: DataInterchange + Sync,
{
/// Write all the metadata and targets the [FileSystemBatchUpdate] to the source
/// [FileSystemRepository] in a single batch operation.
///
/// Note: While this function will atomically write each file, it's possible that this could
/// fail with part of the files written if we experience a system error during the process.
pub async fn commit(self) -> Result<()> {
for (path, tmp_path) in self.targets {
if path.exists() {
debug!("Target path exists. Overwriting: {:?}", path);
}
tmp_path.persist(path)?;
}
for (path, tmp_path) in self.metadata {
if path.exists() {
debug!("Metadata path exists. Overwriting: {:?}", path);
}
tmp_path.persist(path)?;
}
Ok(())
}
}
impl<D> RepositoryProvider<D> for FileSystemBatchUpdate<'_, D>
where
D: DataInterchange + Sync,
{
fn fetch_metadata<'a>(
&'a self,
meta_path: &MetadataPath,
version: MetadataVersion,
) -> BoxFuture<'a, Result<Box<dyn AsyncRead + Send + Unpin + 'a>>> {
let path = self.repo.metadata_path(meta_path, version);
if let Some(temp_path) = self.metadata.get(&path) {
self.repo.fetch_path(temp_path)
} else {
self.repo.fetch_path(&path)
}
}
fn fetch_target<'a>(
&'a self,
target_path: &TargetPath,
) -> BoxFuture<'a, Result<Box<dyn AsyncRead + Send + Unpin + 'a>>> {
let path = self.repo.target_path(target_path);
if let Some(temp_path) = self.targets.get(&path) {
self.repo.fetch_path(temp_path)
} else {
self.repo.fetch_path(&path)
}
}
}
impl<D> RepositoryStorage<D> for FileSystemBatchUpdate<'_, D>
where
D: DataInterchange + Sync,
{
fn store_metadata<'a>(
&'a mut self,
meta_path: &MetadataPath,
version: MetadataVersion,
read: &'a mut (dyn AsyncRead + Send + Unpin + 'a),
) -> BoxFuture<'a, Result<()>> {
let path = self.repo.metadata_path(meta_path, version);
let metadata = &mut self.metadata;
async move {
let mut temp_file = AllowStdIo::new(create_temp_file(&path)?);
copy(read, &mut temp_file).await?;
metadata.insert(path, temp_file.into_inner().into_temp_path());
Ok(())
}
.boxed()
}
fn store_target<'a>(
&'a mut self,
target_path: &TargetPath,
read: &'a mut (dyn AsyncRead + Send + Unpin + 'a),
) -> BoxFuture<'a, Result<()>> {
let path = self.repo.target_path(target_path);
let targets = &mut self.targets;
async move {
let mut temp_file = AllowStdIo::new(create_temp_file(&path)?);
copy(read, &mut temp_file).await?;
targets.insert(path, temp_file.into_inner().into_temp_path());
Ok(())
}
.boxed()
}
}
fn create_temp_file(path: &Path) -> Result<NamedTempFile> {
// We want to atomically write the file to make sure clients can never see a partially written
// file. In order to do this, we'll write to a temporary file in the same directory as our
// target, otherwise we risk writing the temporary file to one mountpoint, and then
// non-atomically copying the file to another mountpoint.
if let Some(parent) = path.parent() {
DirBuilder::new().recursive(true).create(parent)?;
Ok(NamedTempFile::new_in(parent)?)
} else {
Ok(NamedTempFile::new_in(".")?)
}
}
#[cfg(test)]
mod test {
use super::*;
use crate::error::Error;
use crate::interchange::Json;
use crate::metadata::RootMetadata;
use crate::repository::{fetch_metadata_to_string, fetch_target_to_string, Repository};
use assert_matches::assert_matches;
use futures_executor::block_on;
use futures_util::io::AsyncReadExt;
use tempfile;
#[test]
fn file_system_repo_metadata_not_found_error() {
block_on(async {
let temp_dir = tempfile::Builder::new()
.prefix("rust-tuf")
.tempdir()
.unwrap();
let repo = FileSystemRepositoryBuilder::new(temp_dir.path())
.build()
.unwrap();
assert_matches!(
Repository::<_, Json>::new(repo)
.fetch_metadata::<RootMetadata>(
&MetadataPath::root(),
MetadataVersion::None,
None,
vec![],
)
.await,
Err(Error::NotFound)
);
})
}
#[test]
fn file_system_repo_targets() {
block_on(async {
let temp_dir = tempfile::Builder::new()
.prefix("rust-tuf")
.tempdir()
.unwrap();
let mut repo = FileSystemRepositoryBuilder::<Json>::new(temp_dir.path().to_path_buf())
.metadata_prefix("meta")
.targets_prefix("targs")
.build()
.unwrap();
// test that init worked
assert!(temp_dir.path().join("meta").exists());
assert!(temp_dir.path().join("targs").exists());
let data: &[u8] = b"like tears in the rain";
let path = TargetPath::new("foo/bar/baz").unwrap();
repo.store_target(&path, &mut &*data).await.unwrap();
assert!(temp_dir
.path()
.join("targs")
.join("foo")
.join("bar")
.join("baz")
.exists());
let mut buf = Vec::new();
// Enclose `fetch_target` in a scope to make sure the file is closed.
// This is needed for `tempfile` on Windows, which doesn't open the
// files in a mode that allows the file to be opened multiple times.
{
let mut read = repo.fetch_target(&path).await.unwrap();
read.read_to_end(&mut buf).await.unwrap();
assert_eq!(buf.as_slice(), data);
}
// RepositoryProvider implementations do not guarantee data is not corrupt.
let bad_data: &[u8] = b"you're in a desert";
repo.store_target(&path, &mut &*bad_data).await.unwrap();
let mut read = repo.fetch_target(&path).await.unwrap();
buf.clear();
read.read_to_end(&mut buf).await.unwrap();
assert_eq!(buf.as_slice(), bad_data);
})
}
#[test]
fn file_system_repo_batch_update() {
block_on(async {
let temp_dir = tempfile::Builder::new()
.prefix("rust-tuf")
.tempdir()
.unwrap();
let mut repo = FileSystemRepositoryBuilder::<Json>::new(temp_dir.path().to_path_buf())
.metadata_prefix("meta")
.targets_prefix("targs")
.build()
.unwrap();
let meta_path = MetadataPath::new("meta").unwrap();
let meta_version = MetadataVersion::None;
let target_path = TargetPath::new("target").unwrap();
// First, write some stuff to the repository.
let committed_meta = "committed meta";
let committed_target = "committed target";
repo.store_metadata(&meta_path, meta_version, &mut committed_meta.as_bytes())
.await
.unwrap();
repo.store_target(&target_path, &mut committed_target.as_bytes())
.await
.unwrap();
let mut batch = repo.batch_update();
// Make sure we can read back the committed stuff.
assert_eq!(
fetch_metadata_to_string(&batch, &meta_path, meta_version)
.await
.unwrap(),
committed_meta,
);
assert_eq!(
fetch_target_to_string(&batch, &target_path).await.unwrap(),
committed_target,
);
// Next, stage some stuff in the batch_update.
let staged_meta = "staged meta";
let staged_target = "staged target";
batch
.store_metadata(&meta_path, meta_version, &mut staged_meta.as_bytes())
.await
.unwrap();
batch
.store_target(&target_path, &mut staged_target.as_bytes())
.await
.unwrap();
// Make sure it got staged.
assert_eq!(
fetch_metadata_to_string(&batch, &meta_path, meta_version)
.await
.unwrap(),
staged_meta,
);
assert_eq!(
fetch_target_to_string(&batch, &target_path).await.unwrap(),
staged_target,
);
// Next, drop the batch_update. We shouldn't have written the data back to the
// repository.
drop(batch);
assert_eq!(
fetch_metadata_to_string(&repo, &meta_path, meta_version)
.await
.unwrap(),
committed_meta,
);
assert_eq!(
fetch_target_to_string(&repo, &target_path).await.unwrap(),
committed_target,
);
// Do the batch_update again, but this time write the data.
let mut batch = repo.batch_update();
batch
.store_metadata(&meta_path, meta_version, &mut staged_meta.as_bytes())
.await
.unwrap();
batch
.store_target(&target_path, &mut staged_target.as_bytes())
.await
.unwrap();
batch.commit().await.unwrap();
// Make sure the new data got to the repository.
assert_eq!(
fetch_metadata_to_string(&repo, &meta_path, meta_version)
.await
.unwrap(),
staged_meta,
);
assert_eq!(
fetch_target_to_string(&repo, &target_path).await.unwrap(),
staged_target,
);
})
}
}