// SPDX-License-Identifier: AGPL-3.0-or-later

//! This crate contains the core structs and traits used by Blocktree.
//! The [crypto] module contains all of the cryptographic primitives used to implement this system.
//! Most of the functionality for accessing blocks is defined by the [Block] trait. The
//! [BlockMeta] struct is the type which implements block metadata.

pub mod accessor;
pub mod atomic_file;
mod block_path;
pub mod buf_reader;
pub mod collections;
pub mod config_helpers;
pub mod crypto;
pub mod drop_trigger;
pub mod error;
pub mod log;
mod readcap_dict;
pub mod sectored_buf;
#[cfg(test)]
mod test_helpers;
mod trailered;
use readcap_dict::ReadcapDict;

#[macro_use]
extern crate static_assertions;

#[macro_use]
#[cfg(test)]
extern crate lazy_static;

use ::log::error;
use btserde::{read_from, write_to};
use fuse_backend_rs::abi::fuse_abi::{stat64, Attr};
use positioned_io::{ReadAt, Size, WriteAt};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use serde_big_array::BigArray;
use std::{
    collections::{btree_map, BTreeMap},
    convert::{Infallible, TryFrom},
    fmt::{self, Display, Formatter},
    hash::Hash as Hashable,
    io::{self, Read, Seek, SeekFrom, Write},
    net::IpAddr,
    ops::{Add, Deref, Sub},
    os::unix::prelude::MetadataExt,
    time::{Duration, SystemTime},
};
use strum_macros::{Display, EnumDiscriminants, FromRepr};

use accessor::Accessor;
pub use block_path::{BlockPath, BlockPathError, BlockPathGen, BlockPathRef, RelBlockPath};
use crypto::{
    AsymKeyPub, Ciphertext, ConcretePub, Creds, CredsPub, Decrypter, DecrypterExt, EncrypterExt,
    HashKind, MerkleStream, SecretStream, Sign, Signature, Signer, SignerExt, SymKey, SymKeyKind,
    VarHash,
};
use error::{BoxInIoErr, BtErr};
pub use error::{Error, Result};
use trailered::Trailered;

#[derive(Debug)]
pub enum BlockError {
    MissingWritecap,
    IncorrectSize { expected: usize, actual: usize },
    NoBlockKey,
    NoBlockPath,
    UnknownSize,
    ProcRecNotIssued,
    ProcRecRevoked,
    NoInheritedKey,
}

impl Display for BlockError {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        match self {
            BlockError::MissingWritecap => write!(f, "missing writecap"),
            BlockError::IncorrectSize { expected, actual } => {
                write!(f, "incorrect size {actual}, expected {expected}")
            }
            BlockError::NoBlockKey => write!(f, "no block key is present"),
            BlockError::NoBlockPath => write!(f, "no block path was specified"),
            BlockError::UnknownSize => write!(f, "size could not be determined"),
            BlockError::ProcRecNotIssued => {
                write!(f, "a process record was requested by not yet issued")
            }
            BlockError::ProcRecRevoked => write!(f, "this process record has been revoked"),
            BlockError::NoInheritedKey => write!(f, "block metadata has no inherited key"),
        }
    }
}

impl std::error::Error for BlockError {}

// This assertion ensures that conversions from `usize` to `u64` will not cause truncation. This
// prevents this code from compiling for 128 bit platforms, but that's not really a concern for the
// foreseeable future.
// If this assumption is ever to be removed, you'll need to evaluate every occurrence of `as u64`.
const_assert!(::std::mem::size_of::<usize>() <= ::std::mem::size_of::<u64>());

/// The expected size of the IO blocks for the file system. This is used to calculate
/// [SECTOR_SZ_DEFAULT].
pub const EXPECTED_IO_BLOCK: usize = 4096;
/// The number of IO blocks in each sector.
pub const IO_BLOCKS_PER_SECTOR: usize = 256;
/// The default sector size to use for new blocks.
pub const SECTOR_SZ_DEFAULT: usize = EXPECTED_IO_BLOCK * IO_BLOCKS_PER_SECTOR;
/// `SECTOR_SZ_DEFAULT` converted to a `u64`.
pub const SECTOR_U64_DEFAULT: u64 = SECTOR_SZ_DEFAULT as u64;

pub trait MetaReader: AsRef<BlockMeta> + Size {
    fn meta(&self) -> &BlockMeta {
        self.as_ref()
    }

    fn meta_body(&self) -> &BlockMetaBody {
        self.meta().body()
    }
}

impl<T: AsRef<BlockMeta> + Size + ?Sized> MetaReader for T {}

/// Trait for accessing the metadata associated with a block.
pub trait MetaAccess: AsMut<BlockMeta> + MetaReader {
    fn mut_meta(&mut self) -> &mut BlockMeta {
        self.as_mut()
    }

    fn mut_meta_body(&mut self) -> &mut BlockMetaBody {
        self.mut_meta().mut_body()
    }
}

impl<T: AsMut<BlockMeta> + MetaReader + ?Sized> MetaAccess for T {}

pub trait FlushMeta {
    /// Flushes metadata to persistent storage.
    fn flush_meta(&mut self) -> Result<()>;
}

impl<T: FlushMeta + ?Sized> FlushMeta for &mut T {
    fn flush_meta(&mut self) -> Result<()> {
        (*self).flush_meta()
    }
}

/// ### THE BLOCK TRAIT
///
/// Trait for types which provide read and write access to blocks.
pub trait Block: ReadAt + WriteAt + MetaAccess + Sectored + FlushMeta {}

impl<T: ReadAt + WriteAt + MetaAccess + Sectored + FlushMeta + ?Sized> Block for T {}

/// Deserializes an instance of the given type from the given block.
fn read_from_block<T: DeserializeOwned, B: BlockReader + ?Sized>(block: &mut B) -> Result<T> {
    block.rewind()?;
    let mut block = block;
    let dir: T = read_from(&mut block)?;
    Ok(dir)
}

pub trait BlockReader: Read + Seek + AsRef<BlockMeta> + Size + Sectored {
    fn read_dir(&mut self) -> Result<Directory> {
        read_from_block::<Directory, _>(self)
    }

    fn read_proc_rec(&mut self) -> Result<ProcRec> {
        read_from_block::<ProcRec, _>(self)
    }
}

impl<T: Read + Seek + AsRef<BlockMeta> + Size + Sectored + ?Sized> BlockReader for T {}

fn write_to_block<T: Serialize, B: BlockAccessor + ?Sized>(block: &mut B, value: &T) -> Result<()> {
    block.rewind()?;
    let mut sich = block;
    write_to(value, &mut sich)?;
    sich.flush()?;
    Ok(())
}

pub trait BlockAccessor: BlockReader + Write + MetaAccess {
    fn write_dir(&mut self, dir: &Directory) -> Result<()> {
        write_to_block(self, dir)
    }

    fn write_proc_rec(&mut self, proc_rec: &ProcRec) -> Result<()> {
        write_to_block(self, proc_rec)
    }
}

impl<T: Read + Write + Seek + MetaAccess + Sectored + ?Sized> BlockAccessor for T {}

// A trait for streams which only allow reads and writes in fixed sized units called sectors.
pub trait Sectored {
    /// Returns the size of the sector for this stream.
    fn sector_sz(&self) -> usize;

    /// Returns the sector size as a `u64`.
    fn sector_sz64(&self) -> u64 {
        // This is guaranteed not to truncate thanks to the `const_assert!` above
        // `SECTOR_SZ_DEFAULT`.
        self.sector_sz() as u64
    }

