use std::{
collections::BTreeMap,
fs,
io::Read,
os::unix::fs::MetadataExt,
path::{Path, PathBuf},
sync::{Arc, Mutex, OnceLock},
};
use murmur_artifact::ApiKeyReference;
use serde::Deserialize;
use crate::{errors::RuntimeError, trace::ResourceTraceAppender};
pub(crate) const E_RUN_027: &str = "E-RUN-027";
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CredentialSource {
Config { path: PathBuf, name: String },
Environment { name: String },
ManifestLiteral,
}
impl CredentialSource {
pub fn label(&self) -> String {
match self {
Self::Config { path, name } => {
format!("credentials.{name} in {}", path.display())
}
Self::Environment { name } => {
format!("from the environment variable {name}, which is read once at launch")
}
Self::ManifestLiteral => "written literally as inference.api_key in murmur.yaml, \
which is read once at launch"
.to_string(),
}
}
pub fn trace_name(&self) -> &'static str {
match self {
Self::Config { .. } => "config",
Self::Environment { .. } => "environment",
Self::ManifestLiteral => "manifest",
}
}
pub fn credential_name(&self) -> Option<&str> {
match self {
Self::Config { name, .. } | Self::Environment { name } => Some(name),
Self::ManifestLiteral => None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct CredentialRejection {
pub(crate) status: u16,
pub(crate) retried: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CredentialChange {
Rotated { trigger: &'static str },
Rejected { status: u16, retried: bool },
Unreadable { reason: &'static str },
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FileStamp {
Unavailable,
Present {
dev: u64,
ino: u64,
len: u64,
mtime: (i64, i64),
ctime: (i64, i64),
mode: u32,
},
}
impl FileStamp {
fn of(metadata: &fs::Metadata) -> Self {
Self::Present {
dev: metadata.dev(),
ino: metadata.ino(),
len: metadata.len(),
mtime: (metadata.mtime(), metadata.mtime_nsec()),
ctime: (metadata.ctime(), metadata.ctime_nsec()),
mode: metadata.mode(),
}
}
}
#[derive(Deserialize)]
struct CredentialsFile {
#[serde(default)]
credentials: BTreeMap<String, String>,
}
enum EntryRead {
Value(String),
Unreadable(&'static str),
}
fn read_entry(path: &Path, name: &str) -> (FileStamp, EntryRead) {
let mut file = match fs::File::open(path) {
Ok(file) => file,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
return (FileStamp::Unavailable, EntryRead::Unreadable("missing"))
}
Err(_) => return (FileStamp::Unavailable, EntryRead::Unreadable("unreadable")),
};
let stamp = file
.metadata()
.map(|metadata| FileStamp::of(&metadata))
.unwrap_or(FileStamp::Unavailable);
let mut text = String::new();
if file.read_to_string(&mut text).is_err() {
return (stamp, EntryRead::Unreadable("unreadable"));
}
let Ok(parsed) = serde_yaml::from_str::<CredentialsFile>(&text) else {
return (stamp, EntryRead::Unreadable("unparseable"));
};
let read = match parsed.credentials.get(name) {
None => EntryRead::Unreadable("no_entry"),
Some(value) if value.trim().is_empty() => EntryRead::Unreadable("empty_entry"),
Some(value) => EntryRead::Value(value.clone()),
};
(stamp, read)
}
pub(crate) fn config_holds_credential(path: &Path, name: &str) -> bool {
matches!(read_entry(path, name).1, EntryRead::Value(_))
}
struct CredentialState {
value: String,
unreadable_reported_for: Option<(FileStamp, &'static str)>,
pending_rejection: Option<CredentialRejection>,
}
pub(crate) struct InferenceCredential {
source: CredentialSource,
state: Mutex<CredentialState>,
trace: OnceLock<Arc<ResourceTraceAppender>>,
#[cfg(test)]
reads: std::sync::atomic::AtomicUsize,
}
impl std::fmt::Debug for InferenceCredential {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("InferenceCredential")
.field("source", &self.source)
.field("value", &"<redacted>")
.finish()
}
}
impl InferenceCredential {
pub(crate) fn resolve(
reference: &ApiKeyReference,
credentials_file: Option<&Path>,
) -> Result<Self, RuntimeError> {
let name = match reference {
ApiKeyReference::Literal(value) => {
return Ok(Self::new(CredentialSource::ManifestLiteral, value.clone()))
}
ApiKeyReference::Environment(name) => name,
};
if let Some(path) = credentials_file {
let (stamp, read) = read_entry(path, name);
if let EntryRead::Value(value) = read {
if let FileStamp::Present { mode, .. } = stamp {
crate::murmur_home::warn_on_wide_credential_file(path, name, mode);
}
let credential = Self::new(
CredentialSource::Config {
path: path.to_path_buf(),
name: name.clone(),
},
value,
);
credential.count_read();
return Ok(credential);
}
}
match std::env::var(name) {
Ok(value) if !value.is_empty() => Ok(Self::new(
CredentialSource::Environment { name: name.clone() },
value,
)),
_ => Err(RuntimeError::InferenceCredentialNotFound {
variable: name.clone(),
credentials_file: credentials_file.map(Path::to_path_buf),
}),
}
}
fn new(source: CredentialSource, value: String) -> Self {
Self {
source,
state: Mutex::new(CredentialState {
value,
unreadable_reported_for: None,
pending_rejection: None,
}),
trace: OnceLock::new(),
#[cfg(test)]
reads: std::sync::atomic::AtomicUsize::new(0),
}
}
pub(crate) fn source(&self) -> &CredentialSource {
&self.source
}
pub(crate) fn attach_trace(&self, trace: Arc<ResourceTraceAppender>) {
let _ = self.trace.set(trace);
}
pub(crate) async fn current(&self) -> String {
self.refresh(false).await
}
pub(crate) async fn reread_after_rejection(&self) -> String {
self.refresh(true).await
}
async fn refresh(&self, after_rejection: bool) -> String {
let (value, change) = self.refresh_locked(after_rejection);
if let Some(change) = change {
if let CredentialChange::Unreadable { reason } = change {
eprintln!(
"[capsule-runtime] warning: the inference credential {} could not be read \
({reason}); the value read before stays in use",
self.source.label()
);
}
self.write_event(change).await;
}
value
}
fn refresh_locked(&self, after_rejection: bool) -> (String, Option<CredentialChange>) {
let mut state = self.state.lock().unwrap_or_else(|e| e.into_inner());
let CredentialSource::Config { path, name } = &self.source else {
return (state.value.clone(), None);
};
self.count_read();
let (stamp, read) = read_entry(path, name);
let change = match read {
EntryRead::Value(value) => {
state.unreadable_reported_for = None;
if value == state.value {
None
} else {
state.value = value;
Some(CredentialChange::Rotated {
trigger: if after_rejection {
"rejection"
} else {
"file_changed"
},
})
}
}
EntryRead::Unreadable(reason) => {
if state.unreadable_reported_for == Some((stamp, reason)) {
None
} else {
state.unreadable_reported_for = Some((stamp, reason));
Some(CredentialChange::Unreadable { reason })
}
}
};
(state.value.clone(), change)
}
pub(crate) async fn record_rejection(&self, status: u16, retried: bool) {
self.state
.lock()
.unwrap_or_else(|e| e.into_inner())
.pending_rejection = Some(CredentialRejection { status, retried });
self.write_event(CredentialChange::Rejected { status, retried })
.await;
}
pub(crate) fn clear_rejection(&self) {
self.state
.lock()
.unwrap_or_else(|e| e.into_inner())
.pending_rejection = None;
}
pub(crate) fn take_rejection(&self) -> Option<CredentialRejection> {
self.state
.lock()
.unwrap_or_else(|e| e.into_inner())
.pending_rejection
.take()
}
pub(crate) fn rejection_message(&self, rejection: CredentialRejection) -> String {
let mut message = format!(
"the provider rejected the inference credential {} (HTTP {})",
self.source.label(),
rejection.status
);
if rejection.retried {
message.push_str(
"; the credential was re-read after the first rejection and its new value was \
rejected too",
);
}
message
}
pub(crate) fn rejection_hint(&self) -> String {
match &self.source {
CredentialSource::Config { name, .. } => format!(
"replace it with `mur config set -g credentials.{name} <key>`; running capsules \
use it on their next call"
),
other => {
let name = other.credential_name().unwrap_or("<NAME>");
format!(
"this capsule reads its key only at launch; restart it with a valid key, or \
store the key with `mur config set -g credentials.{name} <key>`"
)
}
}
}
pub(crate) fn report_rejection(&self) -> Option<String> {
let rejection = self.take_rejection()?;
let message = self.rejection_message(rejection);
eprintln!("error[{E_RUN_027}]: {message}");
eprintln!(" hint: {}", self.rejection_hint());
Some(message)
}
async fn write_event(&self, change: CredentialChange) {
if let Some(trace) = self.trace.get() {
trace
.write_inference_credential(
self.source.trace_name(),
self.source.credential_name(),
change,
)
.await;
}
}
fn count_read(&self) {
#[cfg(test)]
self.reads.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}
}
#[cfg(test)]
mod tests {
use std::sync::atomic::Ordering;
use super::*;
const OLD: &str = "unit-credential-old-marker";
const NEW: &str = "unit-credential-new-marker";
const NAME: &str = "UNIT_CREDENTIAL_KEY";
fn write_config(path: &Path, value: &str) {
fs::write(
path,
format!("registry:\n default: official\ncredentials:\n {NAME}: {value}\n"),
)
.unwrap();
}
fn replace_config(path: &Path, value: &str) {
let tmp = path.with_file_name(".config.yaml.tmp");
write_config(&tmp, value);
fs::rename(tmp, path).unwrap();
}
fn config_credential(dir: &tempfile::TempDir) -> (PathBuf, InferenceCredential) {
let path = dir.path().join("config.yaml");
write_config(&path, OLD);
let credential = InferenceCredential::resolve(
&ApiKeyReference::Environment(NAME.to_string()),
Some(&path),
)
.unwrap();
(path, credential)
}
fn reads(credential: &InferenceCredential) -> usize {
credential.reads.load(Ordering::SeqCst)
}
fn overwrite_in_place(path: &Path, value: &str) {
let before = fs::metadata(path).unwrap();
let text = fs::read_to_string(path).unwrap().replace(OLD, value);
let mut file = fs::OpenOptions::new().write(true).open(path).unwrap();
std::io::Write::write_all(&mut file, text.as_bytes()).unwrap();
drop(file);
let after = fs::metadata(path).unwrap();
assert_eq!(
(before.dev(), before.ino(), before.len()),
(after.dev(), after.ino(), after.len())
);
}
#[tokio::test]
async fn inference_credential_every_call_reads_the_file() {
let dir = tempfile::tempdir().unwrap();
let (_, credential) = config_credential(&dir);
assert_eq!(credential.source().trace_name(), "config");
for _ in 0..5 {
assert_eq!(credential.current().await, OLD);
}
assert_eq!(reads(&credential), 6);
assert_eq!(credential.refresh_locked(false).1, None);
assert_eq!(reads(&credential), 7);
}
#[tokio::test]
async fn inference_credential_rename_replace_is_seen_on_the_next_call() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
assert_eq!(credential.current().await, OLD);
replace_config(&path, NEW);
assert_eq!(credential.current().await, NEW);
assert_eq!(credential.current().await, NEW);
assert_eq!(reads(&credential), 4);
}
#[tokio::test]
async fn inference_credential_in_place_write_of_a_different_size_is_seen_on_the_next_call() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
write_config(&path, &format!("{NEW}-longer"));
assert_eq!(credential.current().await, format!("{NEW}-longer"));
assert_eq!(reads(&credential), 2);
}
#[tokio::test]
async fn inference_credential_same_size_in_place_edit_is_seen_on_the_next_call() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
assert_eq!(OLD.len(), NEW.len());
assert_eq!(credential.current().await, OLD);
overwrite_in_place(&path, NEW);
assert_eq!(
credential.refresh_locked(false).1,
Some(CredentialChange::Rotated {
trigger: "file_changed"
})
);
assert_eq!(credential.current().await, NEW);
}
#[tokio::test]
async fn inference_credential_torn_read_keeps_the_last_good_value_then_rotates() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
fs::OpenOptions::new()
.write(true)
.truncate(true)
.open(&path)
.unwrap();
assert_eq!(fs::metadata(&path).unwrap().len(), 0);
let (value, change) = credential.refresh_locked(false);
assert_eq!(value, OLD);
assert!(
matches!(change, Some(CredentialChange::Unreadable { .. })),
"{change:?}"
);
assert_eq!(credential.refresh_locked(false), (OLD.to_string(), None));
assert_eq!(credential.current().await, OLD);
assert_eq!(reads(&credential), 4);
write_config(&path, NEW);
assert_eq!(
credential.refresh_locked(false),
(
NEW.to_string(),
Some(CredentialChange::Rotated {
trigger: "file_changed"
})
)
);
assert_eq!(credential.current().await, NEW);
assert_eq!(reads(&credential), 6);
}
#[tokio::test]
async fn inference_credential_read_cost() {
const CALLS: u32 = 10_000;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.yaml");
let filler = "x".repeat(200);
let mut text = format!(
"registry:\n default: official\n url: https://registry.example.invalid/v1\n\
inference:\n endpoint: https://api.example.invalid/v1\n model: test-model\n\
credentials:\n {NAME}: {OLD}-{filler}\n"
);
for index in 1..10 {
text.push_str(&format!(" OTHER_KEY_{index}: other-{index}-{filler}\n"));
}
assert!(text.len() >= 2048, "{}", text.len());
fs::write(&path, text).unwrap();
let credential = InferenceCredential::resolve(
&ApiKeyReference::Environment(NAME.to_string()),
Some(&path),
)
.unwrap();
let started = std::time::Instant::now();
for _ in 0..CALLS {
credential.current().await;
}
let mean = started.elapsed() / CALLS;
println!(
"mean current() cost: {} µs over {CALLS} calls",
mean.as_micros()
);
assert_eq!(reads(&credential), CALLS as usize + 1);
assert!(mean < std::time::Duration::from_millis(5), "{mean:?}");
}
#[tokio::test]
async fn inference_credential_unreadable_source_keeps_the_last_good_value() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
let unreadable = |credential: &InferenceCredential, after_rejection| {
credential.refresh_locked(after_rejection).1
};
fs::remove_file(&path).unwrap();
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Unreadable { reason: "missing" })
);
assert_eq!(unreadable(&credential, false), None);
assert_eq!(unreadable(&credential, true), None);
assert_eq!(credential.current().await, OLD);
std::os::unix::fs::symlink(&path, &path).unwrap();
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Unreadable {
reason: "unreadable"
})
);
assert_eq!(unreadable(&credential, false), None);
fs::remove_file(&path).unwrap();
fs::write(&path, "credentials: [\n").unwrap();
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Unreadable {
reason: "unparseable"
})
);
assert_eq!(unreadable(&credential, false), None);
replace_config(&path, "\"\"");
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Unreadable {
reason: "empty_entry"
})
);
let tmp = path.with_file_name(".config.yaml.tmp");
fs::write(&tmp, "credentials:\n OTHER_KEY: x\n").unwrap();
fs::rename(&tmp, &path).unwrap();
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Unreadable { reason: "no_entry" })
);
assert_eq!(credential.current().await, OLD);
replace_config(&path, NEW);
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Rotated {
trigger: "file_changed"
})
);
assert_eq!(credential.current().await, NEW);
fs::remove_file(&path).unwrap();
assert_eq!(
unreadable(&credential, false),
Some(CredentialChange::Unreadable { reason: "missing" })
);
}
#[tokio::test]
async fn inference_credential_environment_and_literal_sources_never_touch_the_filesystem() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.yaml");
write_config(&path, OLD);
let variable = "UNIT_CREDENTIAL_ENVIRONMENT_ONLY_KEY";
std::env::set_var(variable, NEW);
let environment = InferenceCredential::resolve(
&ApiKeyReference::Environment(variable.to_string()),
Some(&path),
)
.unwrap();
std::env::remove_var(variable);
assert_eq!(
environment.source(),
&CredentialSource::Environment {
name: variable.to_string()
}
);
let literal =
InferenceCredential::resolve(&ApiKeyReference::Literal(OLD.to_string()), Some(&path))
.unwrap();
assert_eq!(literal.source(), &CredentialSource::ManifestLiteral);
fs::remove_file(&path).unwrap();
for (credential, value) in [(&environment, NEW), (&literal, OLD)] {
assert_eq!(credential.current().await, value);
assert_eq!(credential.reread_after_rejection().await, value);
assert_eq!(reads(credential), 0);
}
}
#[test]
fn inference_credential_config_wins_over_the_environment_and_absence_refuses() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.yaml");
write_config(&path, OLD);
std::env::set_var(NAME, NEW);
let credential = InferenceCredential::resolve(
&ApiKeyReference::Environment(NAME.to_string()),
Some(&path),
);
std::env::remove_var(NAME);
assert_eq!(
credential.unwrap().source(),
&CredentialSource::Config {
path: path.clone(),
name: NAME.to_string()
}
);
let missing = "UNIT_CREDENTIAL_NOWHERE_KEY";
let err = InferenceCredential::resolve(
&ApiKeyReference::Environment(missing.to_string()),
Some(&path),
)
.unwrap_err();
let message = err.to_string();
assert!(
message.contains(&format!("credentials.{missing}")),
"{message}"
);
assert!(message.contains(&path.display().to_string()), "{message}");
assert!(message.contains("environment variable"), "{message}");
}
#[tokio::test]
async fn inference_credential_rejection_is_taken_once_and_names_the_source() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
assert_eq!(credential.take_rejection(), None);
credential.record_rejection(401, true).await;
let rejection = credential.take_rejection().unwrap();
assert_eq!(
rejection,
CredentialRejection {
status: 401,
retried: true
}
);
assert_eq!(credential.take_rejection(), None);
let message = credential.rejection_message(rejection);
assert!(
message.contains(&format!("credentials.{NAME}")),
"{message}"
);
assert!(message.contains(&path.display().to_string()), "{message}");
assert!(message.contains("HTTP 401"), "{message}");
credential.record_rejection(401, false).await;
credential.clear_rejection();
assert_eq!(credential.take_rejection(), None);
}
#[tokio::test]
async fn inference_credential_debug_and_labels_carry_no_value() {
let dir = tempfile::tempdir().unwrap();
let (path, credential) = config_credential(&dir);
replace_config(&path, NEW);
credential.current().await;
let literal =
InferenceCredential::resolve(&ApiKeyReference::Literal(OLD.to_string()), None).unwrap();
for credential in [&credential, &literal] {
let rejection = CredentialRejection {
status: 401,
retried: true,
};
for text in [
format!("{credential:?}"),
credential.source().label(),
credential.rejection_message(rejection),
credential.rejection_hint(),
] {
assert!(!text.contains(OLD) && !text.contains(NEW), "{text}");
}
}
}
}