Started implementing btfsd.

Cargo.lock

@@ -243,13 +243,19 @@ dependencies = [
 ]
 
 [[package]]
-name = "btfs"
+name = "btfsd"
 version = "0.1.0"
 dependencies = [
+ "btfproto",
  "btlib",
+ "btlib-tests",
  "btmsg",
+ "ctor",
+ "env_logger",
  "log",
- "tss-esapi",
+ "swtpm-harness",
+ "tempdir",
+ "tokio",
 ]
 
 [[package]]
@@ -272,7 +278,6 @@ dependencies = [
  "swtpm-harness",
  "tempdir",
  "tokio",
- "tss-esapi",
 ]
 
 [[package]]
@@ -312,6 +317,14 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "btlib-tests"
+version = "0.1.0"
+dependencies = [
+ "btlib",
+ "swtpm-harness",
+]
+
 [[package]]
 name = "btmsg"
 version = "0.1.0"

Cargo.toml

@@ -1,13 +1,14 @@
 [workspace]
 members = [
     "crates/btlib",
+    "crates/btlib-tests",
     "crates/btmsg",
     "crates/btserde",
     "crates/swtpm-harness",
     "crates/btfproto",
     "crates/btfproto-tests",
     "crates/btfuse",
-    "crates/btfs",
+    "crates/btfsd",
 ]
 
 [profile.bench]

crates/btfproto-tests/src/local_fs.rs