    /// Returns `Err(Error::IncorrectSize)` if the given size is not equal to the sector size.
    fn assert_sector_sz(&self, actual: usize) -> Result<()> {
        let expected = self.sector_sz();
        if expected != actual {
            Err(bterr!(BlockError::IncorrectSize { expected, actual }))
        } else {
            Ok(())
        }
    }

    /// Returns `Err(Error::IncorrectSize)` if the given size is less than the sector size.
    fn assert_at_least_sector_sz(&self, actual: usize) -> Result<()> {
        let expected = self.sector_sz();
        if actual < expected {
            Err(bterr!(BlockError::IncorrectSize { expected, actual }))
        } else {
            Ok(())
        }
    }

    /// Returns the offset (in bytes) from the beginning of this stream that the given 0-based
    /// sector index corresponds to.
    fn offset_at(&self, index: u64) -> u64 {
        index * self.sector_sz64()
    }
}

impl<T: Sectored + ?Sized> Sectored for &T {
    fn sector_sz(&self) -> usize {
        (**self).sector_sz()
    }
}

impl<T: Sectored + ?Sized> Sectored for &mut T {
    fn sector_sz(&self) -> usize {
        (**self).sector_sz()
    }
}

impl Sectored for ::std::fs::File {
    fn sector_sz(&self) -> usize {
        self.metadata()
            .map(|e| {
                let blksize: usize = e.blksize().try_into().bterr().unwrap();
                blksize * IO_BLOCKS_PER_SECTOR
            })
            .unwrap_or(SECTOR_SZ_DEFAULT)
    }
}

impl<T: Sectored + Size> Sectored for Cursor<T> {
    fn sector_sz(&self) -> usize {
        self.cursor.get_ref().sector_sz()
    }
}

/// The `Read` trait requires the caller to supply the buffer to be read into. This trait is its
/// dual in the sense that the trait implementor is expected to supply its own buffer, which a
/// `Write` instance is given. This can be used to avoid copying in cases where the trait
/// implementor is already buffering data, so it can give a reference to this buffer to the caller
/// (for example in `SectoredBuf`)
pub trait ReadDual: Read {
    fn read_into<W: Write>(&mut self, write: W, count: usize) -> io::Result<usize>;
}

pub trait WriteDual: Write {
    fn write_from<R: Read>(&mut self, read: R, count: usize) -> io::Result<usize>;
}

/// Trait for types which can be extended with zero byes.
pub trait ZeroExtendable {
    /// Extends this stream with the given number of zero bytes. The position of the stream must
    /// be unchanged when this method returns successfully. The state of the stream in the case of
    /// an error is undefined.
    fn zero_extend(&mut self, num_zeros: u64) -> io::Result<()>;
}

/// Trait for streams which can efficiently and infallibly return their current position.
pub trait Positioned {
    /// Returns the position of this stream (byte offset relative to the beginning).
    fn pos(&self) -> usize;
}

impl<T: Positioned> Positioned for &T {
    fn pos(&self) -> usize {
        (**self).pos()
    }
}

impl<T: Positioned> Positioned for &mut T {
    fn pos(&self) -> usize {
        (**self).pos()
    }
}

pub trait TrySeek {
    /// Attempts to seek to the given offset from the start of the stream.
    fn try_seek(&mut self, seek_from: SeekFrom) -> io::Result<()>;
}

pub trait SizeExt: Size {
    fn size_or_err(&self) -> Result<u64> {
        self.size()?.ok_or_else(|| bterr!(BlockError::UnknownSize))
    }
}

impl<T: Size> SizeExt for T {}

/// A version of the `WriteAt` trait, which allows integrity information to be supplied when
/// flushing.
pub trait WriteInteg: WriteAt {
    fn flush_integ(&mut self, integrity: &[u8]) -> io::Result<()>;
}

pub trait Decompose<T> {
    fn into_inner(self) -> T;
}

pub trait Split<L, R> {
    fn split(self) -> (L, R);
    fn combine(left: L, right: R) -> Self;
}

pub trait TryCompose<T, U: Decompose<T>> {
    type Error;
    fn try_compose(self, inner: T) -> std::result::Result<U, Self::Error>;
}

trait Compose<T, U> {
    fn compose(self, inner: T) -> U;
}

impl<T, U: Decompose<T>, S: TryCompose<T, U, Error = Infallible>> Compose<T, U> for S {
    fn compose(self, inner: T) -> U {
        let result = self.try_compose(inner);
        // Safety: Infallible has no values, so `result` must be `Ok`.
        unsafe { result.unwrap_unchecked() }
    }
}
impl AsRef<BlockMeta> for &BlockMeta {
    fn as_ref(&self) -> &BlockMeta {
        self
    }
}

impl AsRef<BlockMeta> for &mut BlockMeta {
    fn as_ref(&self) -> &BlockMeta {
        self
    }
}

impl AsMut<BlockMeta> for &mut BlockMeta {
    fn as_mut(&mut self) -> &mut BlockMeta {
        self
    }
}

/// A wrapper around [positioned_io::SizeCursor] which implements [Split].
#[derive(Debug)]
pub struct Cursor<T: Size> {
    cursor: positioned_io::SizeCursor<T>,
}

impl<T: Size> Cursor<T> {
    /// Creates a new [Cursor] containing the given inner stream.
    pub fn new(inner: T) -> Cursor<T> {
        Self {
            cursor: positioned_io::SizeCursor::new(inner),
        }
    }

    /// Create a new [Cursor] containing the given inner stream and with the given position.
    pub fn new_pos(inner: T, pos: u64) -> Cursor<T> {
        Self {
            cursor: positioned_io::SizeCursor::new_pos(inner, pos),
        }
    }

    /// Returns a reference to the inner stream.
    pub fn get_ref(&self) -> &T {
        self.cursor.get_ref()
    }

    /// Returns a mutable reference to the inner stream.
    pub fn get_mut(&mut self) -> &mut T {
        self.cursor.get_mut()
    }
}

impl<T: ReadAt + Size> Read for Cursor<T> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.cursor.read(buf)
    }
}

impl<T: WriteAt + Size> Write for Cursor<T> {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.cursor.write(buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.cursor.flush()
    }
}

impl<T: Size> Seek for Cursor<T> {
    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
        self.cursor.seek(pos)
    }
}

impl<T: Size> Decompose<T> for Cursor<T> {
    fn into_inner(self) -> T {
        self.cursor.into_inner()
    }
}

impl<U, T: AsRef<U> + Size> AsRef<U> for Cursor<T> {
    fn as_ref(&self) -> &U {
        self.cursor.get_ref().as_ref()
    }
}

impl<U, T: AsMut<U> + Size> AsMut<U> for Cursor<T> {
    fn as_mut(&mut self) -> &mut U {
        self.cursor.get_mut().as_mut()
    }
}

impl<T: Size> Size for Cursor<T> {
    fn size(&self) -> io::Result<Option<u64>> {
        self.cursor.get_ref().size()
    }
}

/// A zero length slice of bytes.
pub const EMPTY_SLICE: &[u8] = &[0u8; 0];

impl<T: Size> Split<Cursor<&'static [u8]>, T> for Cursor<T> {
    fn split(self) -> (Cursor<&'static [u8]>, T) {
        let pos = self.cursor.position();
        (Cursor::new_pos(EMPTY_SLICE, pos), self.cursor.into_inner())
    }

    fn combine(left: Cursor<&'static [u8]>, right: T) -> Self {
        let pos = left.cursor.position();
        Self::new_pos(right, pos)
    }
}

