#![forbid(unsafe_code)]
#![deny(missing_docs)]
use std::time::Duration;
use dynamic_config::{Error, Fetched, Format, RemoteSink, RemoteSource, Watching};
use dynamic_config_store_core::attempts::Attempts;
use dynamic_config_store_core::credential::Issued;
use dynamic_config_store_core::documents::{self, Overlap};
use dynamic_config_store_core::guarded;
pub mod auth;
mod tls;
pub use auth::Auth;
use auth::{Session, Token};
pub use dynamic_config_store_core::tls::TlsConfig;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Keys {
One(String),
Several(Vec<String>),
}
impl Keys {
#[must_use]
pub fn one(path: impl Into<String>) -> Self {
Self::One(path.into())
}
#[must_use]
pub fn several<I, S>(paths: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
Self::Several(paths.into_iter().map(Into::into).collect())
}
fn named(&self) -> &[String] {
match self {
Self::One(path) => std::slice::from_ref(path),
Self::Several(paths) => paths,
}
}
fn describe(&self) -> String {
match self {
Self::One(path) => path.clone(),
Self::Several(paths) => format!("paths {}", paths.join(", ")),
}
}
}
impl From<&str> for Keys {
fn from(path: &str) -> Self {
Self::one(path)
}
}
impl From<String> for Keys {
fn from(path: String) -> Self {
Self::One(path)
}
}
impl From<&String> for Keys {
fn from(path: &String) -> Self {
Self::one(path)
}
}
enum CallError {
Forbidden(Error),
Other(Error),
}
impl CallError {
fn into_error(self) -> Error {
match self {
Self::Forbidden(error) | Self::Other(error) => error,
}
}
}
enum CheckError {
NotKv2(Error),
Transient(Error),
}
pub struct Vault {
address: String,
mount: String,
keys: Keys,
key: String,
auth: Auth,
session: Session,
namespace: Option<String>,
timeout: Duration,
agent: Option<ureq::Agent>,
tls: Option<TlsConfig>,
default_agent: std::sync::OnceLock<Result<ureq::Agent, String>>,
attempts: Attempts,
}
impl Vault {
pub fn new(
address: impl Into<String>,
mount: impl Into<String>,
path: impl Into<Keys>,
) -> Self {
Self {
address: address.into().trim_end_matches('/').to_owned(),
mount: mount.into(),
keys: path.into(),
key: "db".to_owned(),
auth: Auth::Token(String::new()),
session: Session::new(),
namespace: None,
timeout: DEFAULT_TIMEOUT,
agent: None,
tls: None,
default_agent: std::sync::OnceLock::new(),
attempts: Attempts::default(),
}
}
#[must_use]
pub fn with_key(mut self, key: impl Into<String>) -> Self {
self.key = key.into();
self
}
#[must_use]
pub fn with_token(self, token: impl Into<String>) -> Self {
self.with_auth(Auth::token(token))
}
#[must_use]
pub fn with_auth(mut self, auth: Auth) -> Self {
self.auth = auth;
self.session.invalidate();
self
}
#[must_use]
pub fn with_agent(mut self, agent: ureq::Agent) -> Self {
self.agent = Some(agent);
self
}
#[must_use]
pub fn with_tls(mut self, tls: TlsConfig) -> Self {
self.tls = Some(tls);
self.default_agent = std::sync::OnceLock::new();
self
}
#[must_use]
pub fn with_namespace(mut self, namespace: impl Into<String>) -> Self {
self.namespace = Some(namespace.into());
self
}
#[must_use]
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self.default_agent = std::sync::OnceLock::new();
self
}
#[must_use]
pub fn reporting_to(mut self, sink: RemoteSink) -> Self {
self.attempts = Attempts::to(sink);
self
}
pub fn watch<F>(
&self,
watching: &Watching,
interval: Duration,
mut on_change: F,
) -> Result<(), Error>
where
F: FnMut(Fetched) -> Result<(), Error>,
{
self.single_path()?;
let mut seen: Option<u64> = None;
while watching.keep_going() {
match self.current_version() {
Ok(version) if seen.is_none() => seen = Some(version),
Ok(version) if seen != Some(version) => {
match self.read() {
Ok((document, version)) => {
seen = Some(version);
guarded(&mut on_change, document, &self.describe())?;
}
Err(error) => self.attempts.failed(&error),
}
}
Err(CheckError::NotKv2(error)) => {
self.attempts.failed(&error);
return Err(error);
}
Err(CheckError::Transient(error)) => self.attempts.failed(&error),
Ok(_) => {}
}
watching.sleep_for(interval);
}
Ok(())
}
fn single_path(&self) -> Result<&str, Error> {
match &self.keys {
Keys::One(path) => Ok(path),
Keys::Several(_) => Err(Error::remote(format!(
"{}: a source that reads several paths cannot be watched; \
poll `refresh_remote()` on a timer instead",
self.describe()
))),
}
}
fn overlap(&self) -> Overlap {
Overlap::LaterWins
}
fn current_version(&self) -> Result<u64, CheckError> {
let path = self.single_path().map_err(CheckError::Transient)?;
let body = self
.get(&self.metadata_url(path), "metadata")
.map_err(CheckError::Transient)?;
body.get("data")
.and_then(|data| data.get("current_version"))
.and_then(serde_json::Value::as_u64)
.ok_or_else(|| {
CheckError::NotKv2(Error::remote(format!(
"{}: the metadata has no `data.current_version`; is this a KV v2 mount?",
self.describe()
)))
})
}
fn get(&self, url: &str, what: &str) -> Result<serde_json::Value, Error> {
match self.get_once(url, what) {
Err(CallError::Forbidden(_)) if self.can_relogin() => {
self.session.invalidate();
self.get_once(url, what).map_err(CallError::into_error)
}
outcome => outcome.map_err(CallError::into_error),
}
}
fn can_relogin(&self) -> bool {
!matches!(self.auth, Auth::Token(_))
}
fn get_once(&self, url: &str, what: &str) -> Result<serde_json::Value, CallError> {
let token = self.token().map_err(CallError::Other)?;
let agent = self.agent().map_err(CallError::Other)?;
let mut request = agent.get(url).header("X-Vault-Token", &token);
if let Some(namespace) = &self.namespace {
request = request.header("X-Vault-Namespace", namespace);
}
request
.call()
.map_err(|error| {
let described = format!("{}: {error}", self.describe());
match error {
ureq::Error::StatusCode(403) => CallError::Forbidden(Error::auth(described)),
_ => CallError::Other(Error::remote(described)),
}
})?
.body_mut()
.read_json()
.map_err(|error| {
CallError::Other(Error::remote(format!(
"{}: the {what} response was not JSON: {error}",
self.describe()
)))
})
}
fn token(&self) -> Result<String, Error> {
if let Auth::Token(supplied) = &self.auth {
if supplied.is_empty() {
return Err(Error::auth(format!(
"{}: no credentials; call `with_token` or `with_auth`",
self.describe()
)));
}
return Ok(supplied.clone());
}
self.session
.token(|| self.login(), |token| self.renew(token))
}
fn login(&self) -> Result<Issued<Token>, Error> {
let Some(path) = self.auth.path() else {
return Err(Error::remote(format!(
"{}: {} needs no login",
self.describe(),
self.auth.describe()
)));
};
let body = self.auth.body()?;
let url = format!("{}/v1/{path}", self.address);
let mut request = self.agent()?.post(&url);
if let Some(namespace) = &self.namespace {
request = request.header("X-Vault-Namespace", namespace);
}
let response: serde_json::Value = request
.send_json(&body)
.map_err(|error| {
let described = format!(
"{}: logging in with {} failed: {error}",
self.describe(),
self.auth.describe()
);
match error {
ureq::Error::StatusCode(400 | 403) => Error::auth(described),
_ => Error::remote(described),
}
})?
.body_mut()
.read_json()
.map_err(|error| {
Error::remote(format!(
"{}: the login response was not JSON: {error}",
self.describe()
))
})?;
self.token_from(&response, "auth")
}
fn renew(&self, token: &str) -> Result<Issued<Token>, Error> {
let url = format!("{}/v1/auth/token/renew-self", self.address);
let mut request = self.agent()?.post(&url).header("X-Vault-Token", token);
if let Some(namespace) = &self.namespace {
request = request.header("X-Vault-Namespace", namespace);
}
let response: serde_json::Value = request
.send_json(serde_json::json!({}))
.map_err(|error| {
Error::remote(format!("{}: renewal failed: {error}", self.describe()))
})?
.body_mut()
.read_json()
.map_err(|error| {
Error::remote(format!(
"{}: the renewal response was not JSON: {error}",
self.describe()
))
})?;
let mut renewed = self.token_from(&response, "auth")?;
renewed.value.secret = token.to_owned();
Ok(renewed)
}
fn token_from(
&self,
response: &serde_json::Value,
field: &str,
) -> Result<Issued<Token>, Error> {
let auth = response.get(field).ok_or_else(|| {
Error::remote(format!(
"{}: the response has no `{field}` block",
self.describe()
))
})?;
let secret = auth
.get("client_token")
.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.to_owned();
let lease = auth
.get("lease_duration")
.and_then(serde_json::Value::as_u64)
.filter(|seconds| *seconds > 0)
.map(Duration::from_secs);
let renewable = auth
.get("renewable")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
Ok(Issued {
value: Token::new(secret, renewable),
ttl: lease,
})
}
fn agent(&self) -> Result<&ureq::Agent, Error> {
if let Some(agent) = &self.agent {
if self.tls.is_some() {
return Err(Error::remote(format!(
"{}: `with_agent` and `with_tls` were both called; \
an agent already carries its own TLS configuration, so \
this is refused rather than resolved — put the certificate \
authority on the agent, or drop the agent",
self.describe()
)));
}
return Ok(agent);
}
self.default_agent
.get_or_init(|| match &self.tls {
Some(tls) => tls::agent(tls, self.timeout, &self.describe())
.map_err(|error| error.to_string()),
None => Ok(ureq::Agent::config_builder()
.timeout_global(Some(self.timeout))
.build()
.new_agent()),
})
.as_ref()
.map_err(Error::remote)
}
fn url(&self, path: &str) -> String {
format!(
"{}/v1/{}/data/{}",
self.address,
self.mount,
path.trim_start_matches('/')
)
}
fn metadata_url(&self, path: &str) -> String {
format!(
"{}/v1/{}/metadata/{}",
self.address,
self.mount,
path.trim_start_matches('/')
)
}
}
impl Vault {
fn read(&self) -> Result<(Fetched, u64), Error> {
let path = self.single_path()?;
let (document, version) = self.read_one(path)?;
Ok((Fetched::new(document, Format::Json), version))
}
fn read_one(&self, path: &str) -> Result<(String, u64), Error> {
let body = self.get(&self.url(path), "secret")?;
let values = body
.get("data")
.and_then(|data| data.get("data"))
.ok_or_else(|| {
Error::remote(format!(
"{}: `{path}` answered without `data.data`; is this a KV v2 mount?",
self.describe()
))
})?;
let document = serde_json::json!({ &self.key: values });
let version = body
.get("data")
.and_then(|data| data.get("metadata"))
.and_then(|metadata| metadata.get("version"))
.and_then(serde_json::Value::as_u64)
.unwrap_or(0);
Ok((document.to_string(), version))
}
fn documents(&self) -> Result<Vec<(String, String)>, Error> {
let paths = self.keys.named();
let mut documents = Vec::with_capacity(paths.len());
for path in paths {
documents.push((path.clone(), self.read_one(path)?.0));
}
Ok(documents)
}
}
impl std::fmt::Debug for Vault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Vault")
.field("address", &self.address)
.field("mount", &self.mount)
.field("keys", &self.keys)
.field("key", &self.key)
.field("namespace", &self.namespace)
.field("auth", &self.auth)
.finish_non_exhaustive()
}
}
impl RemoteSource for Vault {
fn fetch(&self) -> Result<Fetched, Error> {
let documents = self.documents()?;
documents::merged(&documents, Format::Json, self.overlap(), &self.describe())
}
fn describe(&self) -> String {
format!(
"vault {} {}/{}",
self.address,
self.mount,
self.keys.describe()
)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn debug_never_prints_a_credential() {
let source = Vault::new("http://vault:8200", "secret", "myapp/db")
.with_auth(Auth::app_role("hunter2-role-id", "hunter2-secret-id"));
let printed = format!(
"{source:?} {:?} {:?}",
Auth::token("hunter2-token"),
Auth::userpass("admin", "hunter2-password"),
);
assert!(!printed.contains("hunter2-secret-id"), "{printed}");
assert!(!printed.contains("hunter2-token"), "{printed}");
assert!(!printed.contains("hunter2-password"), "{printed}");
assert!(printed.contains("hunter2-role-id"), "{printed}");
assert!(printed.contains("admin"), "{printed}");
}
}