pub use microvms_domain::names::*;
use crate::error::{Error, ErrorKind};
use crate::region::Region;
use crate::session::Session;
pub trait NameStore: Send + Sync {
fn get(&self, name: &str) -> Result<Option<NameRecord>, Error>;
fn put(&self, record: &NameRecord) -> Result<(), Error>;
fn delete(&self, name: &str) -> Result<bool, Error>;
fn list(&self) -> Result<Vec<NameRecord>, Error>;
fn describe(&self) -> String {
"the name store".to_string()
}
fn release_by_vm(&self, microvm_id: &str) -> Result<Vec<String>, Error> {
let mut released = Vec::new();
for record in self.list()? {
if record.microvm_id == microvm_id && self.delete(&record.name)? {
released.push(record.name);
}
}
released.sort();
Ok(released)
}
}
pub fn resolve(
store: &dyn NameStore,
name: &str,
expected_region: Option<&Region>,
) -> Result<NameRecord, Error> {
resolve_record(name, store.get(name)?, &store.describe(), expected_region)
}
pub const IMPORT_PROBE_PATH: &str = "/dev/null";
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum NameHolder {
Free,
Same,
Other(Option<String>),
}
pub fn holder(store: &dyn NameStore, record: &NameRecord) -> NameHolder {
match store.get(&record.name) {
Ok(None) => NameHolder::Free,
Ok(Some(existing)) if existing.microvm_id == record.microvm_id => NameHolder::Same,
Ok(Some(existing)) => NameHolder::Other(Some(existing.microvm_id)),
Err(_) => NameHolder::Other(None),
}
}
pub struct Probed {
microvm_id: String,
endpoint: String,
agent_token: String,
}
impl std::fmt::Debug for Probed {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Probed")
.field("microvm_id", &self.microvm_id)
.field("endpoint", &self.endpoint)
.finish_non_exhaustive()
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct Imported {
pub replaced: bool,
}
pub async fn probe(session: &Session, record: &NameRecord) -> Result<Probed, Error> {
if session.endpoint() != record.endpoint || session.agent_token() != record.agent_token {
return Err(Error::invalid_arg(format!(
"the session isn't attached with the record for {:?}: a probe proves the token it's sent with, so attach with the record's own endpoint and token",
record.name
)));
}
session.file_exists(IMPORT_PROBE_PATH).await?;
Ok(Probed {
microvm_id: record.microvm_id.clone(),
endpoint: record.endpoint.clone(),
agent_token: record.agent_token.clone(),
})
}
pub fn import_probed(
store: &dyn NameStore,
record: &NameRecord,
probed: Probed,
) -> Result<Imported, Error> {
if probed.microvm_id != record.microvm_id
|| probed.endpoint != record.endpoint
|| probed.agent_token != record.agent_token
{
return Err(Error::invalid_arg(format!(
"the probe was of {}, not the record for {:?}: an import writes only the triple that answered",
probed.microvm_id, record.name
)));
}
let record = NameRecord::from_json(record.to_json())?;
let replaced = match holder(store, &record) {
NameHolder::Free => false,
NameHolder::Same => true,
NameHolder::Other(holder) => return Err(taken(store, &record, holder)),
};
store.put(&record)?;
Ok(Imported { replaced })
}
pub async fn import(
store: &dyn NameStore,
record: &NameRecord,
session: &Session,
) -> Result<Imported, Error> {
if let NameHolder::Other(holder) = holder(store, record) {
return Err(taken(store, record, holder));
}
let probed = probe(session, record).await?;
import_probed(store, record, probed)
}
fn taken(store: &dyn NameStore, record: &NameRecord, holder: Option<String>) -> Error {
Error::new(
ErrorKind::Precondition,
format!(
"{:?} is already registered in {} to {}, and this import names {}: a name is a promise about which VM answers to it, and nothing overwrites one silently",
record.name,
store.describe(),
holder.as_deref().unwrap_or("an unreadable record"),
record.microvm_id,
),
)
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};
use super::*;
use crate::session::testing::{Recorder, Reply, session_with};
#[derive(Default)]
struct Memory {
records: Mutex<BTreeMap<String, NameRecord>>,
torn: Vec<String>,
}
impl NameStore for Memory {
fn get(&self, name: &str) -> Result<Option<NameRecord>, Error> {
if self.torn.iter().any(|torn| torn == name) {
return Err(Error::new(ErrorKind::Precondition, "torn"));
}
Ok(self.records.lock().expect("unpoisoned").get(name).cloned())
}
fn put(&self, record: &NameRecord) -> Result<(), Error> {
self.records
.lock()
.expect("unpoisoned")
.insert(record.name.clone(), record.clone());
Ok(())
}
fn delete(&self, name: &str) -> Result<bool, Error> {
Ok(self
.records
.lock()
.expect("unpoisoned")
.remove(name)
.is_some())
}
fn list(&self) -> Result<Vec<NameRecord>, Error> {
Ok(self
.records
.lock()
.expect("unpoisoned")
.values()
.cloned()
.collect())
}
}
fn record(name: &str, microvm_id: &str) -> NameRecord {
NameRecord::new_at(
name,
microvm_id,
"https://vm.example",
"agent-token-abcdef",
"us-east-1",
1,
)
.expect("a record")
}
fn answered(status: u16) -> Arc<Recorder> {
Recorder::with([Reply::Body(status, Vec::new())])
}
#[tokio::test]
async fn an_import_writes_only_after_the_vm_answers_with_the_records_token() {
let store = Memory::default();
let (refusing, _, _) = session_with(answered(401));
let error = import(&store, &record("ci", "mvm-1"), &refusing)
.await
.expect_err("a refused token");
assert_eq!(error.kind(), ErrorKind::Credentials, "{error}");
assert!(store.get("ci").expect("reads").is_none(), "nothing written");
let (answering, _, _) = session_with(answered(200));
let imported = import(&store, &record("ci", "mvm-1"), &answering)
.await
.expect("an answered probe");
assert_eq!(imported, Imported { replaced: false });
assert_eq!(
store.get("ci").expect("reads").map(|r| r.microvm_id),
Some("mvm-1".into())
);
let (again, _, _) = session_with(answered(404));
let refreshed = import(&store, &record("ci", "mvm-1"), &again)
.await
.expect("absence answers behind the same token check");
assert_eq!(refreshed, Imported { replaced: true });
}
#[tokio::test]
async fn an_import_refuses_a_taken_name_before_any_request() {
let store = Memory {
torn: vec!["torn".into()],
..Memory::default()
};
store.put(&record("ci", "mvm-holder")).expect("seeded");
for name in ["ci", "torn"] {
let recorder = Recorder::with([]);
let (session, _, _) = session_with(Arc::clone(&recorder));
let error = import(&store, &record(name, "mvm-new"), &session)
.await
.expect_err("a taken name");
assert_eq!(error.kind(), ErrorKind::Precondition, "{name}: {error}");
assert!(recorder.requests().is_empty(), "{name}: no request");
}
assert_eq!(
store.get("ci").expect("reads").map(|r| r.microvm_id),
Some("mvm-holder".into())
);
}
#[tokio::test]
async fn a_probe_is_of_one_record_and_its_write_takes_only_that_one() {
let (session, _, _) = session_with(answered(200));
let mut stranger = record("ci", "mvm-1");
stranger.agent_token = "another-token".into();
let error = probe(&session, &stranger).await.expect_err("another token");
assert_eq!(error.kind(), ErrorKind::InvalidArg, "{error}");
let mut other_endpoint = record("ci", "mvm-1");
other_endpoint.endpoint = "https://other.example".into();
let mut other_token = record("ci", "mvm-1");
other_token.agent_token = "another-token".into();
for (row, other) in [
("the probe of another VM", record("ci", "mvm-2")),
("another endpoint", other_endpoint),
("another token", other_token),
] {
let (answering, _, _) = session_with(answered(200));
let probed = probe(&answering, &record("ci", "mvm-1"))
.await
.expect("probed");
let store = Memory::default();
let error = import_probed(&store, &other, probed).expect_err(row);
assert_eq!(error.kind(), ErrorKind::InvalidArg, "{row}: {error}");
assert!(
store.get("ci").expect("reads").is_none(),
"{row}: nothing written"
);
}
}
#[tokio::test]
async fn a_probes_debug_names_the_vm_and_redacts_the_token() {
let (session, _, _) = session_with(answered(200));
let probed = probe(&session, &record("ci", "mvm-1"))
.await
.expect("probed");
let shown = format!("{probed:?}");
assert!(shown.contains("mvm-1"), "{shown}");
assert!(!shown.contains("agent-token-abcdef"), "{shown}");
}
}