/// Extensions to the `Read` trait.
trait ReadExt: Read {
    /// Reads repeatedly until one of the following occur:
    ///  1. The given buffer is full.
    ///  2. A call to `read` returns 0.
    ///  3. A call to `read` returns an error.
    /// The number of bytes read is returned. If an error is returned, then no bytes were read.
    fn fill_buf(&mut self, mut dest: &mut [u8]) -> io::Result<usize> {
        let dest_len_start = dest.len();
        while !dest.is_empty() {
            let byte_ct = match self.read(dest) {
                Ok(byte_ct) => byte_ct,
                Err(err) => {
                    if dest_len_start == dest.len() {
                        return Err(err);
                    } else {
                        // We're not allowed to return an error since we've already read from self.
                        error!("an error occurred in fill_buf: {}", err);
                        break;
                    }
                }
            };
            if 0 == byte_ct {
                break;
            }
            dest = &mut dest[byte_ct..];
        }
        Ok(dest_len_start - dest.len())
    }
}

impl<T: Read> ReadExt for T {}

pub trait SeekFromExt {
    /// Returns the absolute position (offset from the start) this `SeekFrom` refers to.
    /// `curr` is called in the case this `SeekFrom` is `Current`, and is expected to return the
    /// current position.
    /// `end` is called in the case this `SeekFrom` is `End`, and is expected  to return the
    /// the position of the end.
    fn abs<F, G>(&self, curr: F, end: G) -> Result<u64>
    where
        F: FnOnce() -> Result<u64>,
        G: FnOnce() -> Result<u64>;

    /// Like [SeekFromExt::abs] except that an error is always returned when [SeekFrom::End] is
    /// given.
    fn abs_no_end<F>(&self, curr: F) -> Result<u64>
    where
        F: FnOnce() -> Result<u64>,
    {
        self.abs(curr, || Err(bterr!("SeekFrom::End is not supported")))
    }

    /// Converts a C-style `(whence, offset)` pair into a [SeekFrom] enum value.
    /// See the POSIX man page of `lseek` for more details.
    fn whence_offset(whence: u32, offset: u64) -> io::Result<SeekFrom> {
        let whence = whence as i32;
        match whence {
            libc::SEEK_SET => Ok(SeekFrom::Start(offset)),
            libc::SEEK_CUR => Ok(SeekFrom::Current(offset as i64)),
            libc::SEEK_END => Ok(SeekFrom::End(offset as i64)),
            _ => Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "`whence` was not one of `libc::{SEEK_SET, SEEK_CUR, SEEK_END}`",
            )),
        }
    }
}

impl SeekFromExt for SeekFrom {
    fn abs<F, G>(&self, curr: F, end: G) -> Result<u64>
    where
        F: FnOnce() -> Result<u64>,
        G: FnOnce() -> Result<u64>,
    {
        match self {
            SeekFrom::Start(start) => Ok(*start),
            SeekFrom::Current(from_curr) => {
                let curr = curr()?;
                Ok(curr.wrapping_add_signed(*from_curr))
            }
            SeekFrom::End(from_end) => {
                let end = end()?;
                Ok(end.wrapping_add_signed(*from_end))
            }
        }
    }
}

/// The type used to identify blocks.
pub type Inode = u64;

/// A unique identifier for a block.
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone, Hash)]
pub struct BlockId {
    /// The identifier of the server generation the block is stored in.
    pub generation: u64,
    /// The identifier of the inode on the server generation containing the block.
    pub inode: Inode,
}

impl BlockId {
    /// Creates a new [BlockId] with the given values.
    pub fn new(generation: u64, inode: Inode) -> BlockId {
        BlockId { generation, inode }
    }
}

impl Default for BlockId {
    fn default() -> Self {
        BlockId::new(0, 0)
    }
}

/// Metadata that is encrypted.
#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone, Hash)]
pub struct BlockMetaSecrets {
    /// The identifier for the block these secrets are for.
    pub block_id: BlockId,
    /// Mode of file.
    pub mode: u32,
    /// Owner UID of file.
    pub uid: u32,
    /// Owner GID of file.
    pub gid: u32,
    /// Last access time.
    pub atime: Epoch,
    /// Last data modification.
    pub mtime: Epoch,
    /// Last status change.
    pub ctime: Epoch,
    /// Size of the file in bytes.
    pub size: u64,
    /// Number of hard links to the file.
    pub nlink: u32,
    /// The sector size used by the block.
    pub sect_sz: u64,
    /// User controlled metadata.
    pub tags: BTreeMap<String, Vec<u8>>,
}

impl BlockMetaSecrets {
    pub fn new() -> BlockMetaSecrets {
        Self {
            block_id: BlockId::default(),
            mode: 0,
            uid: 0,
            gid: 0,
            atime: Epoch::default(),
            mtime: Epoch::default(),
            ctime: Epoch::default(),
            size: 0,
            nlink: 0,
            sect_sz: SECTOR_U64_DEFAULT,
            tags: BTreeMap::new(),
        }
    }

    pub fn attr(&self) -> Result<Attr> {
        Ok(Attr {
            ino: self.block_id.inode,
            size: self.size,
            atime: self.atime.value(),
            mtime: self.mtime.value(),
            ctime: self.ctime.value(),
            atimensec: 0,
            mtimensec: 0,
            ctimensec: 0,
            mode: self.mode,
            nlink: self.nlink,
            uid: self.uid,
            gid: self.gid,
            rdev: 0,
            blksize: self
                .sect_sz
                .try_into()
                .map_err(|_| bterr!("BlockMetaSecrets::sect_sz could not be converted to a u32"))?,
            blocks: self.sectors(),
            flags: 0,
        })
    }

    pub fn stat(&self) -> Result<stat64> {
        self.attr().map(|e| e.into())
    }

    /// Returns the number of sectors occupied by the block's data.
    pub fn sectors(&self) -> u64 {
        if self.size % self.sect_sz == 0 {
            self.size / self.sect_sz
        } else {
            self.size / self.sect_sz + 1
        }
    }

    pub fn block_id(&self) -> &BlockId {
        &self.block_id
    }

    pub fn sector_sz(&self) -> u64 {
        self.sect_sz
    }
}

impl Default for BlockMetaSecrets {
    fn default() -> Self {
        Self::new()
    }
}

impl AsRef<BlockId> for BlockMetaSecrets {
    fn as_ref(&self) -> &BlockId {
        self.block_id()
    }
}

impl TryFrom<&BlockMetaSecrets> for Attr {
    type Error = crate::Error;

    fn try_from(value: &BlockMetaSecrets) -> Result<Self> {
        value.attr()
    }
}

/// This struct contains all of the metadata fields associated with a block, except for its
/// signature. Since this struct implements `Serialize`, this allows for convenient signature
/// calculations.
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
pub struct BlockMetaBody {
    /// A copy of the block key encrypted using this block's parent's key. If this is None, then
    /// this block is not encrypted.
    inherit: Option<Ciphertext<SymKey>>,
    readcaps: ReadcapDict,
    writecap: Option<Writecap>,
    /// A hash which provides integrity for the contents of the block body.
    integrity: Option<VarHash>,
    /// The public key that corresponds to the private key used to sign these metadata.
    signing_key: AsymKeyPub<Sign>,
    /// Additional metadata which is subject to confidentiality protection.
    secrets: Ciphertext<BlockMetaSecrets>,

    #[serde(skip)]
    /// The path in the blocktree where this block is located. This is initialized in
    /// [BlockStream::new].
    path: BlockPath,
    #[serde(skip)]
    /// The cleartext block key.
    block_key: Option<SymKey>,
    #[serde(skip)]
    /// The clear text secret metadata.
    secrets_struct: Option<BlockMetaSecrets>,
}

