Introduces the confidant CLI

Cargo.toml

@@ -4,6 +4,7 @@ resolver = "2"
 members = [
     "streambed",
     "streambed-confidant",
+    "streambed-confidant-cli",
     "streambed-kafka",
     "streambed-logged",
     "streambed-logged-cli",
@@ -59,5 +60,6 @@ tokio-util = "0.7.4"
 warp = "0.3"
 
 streambed = { path = "streambed", version = "0.10.2" }
+streambed-confidant = { path = "streambed-confidant", version = "0.10.2" }
 streambed-logged = { path = "streambed-logged", version = "0.10.2" }
 streambed-test = { path = "streambed-test", version = "0.10.2" }

streambed-confidant-cli/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "streambed-confidant-cli"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+description = "A CLI for a file-system-based secret store that applies streambed-crypto to data"
+
+[dependencies]
+base64 = { workspace = true }
+clap = { workspace = true }
+env_logger = { workspace = true }
+hex = { workspace = true }
+humantime = { workspace = true }
+rand = { workspace = true }
+serde_json = { workspace = true }
+streambed = { workspace = true }
+streambed-confidant = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+tokio-stream = { workspace = true }
+
+[[bin]]
+name = "confidant"
+path = "src/main.rs"

streambed-confidant-cli/README.md

@@ -0,0 +1,75 @@
+confidant
+===
+
+The `confidant` command provides a utility for conveniently operating on file-based secret store. It is
+often used in conjunction `logged` to encrypt payloads as a pre-stage, or decrypt as a post-stage. No
+assumption is made regarding the nature of a payload beyond it being encrypted using streambed crypto
+functions.
+
+Running with an example encrypt followed by a decrypt
+---
+
+First get the executable:
+
+```
+cargo install streambed-confidant-cli
+```
+
+...or build it from this repo:
+
+```
+cargo build --bin confidant --release
+```
+
+...and make a build available on the PATH (only for `cargo build` and just once per shell session):
+
+```
+export PATH="$PWD/target/release":$PATH
+```
+
+Before you can use `confidant`, you must provide it with a root secret and a "secret id" (password)
+to authenticate the session. Here's an example with some dummy data:
+
+```
+echo "1800af9b273e4b9ea71ec723426933a4" > /tmp/root-secret
+echo "unused-id" > /tmp/ss-secret-id
+```
+
+We also need to create a directory for `confidant` to read and write its secrets. A security feature
+of the `confidant` library is that the directory must have a permission of `600` for the owner user.
+ACLs should then be used to control access for individual processes. Here's how the directory can be
+created:
+
+```
+mkdir /tmp/confidant
+chmod 700 /tmp/confidant
+```
+
+You would normally source the above secrets from your production system, preferably without
+them leaving your production system.
+
+Given the root secret, encrypt some data :
+
+```
+echo '{"value":"SGkgdGhlcmU="}' | \
+  confidant --root-path=/tmp/confidant encrypt --file - --path="default/my-secret-path"
+```
+
+...which would output:
+
+```
+{"value":"EZy4HLnFC4c/W63Qtp288WWFj8U="}
+```
+
+That value is now encrypted with a salt.
+
+We can also decrypt in a similar fashion:
+
+```
+echo '{"value":"EZy4HLnFC4c/W63Qtp288WWFj8U="}' | \
+  confidant --root-path=/tmp/confidant decrypt --file - --path="default/my-secret-path"
+```
+
+...which will yield the original BASE64 value that we encrypted.
+
+Use `--help` to discover all of the options.

streambed-confidant-cli/src/cryptor.rs

@@ -0,0 +1,126 @@
+use std::{
+    collections::HashMap,
+    io::{self, Write},
+    time::Duration,
+};
+
+use rand::RngCore;
+use serde_json::Value;
+use streambed::{
+    crypto::{self, SALT_SIZE},
+    get_secret_value,
+    secret_store::{SecretData, SecretStore},
+};
+use tokio::{sync::mpsc::channel, time};
+
+use crate::errors::Errors;
+
+const FLUSH_DELAY: Duration = Duration::from_millis(100);
+const OUTPUT_QUEUE_SIZE: usize = 10;
+
+pub async fn process_records(
+    ss: impl SecretStore,
+    mut line_reader: impl FnMut() -> Result<Option<String>, io::Error>,
+    mut output: impl Write,
+    path: &str,
+    process: fn(Vec<u8>, Vec<u8>) -> Option<Vec<u8>>,
+    select: &str,
+) -> Result<(), Errors> {
+    let (output_tx, mut output_rx) = channel(OUTPUT_QUEUE_SIZE);
+
+    let processor = async move {
+        while let Some(line) = line_reader()? {
+            let mut record = serde_json::from_str::<Value>(&line).map_err(Errors::from)?;
+            let Some(value) = record.get_mut(select) else {
+                return Err(Errors::CannotSelectValue);
+            };
+            let Some(str_value) = value.as_str() else {
+                return Err(Errors::CannotGetValue);
+            };
+            let Ok(bytes) = base64::decode(str_value) else {
+                return Err(Errors::CannotDecodeValue);
+            };
+            let decrypted_bytes = get_secret_value(&ss, path)
+                .await
+                .and_then(|secret_value| {
+                    let key = hex::decode(secret_value).ok()?;
+                    process(key, bytes)
+                })
+                .ok_or(Errors::CannotDecryptValue)?;
+            let encoded_decrypted_str = base64::encode(decrypted_bytes);
+            *value = Value::String(encoded_decrypted_str);
+            let _ = output_tx.send(record).await;
+        }
+
+        Ok(())
+    };
+
+    let outputter = async move {
+        let mut timeout = FLUSH_DELAY;
+        loop {
+            tokio::select! {
+                record = output_rx.recv() => {
+                    if let Some(record) = record {
+                        let buf = serde_json::to_string(&record)?;
+                        output.write_all(buf.as_bytes())?;
+                        output.write_all(b"\n")?;
+                        timeout = FLUSH_DELAY;
+                    } else {
+                        break;
+                    }
+                }
+                _ = time::sleep(timeout) => {
+                    output.flush()?;
+                    timeout = Duration::MAX;
+                }
+            }
+        }
+        Ok(())
+    };
+
+    tokio::try_join!(processor, outputter).map(|_| ())
+}
+
+pub async fn decrypt(
+    ss: impl SecretStore,
+    line_reader: impl FnMut() -> Result<Option<String>, io::Error>,
+    output: impl Write,
+    path: &str,
+    select: &str,
+) -> Result<(), Errors> {
+    fn process(key: Vec<u8>, mut bytes: Vec<u8>) -> Option<Vec<u8>> {
+        let (salt, data_bytes) = bytes.split_at_mut(crypto::SALT_SIZE);
+        crypto::decrypt(data_bytes, &key.try_into().ok()?, &salt.try_into().ok()?);
+        Some(data_bytes.to_vec())
+    }
+    process_records(ss, line_reader, output, path, process, select).await
+}
+
+pub async fn encrypt(
+    ss: impl SecretStore,
+    line_reader: impl FnMut() -> Result<Option<String>, io::Error>,
+    output: impl Write,
+    path: &str,
+    select: &str,
+) -> Result<(), Errors> {
+    // As a convenience, we create the secret when encrypting if there
+    // isn't one.
+    if get_secret_value(&ss, path).await.is_none() {
+        let mut key = vec![0; 16];
+        rand::thread_rng().fill_bytes(&mut key);
+        let data = HashMap::from([("value".to_string(), hex::encode(key))]);
+        ss.create_secret(path, SecretData { data })
+            .await
+            .map_err(Errors::SecretStore)?;
+    }
+
+    fn process(key: Vec<u8>, mut data_bytes: Vec<u8>) -> Option<Vec<u8>> {
+        let salt = crypto::salt(&mut rand::thread_rng());
+        crypto::encrypt(&mut data_bytes, &key.try_into().ok()?, &salt);
+        let mut buf = Vec::with_capacity(SALT_SIZE + data_bytes.len());
+        buf.extend(salt);
+        buf.extend(data_bytes);
+        Some(buf)
+    }
+    process_records(ss, line_reader, output, path, process, select).await
+}

streambed-confidant-cli/src/errors.rs

@@ -0,0 +1,96 @@
+use std::{
+    error::Error,
+    fmt::{self, Debug},
+    io,
+};
+
+use streambed::secret_store;
+
+#[derive(Debug)]
+pub enum Errors {
+    CannotDecodeRootSecretAsHex,
+    CannotDecodeValue,
+    CannotDecryptValue,
+    CannotEncryptValue,
+    CannotGetValue,
+    CannotSelectValue,
+    EmptyRootSecretFile,
+    EmptySecretIdFile,
+    InvalidRootSecret,
+    InvalidSaltLen,
+    Io(io::Error),
+    RootSecretFileIo(io::Error),
+    SecretIdFileIo(io::Error),
+    SecretStore(secret_store::Error),
+    Serde(serde_json::Error),
+}
+
+impl From<io::Error> for Errors {
+    fn from(value: io::Error) -> Self {
+        Self::Io(value)
+    }
+}
+
+impl From<serde_json::Error> for Errors {
+    fn from(value: serde_json::Error) -> Self {
+        Self::Serde(value)
+    }
+}
+
+impl fmt::Display for Errors {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::CannotDecodeRootSecretAsHex => {
+                f.write_str("Cannot decode the root-secret as hex")
+            }
+            Self::CannotDecodeValue => {
+                f.write_str("Cannot decode the selected value of the JSON object")
+            }
+            Self::CannotDecryptValue => {
+                f.write_str("Cannot decrypt the selected value of the JSON object")
+            }
+            Self::CannotEncryptValue => {
+                f.write_str("Cannot encrypt the selected value of the JSON object")
+            }
+            Self::CannotGetValue => {
+                f.write_str("Cannot get the selected value of the selected field")
+            }
+            Self::CannotSelectValue => {
+                f.write_str("Cannot select the value within the JSON object")
+            }
+            Self::EmptyRootSecretFile => f.write_str("Empty root-secret file"),
+            Self::EmptySecretIdFile => f.write_str("Empty secret-id file"),
+            Self::InvalidRootSecret => {
+                f.write_str("Invalid root secret - must be a hex string of 32 characters")
+            }
+            Self::InvalidSaltLen => f.write_str("Invalid salt length - must be 12 bytes"),
+            Self::Io(e) => fmt::Display::fmt(&e, f),
+            Self::SecretStore(_) => f.write_str("Unauthorised access"),
+            Self::RootSecretFileIo(_) => f.write_str("root-secret file problem"),
+            Self::SecretIdFileIo(_) => f.write_str("secret-id file problem"),
+            Self::Serde(e) => fmt::Display::fmt(&e, f),
+        }
+    }
+}
+
+impl Error for Errors {
+    fn source(&self) -> Option<&(dyn Error + 'static)> {
+        match self {
+            Self::CannotDecodeValue
+            | Self::CannotDecodeRootSecretAsHex
+            | Self::CannotDecryptValue
+            | Self::CannotEncryptValue
+            | Self::CannotGetValue
+            | Self::CannotSelectValue
+            | Self::EmptyRootSecretFile
+            | Self::EmptySecretIdFile
+            | Self::InvalidRootSecret
+            | Self::InvalidSaltLen
+            | Self::SecretStore(_) => None,
+            Self::Io(e) => e.source(),
+            Self::RootSecretFileIo(e) => e.source(),
+            Self::SecretIdFileIo(e) => e.source(),
+            Self::Serde(e) => e.source(),
+        }
+    }
+}