lib.rs
raw
//! # Repository
//!
//! The encrypted, deduplicating storage format for backups.
//!
//! A repository lives on a [`Backend`] — any store offering write-once
//! objects with get/put/list — and holds three kinds of object:
//!
//! - **`header`**: public, write-once bootstrap data: the Argon2 salt
//! and cost parameters, the chunker parameters, and a passphrase
//! verifier. No key material is stored anywhere, on any host: the
//! keys are re-derived from the passphrase (via the expensive KDF,
//! once per session) every time the repository is opened. The
//! passphrase *is* the key: it cannot be changed without re-encrypting
//! everything, and losing it loses the repository.
//! - **chunks**: content-defined (FastCDC) chunks of data, encrypted with
//! ChaCha20-Poly1305 and named by an HMAC-SHA256 of their plaintext, so
//! identical content is stored once without revealing content
//! fingerprints to whoever holds the storage.
//! - **snapshots**: small encrypted records pointing at a manifest — a
//! stream of [`Entry`] values describing every object in one backup —
//! which is itself stored through the chunk store.
//!
//! Every operation is idempotent: re-storing a chunk, a manifest, or a
//! snapshot that already exists is a cheap no-op. That property is what
//! lets a suspended or crashed backup simply redo recent work when it
//! resumes.
pub mod backend;
mod cache;
mod chunker;
mod crypto;
mod manifest;
pub mod os_path;
#[cfg(test)]
mod tests;
use std::io::Read;
use crate::crypto::{HEADER_VERSION, Header, MasterKeys, RepoConfig};
pub use crate::{
backend::{Backend, ObjectKey, ObjectKind, local::LocalBackend},
cache::{CacheIdentity, ChunkCache},
crypto::{ChunkId, LockId, PruneId, SnapshotId},
manifest::{Coverage, Entry, EntryKind, ManifestEntries, Snapshot},
manifest::{Lock, PruneRecord},
};
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("the backend already contains a repository")]
AlreadyInitialized,
#[error("the backend does not contain a repository")]
NotInitialized,
#[error("chunk {0} is not present in the repository")]
MissingChunk(ChunkId),
#[error("chunk {0} did not decrypt to content matching its id")]
CorruptChunk(ChunkId),
#[error("snapshot {0} is not present in the repository")]
MissingSnapshot(SnapshotId),
#[error("decryption failed: wrong key or corrupted data")]
Crypto,
#[error("wrong passphrase for this repository")]
WrongPassword,
#[error("the repository header is malformed")]
InvalidHeader,
#[error("the repository was created by a newer, incompatible version")]
UnsupportedVersion,
#[error("{0} is not a valid object name")]
InvalidObjectName(String),
#[error("backend error: {0}")]
Backend(String),
#[error("the backend is at capacity: {0}")]
BackendBusy(String),
#[error("chunking failed: {0}")]
Chunking(String),
#[error(transparent)]
CiboriumEncode(#[from] ciborium::ser::Error<std::io::Error>),
#[error(transparent)]
CiboriumDecode(#[from] ciborium::de::Error<std::io::Error>),
#[error(transparent)]
IO(#[from] std::io::Error),
}
/// A chunk already hashed and encrypted, awaiting its backend write.
///
/// Produced by [`Repository::prepare_chunk`], consumed by
/// [`Repository::store_prepared`].
pub struct PreparedChunk {
id: ChunkId,
sealed: Vec<u8>,
}
impl PreparedChunk {
pub fn id(&self) -> ChunkId {
self.id
}
/// The size of the encrypted object as it will be stored.
pub fn sealed_len(&self) -> usize {
self.sealed.len()
}
}
/// An open repository: a [`Backend`] plus the unlocked master keys.
///
/// All methods take `&self` and the type is `Send + Sync`, so pipeline
/// threads share one repository behind an `Arc`.
pub struct Repository {
backend: Box<dyn Backend>,
keys: MasterKeys,
config: RepoConfig,
/// A local record of chunks this repository is known to hold, saving
/// a round trip each. Optional, and advisory: without it every chunk
/// is asked about as before.
cache: Option<ChunkCache>,
/// Sealed bytes handed to the backend through this handle; see
/// [`Repository::bytes_written`].
written: std::sync::atomic::AtomicU64,
}
impl Repository {
/// Initializes a new repository on an empty backend: chooses a salt
/// and parameters, derives the keys from `password`, and stores the
/// public header. Nothing secret is written — creating a repository
/// with the same passphrase and header always yields the same keys.
pub fn create(
backend: Box<dyn Backend>,
password: impl AsRef<str>,
) -> Result<Repository, Error> {
if backend.contains(&ObjectKey::Header)? {
return Err(Error::AlreadyInitialized);
}
let mut header = Header::generate();
let keys = MasterKeys::derive(password.as_ref(), &header)?;
header.verifier = keys.verifier(&header);
let mut bytes = Vec::new();
ciborium::into_writer(&header, &mut bytes)?;
backend.put(&ObjectKey::Header, &bytes)?;
Ok(Repository {
backend,
keys,
config: header.config,
cache: None,
written: std::sync::atomic::AtomicU64::new(0),
})
}
/// Opens an existing repository, re-deriving its keys from
/// `password` and the stored header. This is the only point where
/// the expensive key derivation runs; a wrong passphrase is caught
/// here by the header's verifier and fails with
/// [`Error::WrongPassword`] before anything can be written under
/// mismatched keys.
pub fn open(backend: Box<dyn Backend>, password: impl AsRef<str>) -> Result<Repository, Error> {
let Some(bytes) = backend.get(&ObjectKey::Header)? else {
return Err(Error::NotInitialized);
};
let header: Header =
ciborium::from_reader(bytes.as_slice()).map_err(|_| Error::InvalidHeader)?;
if header.version != HEADER_VERSION {
return Err(Error::UnsupportedVersion);
}
let keys = MasterKeys::derive(password.as_ref(), &header)?;
if keys.verifier(&header) != header.verifier {
return Err(Error::WrongPassword);
}
Ok(Repository {
backend,
keys,
config: header.config,
cache: None,
written: std::sync::atomic::AtomicU64::new(0),
})
}
/// Hashes one chunk of plaintext and, unless the repository already
/// holds it, seals it for storage. Splitting this CPU-bound half
/// from [`store_prepared`](Repository::store_prepared) lets a
/// pipeline do the sealing on one thread pool and the (possibly
/// slow, remote) write on another.
pub fn prepare_chunk(
&self,
plaintext: &[u8],
) -> Result<(ChunkId, Option<PreparedChunk>), Error> {
let id = ChunkId::compute(self.keys.id_key(), plaintext);
if let Some(cache) = &self.cache
&& cache.holds(id)
{
return Ok((id, None));
}
if self.backend.contains(&ObjectKey::Chunk(id))? {
self.remember(id);
return Ok((id, None));
}
let sealed = crypto::seal(&self.keys, &chunk_aad(id), plaintext)?;
Ok((id, Some(PreparedChunk { id, sealed })))
}
/// Notes that the backend has confirmed a chunk, so the next run
/// need not ask about it.
fn remember(&self, id: ChunkId) {
if let Some(cache) = &self.cache {
cache.remember(id);
}
}
/// What a local cache for this repository must agree with before it
/// may be trusted: which repository it belongs to, and what that
/// repository had pruned when the cache was written.
///
/// Costs one listing of the prune records, which is why it is taken
/// once, when the cache is attached.
pub fn cache_identity(&self) -> Result<CacheIdentity, Error> {
Ok(CacheIdentity {
fingerprint: self.fingerprint(),
epoch: self.epoch()?,
})
}
/// Attaches a local chunk cache, which lives at a path of the
/// caller's choosing — the repository does not know where the
/// machine keeps such things.
///
/// A cache belongs to exactly one repository: use
/// [`Repository::fingerprint`] in its name. Handing a repository the
/// cache of another one would have it skip storing chunks this one
/// does not hold.
pub fn attach_chunk_cache(&mut self, cache: ChunkCache) {
self.cache = Some(cache);
}
/// The attached cache, if any.
pub fn chunk_cache(&self) -> Option<&ChunkCache> {
self.cache.as_ref()
}
/// A stable name for this repository-and-passphrase pair, for
/// naming per-repository local files. Derived from the keys, so it
/// reveals nothing about the contents and cannot collide with
/// another repository's.
pub fn fingerprint(&self) -> [u8; 16] {
let digest = crypto::keyed_digest(self.keys.id_key(), &[b"repository-fingerprint"]);
digest[..16].try_into().expect("digest is 32 bytes")
}
/// Writes a prepared chunk to the backend. Idempotent, like every
/// repository write.
pub fn store_prepared(&self, chunk: &PreparedChunk) -> Result<(), Error> {
self.put(&ObjectKey::Chunk(chunk.id), &chunk.sealed)?;
self.remember(chunk.id);
Ok(())
}
/// Stores one chunk of plaintext, returning its id and whether it was
/// new. Content already in the repository is not written again, which
/// is both the deduplication mechanism and what makes redoing work
/// after a resume harmless.
pub fn store_chunk(&self, plaintext: &[u8]) -> Result<(ChunkId, bool), Error> {
match self.prepare_chunk(plaintext)? {
(id, Some(chunk)) => {
self.store_prepared(&chunk)?;
Ok((id, true))
}
(id, None) => Ok((id, false)),
}
}
/// Fetches and decrypts a chunk, verifying that its content still
/// matches its id.
pub fn load_chunk(&self, id: ChunkId) -> Result<Vec<u8>, Error> {
let sealed = self
.backend
.get(&ObjectKey::Chunk(id))?
.ok_or(Error::MissingChunk(id))?;
let plaintext = crypto::open(&self.keys, &chunk_aad(id), &sealed)?;
if ChunkId::compute(self.keys.id_key(), &plaintext) != id {
return Err(Error::CorruptChunk(id));
}
Ok(plaintext)
}
/// True if the repository already holds the chunk with this id.
pub fn contains_chunk(&self, id: ChunkId) -> Result<bool, Error> {
self.backend.contains(&ObjectKey::Chunk(id))
}
/// Splits `source` into content-defined chunks using the repository's
/// fixed chunking parameters.
pub fn chunk_stream<'r>(
&'r self,
source: impl Read + 'r,
) -> impl Iterator<Item = Result<Vec<u8>, Error>> + 'r {
chunker::chunks(source, &self.config)
}
/// Serializes a stream of manifest entries, chunks it, and stores the
/// chunks, returning the chunk list for a [`Snapshot`] record.
pub fn store_manifest(
&self,
entries: impl IntoIterator<Item = Entry>,
) -> Result<Vec<ChunkId>, Error> {
let stream = manifest::EntryStream::new(entries.into_iter());
let mut ids = Vec::new();
for chunk in chunker::chunks(stream, &self.config) {
let (id, _) = self.store_chunk(&chunk?)?;
ids.push(id);
}
Ok(ids)
}
/// Iterates over the entries of a stored manifest, fetching its
/// chunks on demand.
pub fn manifest_entries(&self, manifest: Vec<ChunkId>) -> ManifestEntries<'_> {
ManifestEntries::new(self, manifest)
}
/// Derives the snapshot id for a backup of `root` whose manifest is
/// `manifest` and which covers what `coverage` says. The id is a
/// keyed digest rather than a random value, so a backup that is
/// redone after an interruption converges on the same snapshot
/// object instead of creating a duplicate.
///
/// Coverage belongs in the derivation because objects are
/// write-once: a partial snapshot and the complete one that
/// supersedes it can name exactly the same manifest — the last
/// checkpoint of a run having drained every entry — and two records
/// making different claims must not be one object, or the second
/// one would silently not be written.
pub fn snapshot_id(
&self,
root: &std::path::Path,
manifest: &[ChunkId],
coverage: Coverage,
) -> SnapshotId {
let mut parts: Vec<&[u8]> = vec![
b"snapshot-id",
coverage.tag(),
root.as_os_str().as_encoded_bytes(),
];
parts.extend(manifest.iter().map(|id| id.as_bytes().as_slice()));
let digest = crypto::keyed_digest(self.keys.id_key(), &parts);
SnapshotId::from_bytes(digest[..16].try_into().expect("digest is 32 bytes"))
}
/// Derives the id for a snapshot rewritten from another over a
/// narrowed manifest, as an excision makes.
///
/// What it is derived from is the snapshot it replaces, rather than
/// what it now holds. Two backups of one tree can differ only in
/// what an excision took out of them, and snapshots that end up
/// sharing a manifest must stay separate records rather than
/// collapsing into one write-once object and taking each other's
/// place. Deriving from the original also keeps the useful half of
/// content addressing: rewriting the same snapshot the same way
/// lands on the same id, so an interrupted excision converges
/// instead of leaving duplicates behind.
pub fn rewritten_snapshot_id(&self, original: &Snapshot, manifest: &[ChunkId]) -> SnapshotId {
let mut parts: Vec<&[u8]> = vec![b"rewritten-snapshot-id", original.id.as_bytes()];
parts.extend(manifest.iter().map(|id| id.as_bytes().as_slice()));
let digest = crypto::keyed_digest(self.keys.id_key(), &parts);
SnapshotId::from_bytes(digest[..16].try_into().expect("digest is 32 bytes"))
}
/// Stores a snapshot record. This is the final step of a backup: the
/// snapshot only becomes visible once fully written.
pub fn store_snapshot(&self, snapshot: &Snapshot) -> Result<(), Error> {
let mut plaintext = Vec::new();
ciborium::into_writer(snapshot, &mut plaintext)?;
let sealed = crypto::seal(&self.keys, &snapshot_aad(snapshot.id), &plaintext)?;
self.put(&ObjectKey::Snapshot(snapshot.id), &sealed)?;
Ok(())
}
/// Fetches and decrypts a snapshot record.
pub fn load_snapshot(&self, id: SnapshotId) -> Result<Snapshot, Error> {
let sealed = self
.backend
.get(&ObjectKey::Snapshot(id))?
.ok_or(Error::MissingSnapshot(id))?;
let plaintext = crypto::open(&self.keys, &snapshot_aad(id), &sealed)?;
Ok(ciborium::from_reader(plaintext.as_slice())?)
}
/// Removes a snapshot record, leaving its chunks alone.
///
/// Used to retire the partial snapshots a suspended backup
/// publishes, once a newer one or the finished backup has replaced
/// them. Deleting an absent record is not an error, so a crash
/// between publishing and recording the fact costs nothing.
pub fn delete_snapshot(&self, id: SnapshotId) -> Result<(), Error> {
self.backend.delete(&ObjectKey::Snapshot(id))
}
/// Records that a prune has happened, which is what tells every
/// local chunk cache that its contents may no longer be true.
///
/// Written *before* any chunk is swept: a prune interrupted halfway
/// must still invalidate the caches, or one of them could go on
/// vouching for a chunk that is already gone.
pub fn record_prune(&self, record: &PruneRecord) -> Result<(), Error> {
let mut plaintext = Vec::new();
ciborium::into_writer(record, &mut plaintext)?;
let sealed = crypto::seal(&self.keys, &prune_aad(record.id), &plaintext)?;
self.put(&ObjectKey::Prune(record.id), &sealed)?;
Ok(())
}
/// Every prune this repository has recorded, newest last.
pub fn prunes(&self) -> Result<Vec<PruneRecord>, Error> {
let mut ids = Vec::new();
self.backend.list(ObjectKind::Prune, &mut |key| {
if let ObjectKey::Prune(id) = key {
ids.push(id);
}
Ok(())
})?;
let mut records = Vec::new();
for id in ids {
// A record deleted between listing and reading is simply not
// part of the history any more
let Some(sealed) = self.backend.get(&ObjectKey::Prune(id))? else {
continue;
};
let plaintext = crypto::open(&self.keys, &prune_aad(id), &sealed)?;
records.push(ciborium::from_reader(plaintext.as_slice())?);
}
records.sort_by_key(|record: &PruneRecord| record.started);
Ok(records)
}
/// A digest of everything ever removed from this repository.
///
/// Two readings that agree mean nothing has been swept in between,
/// which is precisely what a local chunk cache needs to know before
/// it may be trusted. Derived from the set of prune records rather
/// than a counter, so it needs no ordering, no read-modify-write,
/// and nothing but the write-once puts every backend offers.
pub fn epoch(&self) -> Result<[u8; 16], Error> {
let mut ids = Vec::new();
self.backend.list(ObjectKind::Prune, &mut |key| {
if let ObjectKey::Prune(id) = key {
ids.push(*id.as_bytes());
}
Ok(())
})?;
ids.sort_unstable();
let parts: Vec<&[u8]> = ids.iter().map(|id| id.as_slice()).collect();
let digest = crypto::keyed_digest(self.keys.id_key(), &parts);
Ok(digest[..16].try_into().expect("digest is 32 bytes"))
}
/// Claims the repository for a running backup, which holds chunks
/// that no snapshot names yet.
pub fn take_lock(&self, lock: &Lock) -> Result<(), Error> {
let mut plaintext = Vec::new();
ciborium::into_writer(lock, &mut plaintext)?;
let sealed = crypto::seal(&self.keys, &lock_aad(lock.id), &plaintext)?;
self.put(&ObjectKey::Lock(lock.id), &sealed)?;
Ok(())
}
/// Gives a lock back. Idempotent, like every delete.
pub fn release_lock(&self, id: LockId) -> Result<(), Error> {
self.backend.delete(&ObjectKey::Lock(id))
}
/// Every lock currently held on the repository.
pub fn locks(&self) -> Result<Vec<Lock>, Error> {
let mut ids = Vec::new();
self.backend.list(ObjectKind::Lock, &mut |key| {
if let ObjectKey::Lock(id) = key {
ids.push(id);
}
Ok(())
})?;
let mut locks = Vec::new();
for id in ids {
// A lock released while we were listing is simply gone
let Some(sealed) = self.backend.get(&ObjectKey::Lock(id))? else {
continue;
};
let plaintext = crypto::open(&self.keys, &lock_aad(id), &sealed)?;
locks.push(ciborium::from_reader(plaintext.as_slice())?);
}
Ok(locks)
}
/// Removes a chunk. Only [`crate::prune`]-style sweeping should call
/// this, and only for chunks no snapshot references.
pub fn delete_chunk(&self, id: ChunkId) -> Result<(), Error> {
self.backend.delete(&ObjectKey::Chunk(id))
}
/// The ids of every chunk the repository holds.
pub fn chunks(&self) -> Result<Vec<ChunkId>, Error> {
let mut ids = Vec::new();
self.backend.list(ObjectKind::Chunk, &mut |key| {
if let ObjectKey::Chunk(id) = key {
ids.push(id);
}
Ok(())
})?;
Ok(ids)
}
/// The ids of every snapshot in the repository, in no particular
/// order.
pub fn snapshots(&self) -> Result<Vec<SnapshotId>, Error> {
let mut ids = Vec::new();
self.backend.list(ObjectKind::Snapshot, &mut |key| {
if let ObjectKey::Snapshot(id) = key {
ids.push(id);
}
Ok(())
})?;
Ok(ids)
}
/// Direct access to the underlying backend, for maintenance
/// operations that work below the repository abstraction.
pub fn backend(&self) -> &dyn Backend {
&*self.backend
}
/// Sealed bytes this handle has handed to the backend since it was
/// opened.
///
/// An upper bound on what it added to the store, and the cheap way
/// to know how a store has grown under a run without asking the
/// store again: objects are write-once, so a put of something
/// already there is counted here and adds nothing over there.
pub fn bytes_written(&self) -> u64 {
self.written.load(std::sync::atomic::Ordering::Acquire)
}
/// Writes an object, counting what it hands over. Every write a
/// repository makes goes through here, so that count is the whole
/// of what this handle has put into the store.
fn put(&self, key: &ObjectKey, sealed: &[u8]) -> Result<(), Error> {
self.backend.put(key, sealed)?;
self.written
.fetch_add(sealed.len() as u64, std::sync::atomic::Ordering::AcqRel);
Ok(())
}
}
fn chunk_aad(id: ChunkId) -> Vec<u8> {
let mut aad = b"chunk:".to_vec();
aad.extend_from_slice(id.as_bytes());
aad
}
fn snapshot_aad(id: SnapshotId) -> Vec<u8> {
let mut aad = b"snapshot:".to_vec();
aad.extend_from_slice(id.as_bytes());
aad
}
fn prune_aad(id: PruneId) -> Vec<u8> {
let mut aad = b"prune:".to_vec();
aad.extend_from_slice(id.as_bytes());
aad
}
fn lock_aad(id: LockId) -> Vec<u8> {
let mut aad = b"lock:".to_vec();
aad.extend_from_slice(id.as_bytes());
aad
}