impl BlockMetaBody {
    fn new<C: Creds>(creds: C) -> Result<BlockMetaBody> {
        let block_key = SymKey::generate(SymKeyKind::default())?;
        let secrets = BlockMetaSecrets::default();
        let mut body = BlockMetaBody {
            path: BlockPath::default(),
            inherit: None,
            readcaps: ReadcapDict::new()?,
            writecap: creds.writecap().map(|e| e.to_owned()),
            integrity: None,
            signing_key: creds.public_sign().to_owned(),
            secrets: block_key.ser_encrypt(&secrets)?,
            block_key: Some(block_key),
            secrets_struct: Some(secrets),
        };
        body.add_readcap_for(creds)?;
        Ok(body)
    }

    /// Uses the given symmetric key to decrypt the `inherit` field. If this field is `None`, or if
    /// the decryption fails, then an error is returned. If the block key has already been decrypted
    /// then this method does nothing.
    pub fn unlock_block_key_with_parent_key(&mut self, parent_key: SymKey) -> Result<()> {
        if self.block_key.is_some() {
            return Ok(());
        }
        if let Some(ref inherit) = self.inherit {
            self.block_key = Some(parent_key.ser_decrypt(inherit)?);
            Ok(())
        } else {
            Err(BlockError::NoInheritedKey.into())
        }
    }

    /// The given closure is called with the [BlockMetaSecrets] contained in this struct.
    /// Whatever the closure returns is returned by this method (unless an error occurs).
    /// The secrets are decrypted, if needed, and after the closure returns they are updated.
    /// It's best to use the [BlockMetaBody::secrets] method if all you need is read
    /// access.
    pub fn access_secrets<T, F: FnOnce(&mut BlockMetaSecrets) -> Result<T>>(
        &mut self,
        accessor: F,
    ) -> Result<T> {
        self.decrypt_secrets()?;
        let secrets = self.secrets_struct.as_mut().unwrap();
        let output = accessor(secrets)?;
        self.secrets = self
            .block_key
            .as_ref()
            .ok_or(BlockError::NoBlockKey)?
            .ser_encrypt(secrets)?;
        Ok(output)
    }

    /// Decrypts the [BlockMetaSecrets] in this struct using the block key. After this method is
    /// called the [BlockMetaBody::secrets] method can be called to cheaply access the secrets.
    /// This method should be called by [BlockStream] in its new method, so it shouldn't need to be
    /// called after the fact.
    fn decrypt_secrets(&mut self) -> Result<()> {
        if self.secrets_struct.is_none() {
            let block_key = self.block_key()?;
            self.secrets_struct = Some(block_key.ser_decrypt(&self.secrets)?);
        }
        Ok(())
    }

    /// Returns a reference to the secrets in this struct. This is very cheap operation, just
    /// copying a reference.
    pub fn secrets(&self) -> Result<&BlockMetaSecrets> {
        self.secrets_struct
            .as_ref()
            .ok_or_else(|| bterr!("secrets have not been decrypted"))
    }

    /// Returns a reference to the block key.
    pub fn block_key(&self) -> Result<&SymKey> {
        self.block_key
            .as_ref()
            .ok_or_else(|| bterr!(BlockError::NoBlockKey))
    }

    /// Returns the ID of the block this metadata is part of.
    pub fn block_id(&self) -> Result<&BlockId> {
        let secrets = self.secrets()?;
        Ok(secrets.block_id())
    }

    /// Returns a reference to the integrity value stored in this struct. This value represents the
    /// authenticated value of the block's contents.
    pub fn integrity(&self) -> Option<&[u8]> {
        self.integrity.as_ref().map(|hash| hash.as_slice())
    }

    /// Decrypts the readcap for the given [Creds] and uses the resulting plaintext as the block
    /// key.
    pub fn use_readcap_for<C: Creds>(&mut self, creds: &C) -> Result<&SymKey> {
        let block_key = self.readcaps.get(creds)?;
        self.block_key = Some(block_key);
        self.block_key()
    }

    /// Adds a new readcap for the given [Creds].
    pub fn add_readcap_for<C: CredsPub>(&mut self, creds: C) -> Result<()> {
        let block_key = self
            .block_key
            .as_ref()
            .ok_or_else(|| bterr!(BlockError::NoBlockKey))?;
        self.readcaps.set(creds, block_key)
    }

    /// Returns a reference to the [BlockPath] of the block this metadata is for.
    pub fn path(&self) -> &BlockPath {
        &self.path
    }

    /// Returns a mutable reference to the [BlockPath] of the block this metadata is for.
    pub fn path_mut(&mut self) -> &mut BlockPath {
        &mut self.path
    }
}

/// Signed metadata associated with a block.
#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
pub struct BlockMeta {
    body: BlockMetaBody,
    sig: Signature,
}

impl BlockMeta {
    /// Creates a new [BlockMeta] struct which is authenticated by the given credentials.
    pub fn new<C: Creds>(creds: &C) -> Result<BlockMeta> {
        let body = BlockMetaBody::new(creds)?;
        let sig = Signature::empty(body.signing_key.scheme());
        Ok(BlockMeta { body, sig })
    }

    /// Returns a reference to the [BlockMetaBody] contained in this struct.
    pub fn body(&self) -> &BlockMetaBody {
        self.as_ref()
    }

    /// Returns a mutable reference to the [BlockMetaBody] contained in this struct.
    pub fn mut_body(&mut self) -> &mut BlockMetaBody {
        self.as_mut()
    }
}

impl AsRef<BlockMetaBody> for BlockMeta {
    fn as_ref(&self) -> &BlockMetaBody {
        &self.body
    }
}

impl AsMut<BlockMetaBody> for BlockMeta {
    fn as_mut(&mut self) -> &mut BlockMetaBody {
        &mut self.body
    }
}

/// A stream which is responsible for managing the metadata associated with a block, including
/// validating and authenticating it. This type shouldn't be instantiated direction, instead a
/// [BlockOpenOptions] struct is used.
pub struct BlockStream<T, C> {
    trailered: Trailered<T, BlockMeta>,
    meta: BlockMeta,
    meta_body_buf: Vec<u8>,
    creds: C,
    sect_sz: usize,
}

impl<T: ReadAt + Sectored + Size, C: Creds> BlockStream<T, C> {
    /// Creates a new [BlockStream] using the given inner stream, [Creds], an optional parent
    /// key, and [BlockPath].
    fn new(
        inner: T,
        creds: C,
        parent_key: Option<SymKey>,
        block_path: BlockPath,
    ) -> Result<BlockStream<T, C>> {
        let (trailered, meta) = Trailered::<_, BlockMeta>::new(inner)?;
        let meta = match meta {
            Some(mut meta) => {
                meta.assert_valid(&block_path)?;
                meta.body.path = block_path;
                if let Some(parent_key) = parent_key {
                    meta.body.unlock_block_key_with_parent_key(parent_key)?;
                } else {
                    meta.body.use_readcap_for(&creds)?;
                }
                // We need to use the writecap and signing_key provided by the current credentials.
                meta.body.writecap = Some(
                    creds
                        .writecap()
                        .ok_or(BlockError::MissingWritecap)?
                        .to_owned(),
                );
                meta.body.signing_key = creds.public_sign().to_owned();
                meta.body.decrypt_secrets()?;
                meta
            }
            None => {
                let mut meta = BlockMeta::new(&creds)?;
                meta.body.path = block_path;
                if let Some(parent_key) = parent_key {
                    meta.body.inherit = Some(parent_key.ser_encrypt(meta.body.block_key()?)?);
                } else {
                    meta.body.add_readcap_for(&creds)?;
                }
                meta.body.writecap = Some(
                    creds
                        .writecap()
                        .ok_or(BlockError::MissingWritecap)?
                        .to_owned(),
                );
                meta.body.access_secrets(|secrets| {
                    secrets.sect_sz = trailered.sector_sz64();
                    Ok(())
                })?;
                meta
            }
        };
        let sect_sz = meta.body.secrets()?.sector_sz().try_into().bterr()?;
        Ok(BlockStream {
            trailered,
            meta_body_buf: Vec::new(),
            creds,
            sect_sz,
            meta,
        })
    }
}