@@ -118,7 +118,7 @@ impl LocalFsTest {
 mod tests {
     use super::*;
 
-    use btfproto::{msg::*, local_fs::Error};
+    use btfproto::{local_fs::Error, msg::*};
     use btlib::{Inode, Result, SECTOR_SZ_DEFAULT};
     use bytes::BytesMut;
     use std::{

crates/btfproto/src/local_fs.rs

@@ -5,10 +5,10 @@ use btlib::{
     bterr,
     crypto::{Creds, Decrypter, Signer},
     error::{BtErr, DisplayErr},
-    BlockAccessor, BlockError, BlockMeta, BlockMetaSecrets, BlockOpenOptions, BlockPath,
-    BlockReader, DirEntry, DirEntryKind, Directory, Epoch, FileBlock, FlushMeta, MetaAccess,
-    MetaReader, Positioned, Principaled, Result, Split, TrySeek, WriteDual, ZeroExtendable,
-    AuthzAttrs, IssuedProcRec, Principal, ProcRec,
+    AuthzAttrs, BlockAccessor, BlockError, BlockMeta, BlockMetaSecrets, BlockOpenOptions,
+    BlockPath, BlockReader, DirEntry, DirEntryKind, Directory, Epoch, FileBlock, FlushMeta,
+    IssuedProcRec, MetaAccess, MetaReader, Positioned, Principal, Principaled, ProcRec, Result,
+    Split, TrySeek, WriteDual, ZeroExtendable,
 };
 use btserde::{read_from, write_to};
 use core::future::{ready, Ready};
@@ -20,13 +20,13 @@ use std::{
     fmt::{Display, Formatter},
     fs::File,
     io::{self, Read as IoRead, Seek, SeekFrom, Write as IoWrite},
+    net::{IpAddr, Ipv6Addr, SocketAddr},
     path::{Path, PathBuf},
     sync::{
         atomic::{AtomicU64, Ordering},
         Arc, Mutex, RwLock, RwLockWriteGuard,
     },
     time::Duration,
-    net::{SocketAddr, IpAddr, Ipv6Addr},
 };
 
 pub use private::{Authorizer, AuthzContext, Error, LocalFs, ModeAuthorizer};

crates/btfproto/src/server.rs

@@ -7,7 +7,6 @@ use btlib::{crypto::Creds, BlockPath, IssuedProcRec, Result};
 use btmsg::{receiver, MsgCallback, MsgReceived, Receiver};
 use core::future::Future;
 use std::{io::Read, net::IpAddr, sync::Arc};
-use tokio::runtime::Handle as RuntimeHandle;
 
 pub trait FsProvider: Send + Sync {
     type LookupFut<'c>: Send + Future<Output = Result<LookupReply>>
@@ -266,11 +265,18 @@ impl<P: 'static + Send + Sync + FsProvider> MsgCallback for ServerCallback<P> {
                 FsMsg::Create(create) => FsReply::Create(provider.create(&from, create).await?),
                 FsMsg::Open(open) => FsReply::Open(provider.open(&from, open).await?),
                 FsMsg::Read(read) => {
-                    return provider.read(&from, read, move |data| {
-                        let mut replier = replier.unwrap();
-                        RuntimeHandle::current()
-                            .block_on(replier.reply(FsReply::Read(ReadReply { data })))
+                    let buf = provider.read(&from, read, move |data| {
+                        // TODO: Avoid allocating a buffer on every read. If possible, avoid coping
+                        // data altogether.
+                        let mut buf = Vec::with_capacity(data.len());
+                        buf.extend_from_slice(data);
+                        buf
                     })?;
+                    let mut replier = replier.unwrap();
+                    replier
+                        .reply(FsReply::Read(ReadReply { data: &buf }))
+                        .await?;
+                    return Ok(());
                 }
                 FsMsg::Write(Write {
                     inode,

crates/btfs/src/main.rs

@@ -1,50 +0,0 @@
-use log::info;
-use std::{
-    env::{self, VarError},
-    fs,
-    path::PathBuf,
-    str::FromStr,
-};
-use tss_esapi::{tcti_ldr::TabrmdConfig, Context};
-
-use btlib::crypto::{tpm::TpmCredStore, CredStore};
-
-const PATH_ENV: &str = "BT_DIR";
-const TABRMD_ENV: &str = "BT_TABRMD";
-const TABRMD_CONFIG: &str = "bus_type=session,bus_name=com.intel.tss2.Tabrmd";
-const STATE_FILE: &str = "tpm_state";
-
-fn btdir() -> PathBuf {
-    env::var(PATH_ENV).map(PathBuf::from).unwrap_or_else(|err| {
-        if let VarError::NotPresent = err {
-            let mut path_buf: PathBuf = env::var("HOME").unwrap().into();
-            path_buf.push(".config");
-            path_buf.push("blocktree");
-            path_buf
-        } else {
-            panic!("invalid value of the '{PATH_ENV}' environment variable");
-        }
-    })
-}
-
-fn tabrmd_config() -> String {
-    env::var(TABRMD_ENV).unwrap_or_else(|err| {
-        if let VarError::NotPresent = err {
-            TABRMD_CONFIG.to_string()
-        } else {
-            panic!("invalid value of the '{TABRMD_ENV}' environment variable");
-        }
-    })
-}
-
-fn main() {
-    let btdir = btdir();
-    info!("using btdir= '{}'", btdir.display());
-    fs::create_dir_all(&btdir).unwrap();
-    let tabrmd_config = tabrmd_config();
-    info!("using tabrmd_config = '{tabrmd_config}'");
-    let context =
-        Context::new_with_tabrmd(TabrmdConfig::from_str(&tabrmd_config).unwrap()).unwrap();
-    let cred_store = TpmCredStore::new(context, btdir.join(STATE_FILE)).unwrap();
-    let _node_creds = cred_store.node_creds().unwrap();
-}

crates/btfsd/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "btfsd"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+btlib = { path = "../btlib" }
+btmsg = { path = "../btmsg" }
+btfproto = { path = "../btfproto" }
+tokio = { version = "1.24.2", features = ["rt", "rt-multi-thread", "time"] }
+log = "0.4.17"
+env_logger = "0.9.0"
+
+[dev-dependencies]
+swtpm-harness = { path = "../swtpm-harness" }
+btlib-tests = { path = "../btlib-tests" }
+tempdir = "0.3.7"
+ctor = { version = "0.1.22" }

crates/btfsd/src/config.rs

@@ -0,0 +1,79 @@
+use super::DEFAULT_CONFIG;
+use std::{
+    net::IpAddr,
+    path::{Path, PathBuf},
+};
+
+pub struct Config {
+    pub ip_addr: IpAddr,
+    pub tabrmd: String,
+    pub tpm_state_path: PathBuf,
+    pub block_dir: PathBuf,
+}
+
+impl Config {
+    pub fn builder() -> ConfigBuilder {
+        ConfigBuilder::new()
+    }
+}
+
+#[derive(Default)]
+pub struct ConfigBuilder {
+    pub ip_addr: Option<IpAddr>,
+    pub tabrmd: Option<String>,
+    pub tpm_state_path: Option<PathBuf>,
+    pub block_dir: Option<PathBuf>,
+}
+
+impl ConfigBuilder {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn with_ip_addr(mut self, ip_addr: Option<IpAddr>) -> Self {
+        self.ip_addr = ip_addr;
+        self
+    }
+
+    pub fn with_tabrmd(mut self, tabrmd: Option<String>) -> Self {
+        self.tabrmd = tabrmd;
+        self
+    }
+
+    pub fn with_tpm_state_path(mut self, tpm_state_path: Option<PathBuf>) -> Self {
+        self.tpm_state_path = tpm_state_path;
+        self
+    }
+
+    pub fn with_block_dir(mut self, block_dir: Option<PathBuf>) -> Self {
+        self.block_dir = block_dir;
+        self
+    }
+
+    pub fn build(self) -> Config {
+        Config {
+            ip_addr: self.ip_addr.unwrap_or(DEFAULT_CONFIG.ip_addr),
+            tabrmd: self.tabrmd.unwrap_or(DEFAULT_CONFIG.tabrmd.to_owned()),
+            tpm_state_path: self
+                .tpm_state_path
+                .unwrap_or(Path::new(DEFAULT_CONFIG.tpm_state_path).to_owned()),
+            block_dir: self
+                .block_dir
+                .unwrap_or(Path::new(DEFAULT_CONFIG.block_dir).to_owned()),
+        }
+    }
+}
+
+pub struct ConfigRef<'a> {
+    pub ip_addr: IpAddr,
+    pub tabrmd: &'a str,
+    pub tpm_state_path: &'a str,
+    pub block_dir: &'a str,
+}
+
+pub struct Envvars<'a> {
+    pub ip_addr: &'a str,
+    pub tabrmd: &'a str,
+    pub tpm_state_path: &'a str,
+    pub block_dir: &'a str,
+}

crates/btfsd/src/main.rs

@@ -0,0 +1,153 @@
+mod config;
+
+use btfproto::{
+    local_fs::{LocalFs, ModeAuthorizer},
+    server::{new_fs_server, FsProvider},
+};
+use btlib::{
+    config_helpers::from_envvar,
+    crypto::{tpm::TpmCredStore, CredStore, Creds},
+};
+use btmsg::Receiver;
+use config::{Config, ConfigRef, Envvars};
+use std::{
+    net::{IpAddr, Ipv6Addr},
+    path::PathBuf,
+    str::FromStr,
+    sync::Arc,
+};
+
+const ENVVARS: Envvars<'static> = Envvars {
+    ip_addr: "BTFSD_IPADDR",
+    tabrmd: "BTFSD_TABRMD",
+    tpm_state_path: "BTFSD_TPMSTATE",
+    block_dir: "BTFSD_BLOCKDIR",
+};
+
+const DEFAULT_CONFIG: ConfigRef<'static> = ConfigRef {
+    ip_addr: IpAddr::V6(Ipv6Addr::LOCALHOST),
+    tabrmd: "bus_type=session",
+    tpm_state_path: "./tpm_state",
+    block_dir: "./bt",
+};
+
+fn provider<C: 'static + Send + Sync + Creds>(block_dir: PathBuf, creds: C) -> impl FsProvider {
+    if block_dir.exists() {
+        LocalFs::new_existing(block_dir, creds, ModeAuthorizer).unwrap()
+    } else {
+        std::fs::create_dir_all(&block_dir).unwrap();
+        LocalFs::new_empty(block_dir, 0, creds, ModeAuthorizer).unwrap()
+    }
+}
+
+fn receiver(config: Config) -> impl Receiver {
+    let cred_store = TpmCredStore::from_tabrmd(&config.tabrmd, config.tpm_state_path).unwrap();
+    let node_creds = cred_store.node_creds().unwrap();
+    let provider = Arc::new(provider(config.block_dir, node_creds.clone()));
+    new_fs_server(config.ip_addr, Arc::new(node_creds), provider).unwrap()
+}
+
+#[tokio::main]
+async fn main() {
+    let ip_addr = from_envvar(ENVVARS.ip_addr)
+        .unwrap()
+        .map(|txt| IpAddr::from_str(&txt).unwrap());
+    let tabrmd = from_envvar(ENVVARS.tabrmd).unwrap();
+    let tpm_state_path = from_envvar(ENVVARS.tpm_state_path)
+        .unwrap()
+        .map(PathBuf::from);
+    let block_dir = from_envvar(ENVVARS.block_dir).unwrap().map(PathBuf::from);
+    let config = Config::builder()
+        .with_ip_addr(ip_addr)
+        .with_tabrmd(tabrmd)
+        .with_tpm_state_path(tpm_state_path)
+        .with_block_dir(block_dir)
+        .build();
+    let receiver = receiver(config);
+    receiver.complete().unwrap().await.unwrap();
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use btfproto::{client::FsClient, msg::*};
+    use btlib_tests::TpmCredStoreHarness;
+    use std::future::ready;
+    use tempdir::TempDir;
+
+    struct TestCase {
+        _ip_addr: IpAddr,
+        _harness: TpmCredStoreHarness,
+        _dir: TempDir,
+    }
+
+    impl TestCase {
+        const ROOT_PASSWD: &str = "existential_threat";
+
+        fn new() -> (Self, impl Receiver) {
+            let dir = TempDir::new("btfsd").unwrap();
+            let harness = TpmCredStoreHarness::new(Self::ROOT_PASSWD.to_owned()).unwrap();
+            let ip_addr = IpAddr::V6(Ipv6Addr::LOCALHOST);
+            let config = Config {
+                ip_addr,
+                tabrmd: harness.swtpm().tabrmd_config().to_owned(),
+                tpm_state_path: harness.swtpm().state_path().to_owned(),
+                block_dir: dir.path().join("bt"),
+            };
+            let receiver = receiver(config);
+            (
+                Self {
+                    _dir: dir,
+                    _harness: harness,
+                    _ip_addr: ip_addr,
+                },
+                receiver,
+            )
+        }
+    }
+
+    #[allow(dead_code)]
+    async fn manual_test() {
+        let (_case, rx) = TestCase::new();
+        rx.complete().unwrap().await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn create_write_read() {
+        const FILENAME: &str = "file.txt";
+        const EXPECTED: &[u8] = b"potato";
+        let (_case, rx) = TestCase::new();
+        let tx = rx.transmitter(rx.addr().clone()).await.unwrap();
+        let client = FsClient::new(tx);
+
+        let CreateReply { inode, handle, .. } = client
+            .create(
+                SpecInodes::RootDir.into(),
+                FILENAME,
+                FlagValue::ReadWrite.into(),
+                0o644,
+                0,
+            )
+            .await
+            .unwrap();
+        let WriteReply { written, .. } = client.write(inode, handle, 0, EXPECTED).await.unwrap();
+        assert_eq!(EXPECTED.len() as u64, written);
+        let msg = Read {
+            inode,
+            handle,
+            offset: 0,
+            size: EXPECTED.len() as u64,
+        };
+        let actual = client
+            .read(msg, |reply| {
+                let mut buf = Vec::with_capacity(EXPECTED.len());
+                buf.extend_from_slice(reply.data);
+                ready(buf)
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(EXPECTED, &actual);
+    }
+}

crates/btfuse/Cargo.toml

@@ -9,7 +9,6 @@ edition = "2021"
 btlib = { path = "../btlib" }
 btserde = { path = "../btserde" }
 swtpm-harness = { path = "../swtpm-harness" }
-tss-esapi = { version = "7.1.0", features = ["generate-bindings"] }
 btfproto = { path = "../btfproto" }
 tokio = { version = "1.24.2", features = ["rt", "rt-multi-thread"] }
 fuse-backend-rs = { version = "0.9.6", features = ["async-io"] }

crates/btfuse/src/fuse_daemon.rs

@@ -45,17 +45,17 @@ mod private {
     impl FuseDaemon {
         const FSNAME: &str = "btfuse";
         const FSTYPE: &str = "bt";
-        const NUM_THREADS: usize = 1;
 
         pub fn new<P: 'static + FsProvider>(
             mnt_path: PathBuf,
+            num_threads: usize,
             fallback_path: Arc<BlockPath>,
             provider: P,
         ) -> Result<Self> {
             let server = Arc::new(Server::new(FuseFs::new(provider, fallback_path)));
             let session = Self::session(mnt_path)?;
             let mut set = JoinSet::new();
-            for _ in 0..Self::NUM_THREADS {
+            for _ in 0..num_threads {
                 let server = server.clone();
                 let channel = session.new_channel()?;
                 let future =
@@ -79,10 +79,6 @@ mod private {
         }
 
         /// Opens a channel to the kernel and processes messages received in an infinite loop.
-        ///
-        /// # Warning
-        /// Because the future returned by this method is passed to `unsafe_cast`, you must not
-        /// hold any type which is not `Send` and `Sync` across an `await` point.
         fn server_loop<P: 'static + FsProvider>(
             server: Arc<Server<FuseFs<P>>>,
             mut channel: FuseChannel,

crates/btfuse/src/main.rs

@@ -7,6 +7,7 @@ use config::{Config, ConfigRef, EnvVars};
 
 use btfproto::{local_fs::LocalFs, server::FsProvider};
 use btlib::{
+    config_helpers::from_envvar,
     crypto::{
         tpm::{TpmCredStore, TpmCreds},
         CredStore, Creds, CredsPriv,
@@ -15,17 +16,11 @@ use btlib::{
 };
 use core::future::Future;
 use std::{
-    env::VarError,
     fs::{self},
     io,
     path::{Path, PathBuf},
-    str::FromStr,
     sync::Arc,
 };
-use tss_esapi::{
-    tcti_ldr::{TabrmdConfig, TctiNameConf},
-    Context,
-};
 
 const ENVVARS: EnvVars = EnvVars {
     tabrmd: "BT_TABRMD",
@@ -58,8 +53,7 @@ impl<T: AsRef<Path>> PathExt for T {
 }
 
 fn node_creds(state_file: PathBuf, tabrmd_cfg: &str) -> Result<TpmCreds> {
-    let context = Context::new(TctiNameConf::Tabrmd(TabrmdConfig::from_str(tabrmd_cfg)?))?;
-    let cred_store = TpmCredStore::new(context, state_file)?;
+    let cred_store = TpmCredStore::from_tabrmd(tabrmd_cfg, state_file)?;
     cred_store.node_creds()
 }
 
@@ -76,18 +70,6 @@ fn provider<C: 'static + Creds + Send + Sync>(
     }
 }
 
-fn from_envvar(envvar_name: &str) -> Option<String> {
-    match std::env::var(envvar_name) {
-        Ok(uid_map) => Some(uid_map),
-        Err(err) => match err {
-            VarError::NotPresent => None,
-            VarError::NotUnicode(_) => {
-                panic!("environment variable {envvar_name} contained non-unicode characters");
-            }
-        },
-    }
-}
-
 fn run_daemon(config: Config) -> impl Send + Sync + Future<Output = ()> {
     let node_creds =
         node_creds(config.tpm_state_file, &config.tabrmd).expect("failed to get node creds");
@@ -100,7 +82,7 @@ fn run_daemon(config: Config) -> impl Send + Sync + Future<Output = ()> {
     };
     let provider = provider(config.block_dir, node_creds).expect("failed to create FS provider");
 
-    let mut daemon = FuseDaemon::new(config.mnt_dir, fallback_path, provider)
+    let mut daemon = FuseDaemon::new(config.mnt_dir, config.threads, fallback_path, provider)
         .expect("failed to create FUSE daemon");
     async move { daemon.finished().await }
 }
@@ -112,8 +94,8 @@ async fn main() {
     let builder = Config::builder()
         .with_block_dir(args.next())
         .with_mnt_dir(args.next())
-        .with_tabrmd(from_envvar(ENVVARS.tabrmd))
-        .with_mnt_options(from_envvar(ENVVARS.mnt_options));
+        .with_tabrmd(from_envvar(ENVVARS.tabrmd).unwrap())
+        .with_mnt_options(from_envvar(ENVVARS.mnt_options).unwrap());
     let config = builder.build();
     run_daemon(config).await;
 }
@@ -234,7 +216,7 @@ mod test {
             let state_path: PathBuf = swtpm.state_path().to_owned();
             let cred_store = {
                 let context = swtpm.context().unwrap();
-                TpmCredStore::new(context, state_path.clone()).unwrap()
+                TpmCredStore::from_context(context, state_path.clone()).unwrap()
             };
             let root_creds = cred_store.gen_root_creds(ROOT_PASSWD).unwrap();
             let mut node_creds = cred_store.node_creds().unwrap();

crates/btfs/Cargo.toml → crates/btlib-tests/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "btfs"
+name = "btlib-tests"
 version = "0.1.0"
 edition = "2021"
 
@@ -7,6 +7,4 @@ edition = "2021"
 
 [dependencies]
 btlib = { path = "../btlib" }
-btmsg = { path = "../btmsg" }
-tss-esapi = { version = "7.1.0", features = ["generate-bindings"] }
-log = "0.4.17"
+swtpm-harness = { path = "../swtpm-harness" }

crates/btlib-tests/src/lib.rs

@@ -0,0 +1,2 @@
+mod tpm_cred_store_harness;
+pub use tpm_cred_store_harness::TpmCredStoreHarness;

crates/btlib-tests/src/tpm_cred_store_harness.rs

@@ -0,0 +1,52 @@
+//! Module containing [TpmCredStoreHarness].
+
+use btlib::{
+    crypto::{tpm::TpmCredStore, CredStore, Creds},
+    error::AnyhowErrorExt,
+    Epoch, Principaled, Result,
+};
+use core::time::Duration;
+use swtpm_harness::SwtpmHarness;
+
+/// A test harness which allows a [TpmCredStore] to be accessed.
+pub struct TpmCredStoreHarness {
+    root_passwd: String,
+    cred_store: TpmCredStore,
+    swtpm: SwtpmHarness,
+}
+
+impl TpmCredStoreHarness {
+    /// Creates a new test harness by starting a new instance of swtpm, generating root and node
+    /// creds, and issuing a writecap to the node creds.
+    pub fn new(root_passwd: String) -> Result<Self> {
+        let swtpm = SwtpmHarness::new().bterr()?;
+        let cred_store =
+            TpmCredStore::from_context(swtpm.context()?, swtpm.state_path().to_owned())?;
+        let root_creds = cred_store.gen_root_creds(&root_passwd).unwrap();
+        let mut node_creds = cred_store.node_creds().unwrap();
+        let expires = Epoch::now() + Duration::from_secs(3600);
+        let writecap = root_creds
+            .issue_writecap(node_creds.principal(), vec![], expires)
+            .unwrap();
+        cred_store
+            .assign_node_writecap(&mut node_creds, writecap)
+            .unwrap();
+        Ok(Self {
+            root_passwd,
+            swtpm,
+            cred_store,
+        })
+    }
+
+    pub fn root_passwd(&self) -> &str {
+        &self.root_passwd
+    }
+
+    pub fn swtpm(&self) -> &SwtpmHarness {
+        &self.swtpm
+    }
+
+    pub fn cred_store(&self) -> &TpmCredStore {
+        &self.cred_store
+    }
+}

crates/btlib/src/collections.rs

@@ -3,4 +3,4 @@
 pub mod hash_map_with_default;
 pub use hash_map_with_default::HashMapWithDefault;
 pub mod bijection;
-pub use bijection::Bijection;
+pub use bijection::Bijection;

crates/btlib/src/collections/bijection.rs

@@ -1,12 +1,9 @@
 //! This module contains the [Bijection] type.
 
 use super::HashMapWithDefault;
-use std::{
-    borrow::Borrow,
-    hash::Hash,
-};
 use log::error;
- 
+use std::{borrow::Borrow, hash::Hash};
+
 pub struct Bijection<K, V> {
     k2v: HashMapWithDefault<K, V>,
     v2k: HashMapWithDefault<V, K>,
@@ -48,7 +45,7 @@ impl<K: Clone + Hash + Eq, V: Clone + Hash + Eq> Bijection<K, V> {
 impl<K: Hash + Eq, V> Bijection<K, V> {
     pub fn value<Q: ?Sized + Hash + Eq>(&self, key: &Q) -> &V
     where
-        K: Borrow<Q>
+        K: Borrow<Q>,
     {
         self.k2v.get(key)
     }
@@ -57,8 +54,8 @@ impl<K: Hash + Eq, V> Bijection<K, V> {
 impl<K, V: Hash + Eq> Bijection<K, V> {
     pub fn key<R: ?Sized + Hash + Eq>(&self, value: &R) -> &K
     where
-        V: Borrow<R>
+        V: Borrow<R>,
     {
         self.v2k.get(value)
     }
-}
+}

crates/btlib/src/collections/hash_map_with_default.rs

@@ -2,12 +2,7 @@
 //! a default value to be defined.
 //! When a key is not found in the [HashMap] the default value is returned, instead of [None].
 
-use std::{
-    borrow::Borrow,
-    collections::HashMap,
-    hash::Hash,
-    ops::Index,
-};
+use std::{borrow::Borrow, collections::HashMap, hash::Hash, ops::Index};
 
 pub struct HashMapWithDefault<K, V> {
     hash_map: HashMap<K, V>,
@@ -16,7 +11,10 @@ pub struct HashMapWithDefault<K, V> {
 
 impl<K, V> HashMapWithDefault<K, V> {
     pub fn new(default: V) -> Self {
-        Self { hash_map: HashMap::new(), default }
+        Self {
+            hash_map: HashMap::new(),
+            default,
+        }
     }
 
     pub fn get_ref(&self) -> &HashMap<K, V> {

crates/btlib/src/config_helpers.rs

@@ -0,0 +1,20 @@
+//! This module contains functions which are useful for configuring applications.
+
+use std::env::VarError;
+
+use crate::{bterr, Result};
+
+/// Returns the value of the given environment variable, if it is defined and contains all unicode
+/// characters. `Ok(None)` is returned if the environment variable is not defined, and [Err] is
+/// returned if it contains non-unicode characters.
+pub fn from_envvar(envvar_name: &str) -> Result<Option<String>> {
+    match std::env::var(envvar_name) {
+        Ok(uid_map) => Ok(Some(uid_map)),
+        Err(err) => match err {
+            VarError::NotPresent => Ok(None),
+            VarError::NotUnicode(_) => Err(bterr!(
+                "environment variable {envvar_name} contained non-unicode characters"
+            )),
+        },
+    }
+}

crates/btlib/src/crypto/tpm.rs

@@ -18,6 +18,7 @@ use std::{
     ops::Deref,
     os::{raw::c_char, unix::fs::PermissionsExt},
     path::{Path, PathBuf},
+    str::FromStr,
     sync::{Arc, RwLock, RwLockWriteGuard},
     time::Duration,
 };
@@ -853,7 +854,14 @@ impl TpmCredStore {
     const ENCRYPT_SCHEME: Encrypt = Encrypt::RSA_OAEP_2048_SHA_256;
     const DEFAULT_WRITECAP_EXP: Duration = Duration::from_secs(60 * 60 * 24 * 365 * 10);
 
-    pub fn new(mut context: Context, state_path: PathBuf) -> Result<TpmCredStore> {
+    /// Connects to the TPM via the TPM Access Broker Resource Manager Daemon (TABRMD), as
+    /// specified by the provided configuration string.
+    pub fn from_tabrmd(tabrmd_cfg: &str, state_path: PathBuf) -> Result<TpmCredStore> {
+        let context = Context::new(TctiNameConf::Tabrmd(TabrmdConfig::from_str(tabrmd_cfg)?))?;
+        Self::from_context(context, state_path)
+    }
+
+    pub fn from_context(mut context: Context, state_path: PathBuf) -> Result<TpmCredStore> {
         let storage = Storage::load_or_init(&state_path)?;
         let session = context.start_default_auth_session()?;
         context.set_sessions((Some(session), None, None));
@@ -1624,7 +1632,7 @@ mod test {
     fn tpm_cred_store_new() -> Result<()> {
         let harness = SwtpmHarness::new().bterr()?;
         let cookie_path = harness.dir_path().join("cookie.bin");
-        let store = TpmCredStore::new(harness.context()?, cookie_path.to_owned())?;
+        let store = TpmCredStore::from_context(harness.context()?, cookie_path.to_owned())?;
         let cookie = File::open(&cookie_path)?;
         let metadata = cookie.metadata()?;
         let actual = metadata.permissions().mode();
@@ -1638,7 +1646,7 @@ mod test {
     fn gen_creds() -> Result<()> {
         let harness = SwtpmHarness::new().bterr()?;
         let cookie_path = harness.dir_path().join("cookie.bin");
-        let store = TpmCredStore::new(harness.context()?, cookie_path)?;
+        let store = TpmCredStore::from_context(harness.context()?, cookie_path)?;
         store.gen_node_creds()?;
         Ok(())
     }
@@ -1681,7 +1689,8 @@ mod test {
     /// in the returned tuple is significant, as TpmCredStore must be dropped _before_ SwtpmHarness.
     fn test_store() -> Result<(SwtpmHarness, TpmCredStore)> {
         let harness = SwtpmHarness::new().bterr()?;
-        let store = TpmCredStore::new(harness.context()?, harness.state_path().to_owned())?;
+        let store =
+            TpmCredStore::from_context(harness.context()?, harness.state_path().to_owned())?;
         Ok((harness, store))
     }
 
@@ -1760,7 +1769,8 @@ mod test {
             creds.principal()
         };
         drop(store);
-        let store = TpmCredStore::new(harness.context()?, harness.state_path().to_owned())?;
+        let store =
+            TpmCredStore::from_context(harness.context()?, harness.state_path().to_owned())?;
         let creds = store.node_creds()?;
         let actual = creds.principal();
         assert_eq!(expected, actual);
@@ -1835,7 +1845,8 @@ mod test {
             creds.principal()
         };
         drop(store);
-        let store = TpmCredStore::new(harness.context()?, harness.state_path().to_owned())?;
+        let store =
+            TpmCredStore::from_context(harness.context()?, harness.state_path().to_owned())?;
         let creds = store.root_creds(PASSWORD)?;
         let actual = creds.principal();
         assert_eq!(expected, actual);
@@ -1849,7 +1860,8 @@ mod test {
         let (harness, store) = test_store()?;
         store.gen_root_creds("Galileo")?;
         drop(store);
-        let store = TpmCredStore::new(harness.context()?, harness.state_path().to_owned())?;
+        let store =
+            TpmCredStore::from_context(harness.context()?, harness.state_path().to_owned())?;
         let creds = store.root_creds("Figaro")?;
         assert!(sign_verify_test(&creds).is_err());
         assert!(encrypt_decrypt_test(&creds).is_err());

crates/btlib/src/lib.rs

@@ -1,6 +1,8 @@
 pub mod accessor;
 mod block_path;
 pub mod buf_reader;
+pub mod collections;
+pub mod config_helpers;
 /// Code which enables cryptographic operations.
 pub mod crypto;
 pub mod error;
@@ -9,7 +11,6 @@ pub mod sectored_buf;
 #[cfg(test)]
 mod test_helpers;
 mod trailered;
-pub mod collections;
 
 #[macro_use]
 extern crate static_assertions;

crates/btmsg/src/lib.rs

@@ -7,7 +7,7 @@ mod callback_framed;
 use callback_framed::CallbackFramed;
 pub use callback_framed::DeserCallback;
 
-use btlib::{bterr, crypto::Creds, BlockPath, Result, Writecap};
+use btlib::{bterr, crypto::Creds, error::DisplayErr, BlockPath, Result, Writecap};
 use btserde::write_to;
 use bytes::{BufMut, BytesMut};
 use core::{
@@ -24,17 +24,18 @@ use std::{
     hash::Hash,
     marker::PhantomData,
     net::{IpAddr, Ipv6Addr, SocketAddr},
-    sync::Arc,
+    result::Result as StdResult,
+    sync::{Arc, Mutex as StdMutex},
 };
 use tokio::{
-    runtime::Handle,
     select,
-    sync::{broadcast, Mutex, OwnedSemaphorePermit, Semaphore},
+    sync::{broadcast, Mutex},
+    task::{JoinError, JoinHandle},
 };
 use tokio_util::codec::{Encoder, Framed, FramedParts, FramedWrite};
 
 /// Returns a [Receiver] bound to the given [IpAddr] which receives messages at the bind path of
-/// the given [Writecap] of the given credentials. The returned type can be used to make
+/// the [Writecap] in the given credentials. The returned type can be used to make
 /// [Transmitter]s for any path.
 pub fn receiver<C: 'static + Creds + Send + Sync, F: 'static + MsgCallback>(
     ip_addr: IpAddr,
@@ -197,6 +198,13 @@ pub trait Receiver {
 
     /// Creates a [Transmitter] which is connected to the given address.
     fn transmitter(&self, addr: Arc<BlockAddr>) -> Self::TransmitterFut<'_>;
+
+    type CompleteErr: std::error::Error;
+    type CompleteFut<'a>: 'a + Future<Output = StdResult<(), Self::CompleteErr>> + Send
+    where
+        Self: 'a;
+    /// Returns a future which completes when this [Receiver] has completed (which may be never).
+    fn complete(&self) -> Result<Self::CompleteFut<'_>>;
 }
 
 /// A type which can be used to transmit messages.
@@ -411,12 +419,10 @@ struct QuicReceiver {
     stop_tx: broadcast::Sender<()>,
     endpoint: Endpoint,
     resolver: Arc<CertResolver>,
+    join_handle: StdMutex<Option<JoinHandle<()>>>,
 }
 
 impl QuicReceiver {
-    /// This defines the maximum number of blocking tasks which can be spawned at once.
-    const BLOCKING_LIMIT: usize = 16;
-
     fn new<F: 'static + MsgCallback>(
         recv_addr: Arc<BlockAddr>,
         resolver: Arc<CertResolver>,
@@ -425,12 +431,13 @@ impl QuicReceiver {
         let socket_addr = recv_addr.socket_addr()?;
         let endpoint = Endpoint::server(server_config(resolver.clone())?, socket_addr)?;
         let (stop_tx, stop_rx) = broadcast::channel(1);
-        tokio::spawn(Self::server_loop(endpoint.clone(), callback, stop_rx));
+        let join_handle = tokio::spawn(Self::server_loop(endpoint.clone(), callback, stop_rx));
         Ok(Self {
             recv_addr,
             stop_tx,
             endpoint,
             resolver,
+            join_handle: StdMutex::new(Some(join_handle)),
         })
     }
 
@@ -439,7 +446,6 @@ impl QuicReceiver {
         callback: F,
         mut stop_rx: broadcast::Receiver<()>,
     ) {
-        let blocking_permits = Arc::new(Semaphore::new(Self::BLOCKING_LIMIT));
         loop {
             let connecting = await_or_stop!(endpoint.accept(), stop_rx.recv());
             let connection = unwrap_or_continue!(connecting.await, |err| error!(
@@ -449,7 +455,6 @@ impl QuicReceiver {
                 connection,
                 callback.clone(),
                 stop_rx.resubscribe(),
-                blocking_permits.clone(),
             ));
         }
     }
@@ -458,7 +463,6 @@ impl QuicReceiver {
         connection: Connection,
         callback: F,
         mut stop_rx: broadcast::Receiver<()>,
-        blocking_permits: Arc<Semaphore>,
     ) {
         let client_path = unwrap_or_return!(
             Self::client_path(connection.peer_identity()),
@@ -468,20 +472,14 @@ impl QuicReceiver {
             let result = await_or_stop!(connection.accept_bi().map(Some), stop_rx.recv());
             let (send_stream, recv_stream) =
                 unwrap_or_continue!(result, |err| error!("error accepting stream: {err}"));
-            let permit = unwrap_or_continue!(blocking_permits.clone().acquire_owned().await);
             let client_path = client_path.clone();
             let callback = callback.clone();
-            // spawn_blocking is used to allow the user supplied callback to to block without
-            // disrupting the main thread pool.
-            tokio::task::spawn_blocking(move || {
-                Handle::current().block_on(Self::handle_message(
-                    client_path,
-                    send_stream,
-                    recv_stream,
-                    permit,
-                    callback,
-                ))
-            });
+            tokio::task::spawn(Self::handle_message(
+                client_path,
+                send_stream,
+                recv_stream,
+                callback,
+            ));
         }
     }
 
@@ -489,8 +487,6 @@ impl QuicReceiver {
         client_path: Arc<BlockPath>,
         send_stream: SendStream,
         recv_stream: RecvStream,
-        // This argument must be kept alive until this method returns.
-        _permit: OwnedSemaphorePermit,
         callback: F,
     ) {
         let framed_msg = FramedWrite::new(send_stream, MsgEncoder::new());
@@ -546,6 +542,16 @@ impl Receiver for QuicReceiver {
             QuicTransmitter::from_endpoint(self.endpoint.clone(), addr, self.resolver.clone()).await
         })
     }
+
+    type CompleteErr = JoinError;
+    type CompleteFut<'a> = JoinHandle<()>;
+    fn complete(&self) -> Result<Self::CompleteFut<'_>> {
+        let mut guard = self.join_handle.lock().display_err()?;
+        let handle = guard
+            .take()
+            .ok_or_else(|| bterr!("join handle has already been taken"))?;
+        Ok(handle)
+    }
 }
 
 macro_rules! cleanup_on_err {

crates/btserde/src/reader.rs

@@ -5,7 +5,7 @@ use crate::{error::MapError, Error, Result};
 use std::io::Read;
 
 pub trait Reader<'de> {
-    /// Reads exactly the enough bytes to fill the given buffer or returns an error.
+    /// Reads exactly enough bytes to fill the given buffer or returns an error.
     fn read_exact(&mut self, buf: &mut [u8]) -> Result<()>;
     /// Returns true if the `borrow_bytes` method is supported by this instance.
     fn can_borrow() -> bool;