impl<T: WriteAt + Size, C: Signer> BlockStream<T, C> {
    fn sign_flush_meta(&mut self) -> io::Result<()> {
        self.meta_body_buf.clear();
        self.meta.sig = self
            .creds
            .ser_sign_into(&self.meta.body, &mut self.meta_body_buf)?;
        self.trailered.flush(&self.meta)
    }
}

impl<T: WriteAt + Size, C: Signer + Principaled + Decrypter> WriteAt for BlockStream<T, C> {
    fn write_at(&mut self, pos: u64, buf: &[u8]) -> io::Result<usize> {
        self.trailered.write_at(pos, buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        Err(io::Error::new(
            io::ErrorKind::Unsupported,
            "BlockStream::flush is not supported. Use BlockStream::flush_integ",
        ))
    }
}

impl<T: WriteAt + Size, C: Signer + Principaled + Decrypter> WriteInteg for BlockStream<T, C> {
    fn flush_integ(&mut self, integrity: &[u8]) -> io::Result<()> {
        let meta_body = &mut self.meta.body;
        let integ = meta_body.integrity.get_or_insert_with(VarHash::default);
        integ.as_mut().copy_from_slice(integrity);
        self.sign_flush_meta()
    }
}

impl<T: ReadAt + Size, C> ReadAt for BlockStream<T, C> {
    fn read_at(&self, pos: u64, buf: &mut [u8]) -> io::Result<usize> {
        self.trailered.read_at(pos, buf)
    }
}

impl<T, C> AsRef<BlockMeta> for BlockStream<T, C> {
    fn as_ref(&self) -> &BlockMeta {
        &self.meta
    }
}

impl<T, C> AsMut<BlockMeta> for BlockStream<T, C> {
    fn as_mut(&mut self) -> &mut BlockMeta {
        &mut self.meta
    }
}

impl<T, C> Sectored for BlockStream<T, C> {
    fn sector_sz(&self) -> usize {
        self.sect_sz
    }
}

impl<T: Size, C> Size for BlockStream<T, C> {
    fn size(&self) -> io::Result<Option<u64>> {
        self.trailered.size()
    }
}

impl<T: WriteAt + Size, C: Signer> FlushMeta for BlockStream<T, C> {
    fn flush_meta(&mut self) -> Result<()> {
        self.sign_flush_meta().map_err(|err| err.into())
    }
}

impl<T, C> Decompose<(T, C)> for BlockStream<T, C> {
    fn into_inner(self) -> (T, C) {
        (self.trailered.into_inner(), self.creds)
    }
}

/// Describes how to open a block, including the inner stream from which the block can be read and
/// the credentials to use to open it.
pub struct BlockOpenOptions<T, C> {
    inner: T,
    creds: C,
    encrypt: bool,
    block_path: Option<BlockPath>,
    parent_key: Option<SymKey>,
}

impl BlockOpenOptions<(), ()> {
    /// Creates a new empty [BlockOpenOptions].
    pub fn new() -> BlockOpenOptions<(), ()> {
        BlockOpenOptions {
            inner: (),
            creds: (),
            encrypt: true,
            block_path: Default::default(),
            parent_key: None,
        }
    }
}

impl<T, C> BlockOpenOptions<T, C> {
    /// Configures the given stream to be used for reading and writing the block.
    pub fn with_inner<U>(self, inner: U) -> BlockOpenOptions<U, C> {
        BlockOpenOptions {
            inner,
            creds: self.creds,
            encrypt: self.encrypt,
            block_path: self.block_path,
            parent_key: self.parent_key,
        }
    }

    /// Configures the given [Creds] to be used to open the block.
    pub fn with_creds<D>(self, creds: D) -> BlockOpenOptions<T, D> {
        BlockOpenOptions {
            inner: self.inner,
            creds,
            encrypt: self.encrypt,
            block_path: self.block_path,
            parent_key: self.parent_key,
        }
    }

    /// Specifies whether or not the block contents will be encrypted.
    pub fn with_encrypt(mut self, encrypt: bool) -> Self {
        self.encrypt = encrypt;
        self
    }

    /// Specifies the path of the block. If the block is new, this will be the path it's created
    /// at. If an existing block is being opened, then this is used for validating the block.
    pub fn with_block_path(mut self, block_path: BlockPath) -> Self {
        self.block_path = Some(block_path);
        self
    }

    /// Configures the given parent key to be used for unlocking the block key.
    pub fn with_parent_key(mut self, parent_key: Option<SymKey>) -> Self {
        self.parent_key = parent_key;
        self
    }
}

/// The composition of a [BlockStream] in a [MerkleStream].
pub type ConcreteBlock<T, C> = MerkleStream<BlockStream<T, C>>;
/// A [ConcreteBlock] which stores its data in a file in the filesystem.
pub type FileBlock<C> = ConcreteBlock<std::fs::File, C>;

impl<T: ReadAt + WriteAt + Size + Sectored + 'static, C: Creds + 'static> BlockOpenOptions<T, C> {
    /// Opens the block without wrapping it in an [Accessor].
    pub fn open_bare(self) -> Result<ConcreteBlock<T, C>> {
        let block_path = self.block_path.ok_or(BlockError::NoBlockPath)?;
        let stream = BlockStream::new(self.inner, self.creds, self.parent_key, block_path)?;
        let mut stream = MerkleStream::new(stream)?;
        stream.assert_root_integrity()?;
        Ok(stream)
    }

    /// Opens the block and wraps it in an [Accessor] before returning it.
    pub fn open(self) -> Result<Accessor<ConcreteBlock<T, C>>> {
        let stream = self.open_bare()?;
        let stream = Accessor::new(stream)?;
        Ok(stream)
    }
}

impl Default for BlockOpenOptions<(), ()> {
    fn default() -> Self {
        Self::new()
    }
}

/// An envelopment of a key, which is tagged with the principal who the key is meant for.
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
pub struct Readcap {
    /// The principal this `Readcap` was issued to.
    issued_to: Principal,
    /// An encipherment of a block key using the public key of the principal.
    key: Ciphertext<SymKey>,
}

impl Readcap {
    pub fn new(issued_to: VarHash, key: Ciphertext<SymKey>) -> Readcap {
        Readcap {
            issued_to: Principal(issued_to),
            key,
        }
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
pub struct WritecapBody {
    /// The principal this `Writecap` was issued to.
    issued_to: Principal,
    /// The path where this write caps's validity begins.
    path: BlockPath,
    /// The point in time after which this write cap is no longer valid.
    expires: Epoch,
    /// The public key used to sign this write cap.
    signing_key: AsymKeyPub<Sign>,
}

/// Verifies that a principal is authorized to write blocks in a tree.
#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
pub struct Writecap {
    /// A container for the fields of this writecap which are covered by the signature.
    body: WritecapBody,
    /// A digital signature which covers all of the fields in the write cap except for next.
    signature: Signature,
    /// The next write cap in the chain leading back to the root.
    next: Option<Box<Writecap>>,
}

impl Writecap {
    /// Returns the root key that was used to sign this writecap.
    pub fn root_signing_key(&self) -> &AsymKeyPub<Sign> {
        let mut writecap = self;
        while writecap.next.is_some() {
            writecap = writecap.next.as_ref().unwrap();
        }
        &writecap.body.signing_key
    }

    pub fn issued_to(&self) -> &Principal {
        &self.body.issued_to
    }

    /// Returns the [Epoch] representing the instant when this writecap expires.
    pub fn expires(&self) -> Epoch {
        self.body.expires
    }

    /// Returns the principal of the root key which was used to sign this writecap.
    pub fn root_principal(&self) -> Principal {
        self.root_signing_key().principal()
    }

    /// Returns the path to the root block of the blocktree that the root principal owns.
    pub fn root_block_path(&self) -> BlockPath {
        BlockPath::from_components(self.root_principal(), std::iter::empty())
    }

    /// Returns a reference to the path contained in this [Writecap].
    pub fn path(&self) -> &BlockPath {
        &self.body.path
    }

    /// Returns the path that the [Principal] this [Writecap] was issued to is allowed to bind.
    pub fn bind_path(&self) -> BlockPath {
        let mut path = self.body.path.clone();
        path.push_component(&self.body.issued_to);
        path
    }

    /// Returns [Ok] if and only if this [Writecap] was issued to `principal`.
    pub fn assert_issued_to(&self, principal: &Principal) -> Result<()> {
        if self.issued_to() == principal {
            Ok(())
        } else {
            Err(crypto::Error::NotIssuedTo.into())
        }
    }
}

/// Fragments are created from blocks using Erasure Encoding and stored with other nodes in the
/// network to provide availability and redundancy of data.
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Fragment {
    /// The path to the block this fragment is from.
    path: BlockPath,
    /// The serial number of this fragment.
    serial: FragmentSerial,
    /// The actual data.
    body: Vec<u8>,
}

impl Fragment {
    /// Create a new fragment with the given fields. If `path_str` cannot be parsed then a failed
    /// `Result` is returned containing a `PathError`.
    pub fn new(
        path_str: &str,
        serial_num: u32,
        body: Vec<u8>,
    ) -> std::result::Result<Fragment, BlockPathError> {
        let result = BlockPath::try_from(path_str);
        Ok(Fragment {
            path: result?,
            serial: FragmentSerial(serial_num),
            body,
        })
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
/// Structure for keeping track of server information in a directory.
pub struct ServerRecord {
    /// The most up-to-date address for this server.
    pub addr: IpAddr,
    /// The public credentials for this server.
    pub pub_creds: ConcretePub,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
/// Attributes associated with a principal which are used for authorization decisions.
pub struct AuthzAttrs {
    /// The user ID of the process being authorized.
    pub uid: u32,
    /// The group ID of the process being authorized.
    pub gid: u32,
    /// The group IDs of the supplemental groups in which a process is a member.
    pub supp_gids: Vec<u32>,
}

/// A record which stores the authorization attributes, credentials, writecap, and IP address of a
/// process.
#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
pub struct IssuedProcRec {
    /// The last known IP address of the process.
    pub addr: IpAddr,
    /// The public credentials of the process.
    pub pub_creds: ConcretePub,
    /// The writecap that was issued to the process.
    pub writecap: Writecap,
    /// The authorization attributes associated with the process.
    pub authz_attrs: AuthzAttrs,
}

/// Structure stored in process blocks for keeping track of process credentials and location.
#[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
pub enum ProcRec {
    Requested {
        addr: IpAddr,
        pub_creds: ConcretePub,
    },
    Valid(IssuedProcRec),
    Revoked(IssuedProcRec),
}

impl ProcRec {
    pub fn validate(self) -> Result<IssuedProcRec> {
        match self {
            Self::Requested { .. } => Err(BlockError::ProcRecNotIssued.into()),
            Self::Valid(valid) => Ok(valid),
            Self::Revoked(..) => Err(BlockError::ProcRecRevoked.into()),
        }
    }
}

#[derive(Debug, PartialEq, Serialize, Deserialize, EnumDiscriminants, Clone)]
#[strum_discriminants(derive(FromRepr, Display, Serialize, Deserialize))]
#[strum_discriminants(name(DirEntryKind))]
pub enum DirEntry {
    Directory(Inode),
    File(Inode),
    Server(Inode),
    Process(Inode),
}

impl DirEntry {
    pub fn inode(&self) -> Inode {
        match self {
            Self::Directory(inode) => *inode,
            Self::File(inode) => *inode,
            Self::Server(inode) => *inode,
            Self::Process(inode) => *inode,
        }
    }

    pub fn kind(&self) -> u8 {
        match self {
            Self::Directory(..) => libc::DT_DIR,
            Self::File(..) => libc::DT_REG,
            Self::Server(..) => libc::DT_UNKNOWN,
            Self::Process(..) => libc::DT_UNKNOWN,
        }
    }
}

/// This is the body contained in directory blocks.
#[derive(Debug, PartialEq, Serialize, Deserialize)]
pub struct Directory {
    /// This block's descendants.
    entries: BTreeMap<String, DirEntry>,
}

impl Directory {
    pub fn new() -> Directory {
        Directory {
            entries: BTreeMap::new(),
        }
    }

    pub fn add_file(&mut self, name: String, inode: Inode) -> Result<()> {
        let entry = match self.entries.entry(name) {
            btree_map::Entry::Occupied(entry) => {
                return Err(bterr!("directory already contains entry '{}'", entry.key()));
            }
            btree_map::Entry::Vacant(entry) => entry,
        };
        entry.insert(DirEntry::File(inode));
        Ok(())
    }

    pub fn num_entries(&self) -> usize {
        self.entries.len()
    }

    pub fn entries(&self) -> impl Iterator<Item = (&str, &DirEntry)> {
        self.entries
            .iter()
            .map(|(name, entry)| (name.as_str(), entry))
    }

    pub fn entry(&self, name: &str) -> Option<&DirEntry> {
        self.entries.get(name)
    }

    pub fn contains_entry(&self, name: &str) -> bool {
        self.entries.contains_key(name)
    }

    pub fn insert_entry(&mut self, name: String, entry: DirEntry) -> Option<DirEntry> {
        self.entries.insert(name, entry)
    }

    pub fn remove_entry(&mut self, name: &str) -> Option<DirEntry> {
        self.entries.remove(name)
    }
}

impl Default for Directory {
    fn default() -> Self {
        Self::new()
    }
}

/// Keeps track of which principal is storing a fragment.
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Clone)]
pub struct FragmentRecord {
    /// The fragment serial number this record is for.
    serial: FragmentSerial,
    /// The principal who is storing this fragment.
    stored_by: Principal,
}

impl FragmentRecord {
    /// Creates a new `FragmentRecord` whose `serial` and `stored_by` fields are set to
    /// the given values.
    pub fn new(serial: u32, stored_by: VarHash) -> FragmentRecord {
        FragmentRecord {
            serial: FragmentSerial(serial),
            stored_by: Principal(stored_by),
        }
    }
}

/// An identifier for a security principal, which is any entity that can be authenticated.
#[derive(
    Debug, PartialEq, Eq, Serialize, Deserialize, Hashable, Clone, Default, PartialOrd, Ord,
)]
pub struct Principal(VarHash);

impl Principal {
    pub fn kind(&self) -> HashKind {
        HashKind::from(&self.0)
    }

    pub fn as_slice(&self) -> &[u8] {
        self.0.as_slice()
    }

    pub fn as_mut_slice(&mut self) -> &mut [u8] {
        self.0.as_mut_slice()
    }
}

impl AsRef<[u8]> for Principal {
    fn as_ref(&self) -> &[u8] {
        self.as_slice()
    }
}

impl AsMut<[u8]> for Principal {
    fn as_mut(&mut self) -> &mut [u8] {
        self.as_mut_slice()
    }
}

impl TryFrom<&str> for Principal {
    type Error = Error;

    fn try_from(value: &str) -> Result<Self> {
        Ok(Principal(value.try_into()?))
    }
}

impl Display for Principal {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        self.0.fmt(f)
    }
}

/// Trait for types which are owned by a `Principal`.
pub trait Principaled {
    /// Returns the `Principal` that owns `self`, using the given hash algorithm.
    fn principal_of_kind(&self, kind: HashKind) -> Principal;

    /// Returns the `Principal` that owns `self`, using the default hash algorithm.
    fn principal(&self) -> Principal {
        self.principal_of_kind(HashKind::default())
    }
}

impl<T: ?Sized + Principaled, P: Deref<Target = T>> Principaled for P {
    fn principal_of_kind(&self, kind: HashKind) -> Principal {
        self.deref().principal_of_kind(kind)
    }
}

/// An instant in time represented by the number of seconds since January 1st 1970, 00:00:00 UTC.
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq, PartialOrd, Ord, Default, Hash)]
pub struct Epoch(u64);

impl Epoch {
    /// Returns the current epoch time.
    pub fn now() -> Epoch {
        let now = SystemTime::now();
        // If the system clock is before the unix epoch, just panic.
        let epoch = now.duration_since(SystemTime::UNIX_EPOCH).unwrap();
        Epoch(epoch.as_secs())
    }

    pub fn from_value(value: u64) -> Self {
        Self(value)
    }

    pub fn value(self) -> u64 {
        self.0
    }

    pub fn to_unix(self) -> libc::time_t {
        self.0 as libc::time_t
    }
}

impl Copy for Epoch {}

impl From<u64> for Epoch {
    fn from(value: u64) -> Self {
        Epoch::from_value(value)
    }
}

impl From<Epoch> for u64 {
    fn from(value: Epoch) -> Self {
        value.value()
    }
}

impl Add<Duration> for Epoch {
    type Output = Self;
    fn add(self, other: Duration) -> Self {
        Epoch(self.0 + other.as_secs())
    }
}

impl Sub<Duration> for Epoch {
    type Output = Self;
    fn sub(self, other: Duration) -> Self {
        Epoch(self.0 - other.as_secs())
    }
}

impl Sub<Epoch> for Epoch {
    type Output = Duration;
    fn sub(self, other: Epoch) -> Self::Output {
        Duration::from_secs(self.0 - other.0)
    }
}

/// The serial number of a block fragment.
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize, Hashable, Clone)]
pub struct FragmentSerial(u32);

#[cfg(test)]
mod tests {
    use btserde::{from_vec, to_vec};
    use std::{fs::OpenOptions, io::Write, path::PathBuf};
    use tempdir::TempDir;

    use super::*;
    use crate::{
        crypto::{ConcreteCreds, CredsPriv},
        sectored_buf::SectoredBuf,
        test_helpers::{node_creds, root_creds, SectoredCursor, SECTOR_SZ_DEFAULT},
        Cursor as PioCursor,
    };

    /// Tests that the `BlockMetaBody` struct has an updated secrets struct after it is modified
    /// in the `access_secrets` method.
    #[test]
    fn block_meta_body_secrets_updated_after_access() {
        const UID: u32 = 1000;
        let creds = test_helpers::NODE_CREDS.clone();

        let vec = {
            let mut body = BlockMetaBody::new(&creds).expect("failed to create meta body");
            body.access_secrets(|secrets| {
                secrets.uid = UID;
                Ok(())
            })
            .expect("access secrets failed");
            to_vec(&body).expect("to_vec failed")
        };

        let mut body = from_vec::<BlockMetaBody>(&vec).expect("from_vec failed");
        body.use_readcap_for(&creds)
            .expect("unlock_block_key failed");
        let actual_uid = body
            .access_secrets(|secrets| Ok(secrets.uid))
            .expect("access_secrets failed");
        assert_eq!(UID, actual_uid);
    }

    struct InMemTestCase {
        node_creds: ConcreteCreds,
        block_path: BlockPath,
        block_id: BlockId,
    }

    type EncBlock = SectoredBuf<
        SecretStream<PioCursor<MerkleStream<BlockStream<SectoredCursor<Vec<u8>>, ConcreteCreds>>>>,
    >;

    impl InMemTestCase {
        fn new() -> InMemTestCase {
            let components = ["nodes", "phone"];
            let node_creds = {
                let mut node_creds = node_creds().clone();
                let writecap = root_creds()
                    .issue_writecap(
                        node_creds.principal(),
                        &mut components.into_iter(),
                        Epoch::now() + Duration::from_secs(3600),
                    )
                    .expect("failed to issue writecap");
                node_creds.set_writecap(writecap).unwrap();
                node_creds
            };
            let block_path =
                BlockPath::from_components(root_creds().principal(), components.into_iter());
            let block_id = BlockId::default();
            Self {
                node_creds,
                block_path,
                block_id,
            }
        }

        fn stream(&self, vec: Vec<u8>) -> EncBlock {
            let inner = SectoredCursor::new(vec, SECTOR_SZ_DEFAULT).require_sect_sz(false);
            let mut stream = BlockStream::new(
                inner,
                self.node_creds.clone(),
                None,
                self.block_path.clone(),
            )
            .unwrap();
            stream
                .mut_meta_body()
                .access_secrets(|secrets| {
                    secrets.block_id = self.block_id.clone();
                    Ok(())
                })
                .unwrap();
            let block_key = stream.meta_body().block_key().unwrap().to_owned();
            let mut stream = MerkleStream::new(stream).unwrap();
            stream.assert_root_integrity().unwrap();
            let stream = PioCursor::new(stream);
            let stream = SecretStream::new(block_key).try_compose(stream).unwrap();
            SectoredBuf::new(stream).unwrap()
        }

        fn into_vec(stream: EncBlock) -> Vec<u8> {
            stream
                .into_inner()
                .into_inner()
                .into_inner()
                .into_inner()
                .into_inner()
                .0
                .into_inner()
        }
    }

    #[test]
    fn block_write_read_with_cursor() {
        const EXPECTED: &[u8] = b"Silly sordid sulking sultans.";
        let case = InMemTestCase::new();
        let mut stream = case.stream(Vec::new());

        stream.write_all(EXPECTED).unwrap();
        stream.flush().unwrap();

        let vec = InMemTestCase::into_vec(stream);
        let mut stream = case.stream(vec);
        let mut actual = [0u8; EXPECTED.len()];
        stream.read(&mut actual).unwrap();

        assert_eq!(EXPECTED, actual);
    }

    #[test]
    fn block_write_multiple() {
        const ITER: usize = 16;
        let case = InMemTestCase::new();
        let mut stream = case.stream(Vec::new());
        let expected = vec![1u8; stream.sector_sz()];

        for _ in 0..ITER {
            stream.write(&expected).unwrap();
        }
        stream.flush().unwrap();
    }

    pub struct BlockTestCase {
        temp_dir: TempDir,
        root_path: BlockPath,
        node_path: BlockPath,
        root_creds: ConcreteCreds,
        node_creds: ConcreteCreds,
    }

    impl BlockTestCase {
        fn new() -> BlockTestCase {
            let temp_dir = TempDir::new("block_test").expect("failed to create temp dir");
            let root_creds = test_helpers::ROOT_CREDS.clone();
            let mut node_creds = test_helpers::NODE_CREDS.clone();
            let components = ["nodes", "phone"];
            let writecap = root_creds
                .issue_writecap(
                    node_creds.principal(),
                    &mut components.into_iter(),
                    Epoch::now() + Duration::from_secs(3600),
                )
                .expect("failed to issue writecap");
            node_creds.set_writecap(writecap).unwrap();
            let case = BlockTestCase {
                temp_dir,
                node_path: BlockPath::from_components(
                    root_creds.principal(),
                    components.into_iter(),
                ),
                root_path: BlockPath::from_components(root_creds.principal(), std::iter::empty()),
                node_creds,
                root_creds,
            };
            std::fs::create_dir_all(case.fs_path(&case.node_path))
                .expect("failed to create node path");
            case
        }

        fn fs_path(&self, path: &crate::BlockPath) -> PathBuf {
            let mut fs_path = self.temp_dir.path().to_owned();
            fs_path.extend(path.components());
            fs_path
        }

        fn open_new(&mut self, path: crate::BlockPath) -> Accessor<impl Block + 'static> {
            let file = OpenOptions::new()
                .create_new(true)
                .read(true)
                .write(true)
                .open(&self.fs_path(&path))
                .expect("failed to open file");
            let mut block = BlockOpenOptions::new()
                .with_inner(file)
                .with_creds(self.node_creds.clone())
                .with_encrypt(true)
                .with_block_path(path)
                .open()
                .expect("failed to open block");
            *block.mut_meta_body().path_mut() =
                self.node_creds.writecap().unwrap().body.path.clone();
            block
        }

        fn open_existing(&mut self, path: crate::BlockPath) -> Accessor<impl Block + 'static> {
            let file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(&self.fs_path(&path))
                .expect("failed to reopen file");
            BlockOpenOptions::new()
                .with_inner(file)
                .with_creds(self.node_creds.clone())
                .with_encrypt(true)
                .with_block_path(path)
                .open()
                .expect("failed to reopen block")
        }

        /// Returns a path in the directory which the node creds have permission to write to.
        fn node_path(&self, file_name: &str) -> BlockPath {
            let mut path = self.node_path.clone();
            path.push_component(file_name.to_owned());
            path
        }
    }

    #[test]
    fn block_can_create_empty() {
        let case = BlockTestCase::new();
        BlockOpenOptions::new()
            .with_inner(SectoredCursor::new(Vec::<u8>::new(), SECTOR_SZ_DEFAULT))
            .with_creds(case.node_creds)
            .with_encrypt(true)
            .with_block_path(case.root_path)
            .open()
            .expect("failed to open block");
    }

    #[test]
    fn block_contents_persisted() {
        const EXPECTED: &[u8] = b"Silly sordid sulking sultans.";

        let mut case = BlockTestCase::new();
        let path = case.node_path("test.blk");
        {
            let mut block = case.open_new(path.clone());
            block.write(EXPECTED).expect("failed to write");
            block.flush().expect("flush failed");
        }
        let mut block = case.open_existing(path);
        let mut actual = [0u8; EXPECTED.len()];
        block.read(&mut actual).expect("read failed");
        assert_eq!(EXPECTED, actual);
    }

    #[test]
    fn block_write_twice() {
        const EXPECTED: &[u8] = b"Cool callous calamitous colonels.";
        const MID: usize = EXPECTED.len() / 2;

        let mut case = BlockTestCase::new();
        let path = case.node_path("test.blk");
        {
            let mut block = case.open_new(path.clone());
            block.write(&EXPECTED[..MID]).expect("first write failed");
            block.flush().expect("first flush failed");
        }
        {
            let mut block = case.open_existing(path.clone());
            block
                .seek(SeekFrom::Start(MID.try_into().unwrap()))
                .expect("seek failed");
            block.write(&EXPECTED[MID..]).expect("second write failed");
            block.flush().expect("second flush failed");
        }
        {
            let mut block = case.open_existing(path);
            let mut actual = [0u8; EXPECTED.len()];
            block.read(&mut actual).expect("read failed");
            assert_eq!(EXPECTED, actual);
        }
    }

    #[test]
    fn block_write_with_different_creds() {
        const EXPECTED: &[u8] = b"Cool callous calamitous colonels.";
        const MID: usize = EXPECTED.len() / 2;

        let mut case = BlockTestCase::new();
        let path = case.node_path("test.blk");
        let app_creds = {
            let mut app_creds = ConcreteCreds::generate().expect("failed to generate app creds");
            let writecap = case
                .root_creds
                .issue_writecap(
                    app_creds.principal(),
                    &mut path.components(),
                    Epoch::now() + Duration::from_secs(60),
                )
                .expect("failed to issue writecap");
            app_creds.set_writecap(writecap).unwrap();
            app_creds
        };
        {
            let mut block = case.open_new(path.clone());
            block
                .mut_meta_body()
                .add_readcap_for(&app_creds)
                .expect("failed to add readcap");
            block.write(&EXPECTED[..MID]).expect("first write failed");
            block.flush().expect("first flush failed");
        }
        {
            let file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(case.fs_path(&path))
                .expect("failed to reopen file");
            let mut block = BlockOpenOptions::new()
                .with_inner(file)
                // Note that this write is performed using app_creds.
                .with_creds(app_creds)
                .with_encrypt(true)
                .with_block_path(path.clone())
                .open()
                .expect("failed to reopen block");
            block
                .seek(SeekFrom::Start(MID.try_into().unwrap()))
                .expect("seek failed");
            block.write(&EXPECTED[MID..]).expect("second write failed");
            block.flush().expect("second flush failed");
        }
        {
            let file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(case.fs_path(&path))
                .expect("failed to reopen file");
            let mut block = BlockOpenOptions::new()
                .with_inner(file)
                .with_creds(case.node_creds)
                .with_encrypt(true)
                .with_block_path(path)
                .open()
                .expect("failed to reopen block");
            let mut actual = [0u8; EXPECTED.len()];
            block.read(&mut actual).expect("read failed");
            assert_eq!(EXPECTED, actual);
        }
    }

    #[test]
    fn block_try_seek_and_get_buf() {
        const DIVISOR: usize = 8;
        let mut case = BlockTestCase::new();
        let path = case.node_path("blob.dat");
        let mut block = case.open_new(path);
        let sect_sz = block.sector_sz();
        let read_sz = sect_sz / DIVISOR;
        let mut expected = vec![0u8; read_sz];

        for index in 0..(DIVISOR as u8 + 2) {
            expected.fill(index + 1);
            block.write(&expected).unwrap();
        }

        block.rewind().unwrap();
        for index in 0..(DIVISOR as u8 + 2) {
            let offset = (read_sz * index as usize) as u64;
            block.try_seek(SeekFrom::Start(offset)).unwrap();
            let actual = block.get_buf(offset, read_sz as u64).unwrap();
            expected.fill(index + 1);
            assert!(actual == expected);
        }
    }

    /// Tests that the last component of a [Writecap]'s bind path is the string representation of
    /// the [Principal] to which the [Writecap] was issued.
    #[test]
    fn writecap_bind_path_last_component_is_principal() {
        let creds = node_creds();
        let writecap = creds.writecap().unwrap();
        let expected = writecap.issued_to().to_string();

        let bind_path = writecap.bind_path();
        let actual = bind_path.components().last().unwrap();

        assert_eq!(expected, actual);
    }
}