ms2k/lazaret
SHA256
Watch
0
Code Releases 3 Activity

Decouple client contracts and close private role registries

Browse source
This commit is contained in:
ms2k 2026-07-19 03:44:30 +09:00
commit 22325aa76a
15 changed files with 1041 additions and 321 deletions
Download patch file Download diff file Expand all files Collapse all files

3
crates/lazaret-host-coordinator/Cargo.toml
View file

@ -22,7 +22,9 @@ harness = false
[dependencies]
ed25519-dalek.workspace = true
getrandom.workspace = true
lazaret-control-client = { version = "=0.0.0", path = "../lazaret-control-client" }
lazaret-core = { version = "=0.0.0", path = "../lazaret-core" }
lazaret-domain-state = { version = "=0.0.0", path = "../lazaret-domain-state" }
lazaret-ignored-workspace = { version = "=0.0.0", path = "../lazaret-ignored-workspace" }
lazaret-linux = { version = "=0.0.0", path = "../lazaret-linux" }
lazaret-prepared-lower = { version = "=0.0.0", path = "../lazaret-prepared-lower" }
@ -40,6 +42,7 @@ tokio.workspace = true
[dev-dependencies]
tempfile = "=3.27.0"
lazaret-client = { version = "=0.0.0", path = "../lazaret-client" }
lazaret-escalation-client = { version = "=0.0.0", path = "../lazaret-escalation-client" }
x509-parser.workspace = true

2
crates/lazaret-host-coordinator/src/client_bootstrap.rs
View file

@ -28,7 +28,7 @@ use crate::escalation::{EscalationService, EscalationServiceHandle};
use crate::runtime_selector::{InstalledProductPaths, select_runtime};
use crate::waiter::BootstrapWaiterFace;
pub const PRIVATE_CLIENT_BOOTSTRAP_ROLE: &str = "--lazaret-private-client-bootstrap-v1";
pub use lazaret_core::protocol::PRIVATE_CLIENT_BOOTSTRAP_ROLE;
const MAX_CONCURRENT_CONNECTIONS: usize = 32;
#[derive(Debug)]

57
crates/lazaret-host-coordinator/src/control.rs
View file

@ -10,6 +10,8 @@ use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};
use ed25519_dalek::SigningKey;
pub use lazaret_control_client::{CONTROL_ALPN, CONTROL_SERVER_NAME, PendingClientEnrollment};
use lazaret_control_client::{ControlClient, ControlClientError};
use lazaret_core::codec::{CodecError, decode, encode};
use lazaret_core::configuration::ConfinedPathPlan;
use lazaret_core::identity::{
@ -38,7 +40,7 @@ use rcgen::{
use rustix::fs::{Mode, OFlags, fchmod, openat};
use rustix::net::sockopt::socket_peercred;
use rustls::pki_types::{
CertificateDer, CertificateSigningRequestDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName,
CertificateDer, CertificateSigningRequestDer, PrivateKeyDer, PrivatePkcs8KeyDer,
};
use rustls::server::{NoServerSessionStorage, WebPkiClientVerifier};
use rustls::{ClientConfig, ClientConnection, RootCertStore, ServerConfig, ServerConnection};
@ -53,12 +55,9 @@ use lazaret_linux::filesystem::mount_identity;
use lazaret_linux::process::{ProcessExit, TargetCommand, child_broker};
use lazaret_runtime_supervisor::sandbox::RunscCommand;
pub const CONTROL_ALPN: &[u8] = b"lazaret-control/1";
pub const CONTROL_SERVER_NAME: &str = "coordinator.lazaret.invalid";
pub const MAX_ENROLLMENT_BYTES: usize = 8_192;
const MAX_CONTROL_BYTES: usize = 65_536;
pub(crate) const CONTROL_SOCKET_NAME: &str = "c.sock";
pub(crate) const PUBLICATION_IDENTITY_NAME: &str = "protocol-identity.postcard";
pub(crate) use lazaret_control_client::{CONTROL_SOCKET_NAME, PUBLICATION_IDENTITY_NAME};
pub(crate) const STAGED_CONTROL_SOCKET_NAME: &str = ".c.staged";
pub(crate) const STAGED_PUBLICATION_IDENTITY_NAME: &str = ".protocol-identity.staged";
@ -69,6 +68,7 @@ pub enum ControlError {
Certificate(rcgen::Error),
Tls(rustls::Error),
Protocol(ProtocolValidationError),
Client(ControlClientError),
DomainStore(DomainStoreError),
InvalidPath(&'static str),
InvalidPublication(&'static str),
@ -90,6 +90,7 @@ impl fmt::Display for ControlError {
Self::Certificate(error) => write!(formatter, "certificate operation failed: {error}"),
Self::Tls(error) => write!(formatter, "TLS failed: {error}"),
Self::Protocol(error) => write!(formatter, "protocol validation failed: {error}"),
Self::Client(error) => write!(formatter, "{error}"),
Self::DomainStore(error) => write!(formatter, "session persistence failed: {error}"),
Self::InvalidPath(reason) => write!(formatter, "invalid control path: {reason}"),
Self::InvalidPublication(reason) => {
@ -124,6 +125,7 @@ impl std::error::Error for ControlError {
Self::Certificate(error) => Some(error),
Self::Tls(error) => Some(error),
Self::Protocol(error) => Some(error),
Self::Client(error) => Some(error),
Self::DomainStore(error) => Some(error),
_ => None,
}
@ -145,6 +147,7 @@ from_error!(CodecError, Codec);
from_error!(rcgen::Error, Certificate);
from_error!(rustls::Error, Tls);
from_error!(ProtocolValidationError, Protocol);
from_error!(ControlClientError, Client);
from_error!(DomainStoreError, DomainStore);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
@ -162,50 +165,6 @@ impl FileIdentity {
}
}
pub struct PendingClientEnrollment {
request: ClientEnrollmentRequest,
private_key: PrivatePkcs8KeyDer<'static>,
}
impl PendingClientEnrollment {
/// Creates an Ed25519 CSR whose signed subject binds the exact generation and Client CLI ID.
///
/// # Errors
/// Returns an error when key generation, encoding, or CSR signing fails.
pub fn create(
generation_identity: GenerationIdentity,
cli_id: ClientId,
) -> Result<Self, ControlError> {
let binding = enrollment_binding(&generation_identity, cli_id)?;
let mut params = CertificateParams::default();
params.distinguished_name = DistinguishedName::new();
params.distinguished_name.push(DnType::CommonName, binding);
params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
params.extended_key_usages = vec![ExtendedKeyUsagePurpose::ClientAuth];
let key = KeyPair::generate_for(&PKCS_ED25519)?;
let csr = params.serialize_request(&key)?;
Ok(Self {
request: ClientEnrollmentRequest {
protocol_version: ProtocolVersion::V1,
generation_identity,
cli_id,
certificate_signing_request_der: csr.der().to_vec(),
},
private_key: PrivatePkcs8KeyDer::from(key.serialize_der()),
})
}
#[must_use]
pub const fn request(&self) -> &ClientEnrollmentRequest {
&self.request
}
#[must_use]
pub fn private_key(&self) -> PrivateKeyDer<'static> {
PrivateKeyDer::Pkcs8(self.private_key.clone_key())
}
}
pub struct TrustAnchor {
generation_identity: GenerationIdentity,
tls_ca: CertifiedIssuer<'static, KeyPair>,

52
crates/lazaret-host-coordinator/src/control/publication.rs
View file

@ -479,30 +479,7 @@ impl PublishedControlFace {
publication: &ProtocolPublicationIdentity,
client: Option<(&TlsCredential, PrivateKeyDer<'static>)>,
) -> Result<ClientConfig, ControlError> {
let mut roots = RootCertStore::empty();
roots.add(CertificateDer::from(
publication.tls_ca_certificate_der.clone(),
))?;
let builder = ClientConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
.with_root_certificates(roots);
let mut config = if let Some((credential, private_key)) = client {
verify_tls_credential(
credential,
&publication.generation_identity,
TlsRole::ClientCli,
credential.subject_id(),
)?;
builder.with_client_auth_cert(
vec![CertificateDer::from(credential.certificate_der.clone())],
private_key,
)?
} else {
builder.with_no_client_auth()
};
config.alpn_protocols = vec![CONTROL_ALPN.to_vec()];
config.enable_early_data = false;
config.resumption = rustls::client::Resumption::disabled();
Ok(config)
ControlClient::config(publication, client).map_err(ControlError::from)
}
/// Connects a rustls stream to the fixed control socket using an existing configuration.
@ -515,24 +492,7 @@ impl PublishedControlFace {
publication: &ProtocolPublicationIdentity,
) -> Result<rustls::StreamOwned<ClientConnection, std::os::unix::net::UnixStream>, ControlError>
{
let socket = std::os::unix::net::UnixStream::connect(socket_path)?;
let server_name = ServerName::try_from(CONTROL_SERVER_NAME)
.map_err(|_| ControlError::InvalidCertificateProfile)?;
let mut connection = ClientConnection::new(Arc::new(config), server_name)?;
let mut socket = socket;
while connection.is_handshaking() {
connection.complete_io(&mut socket)?;
}
let [certificate] = connection.peer_certificates().unwrap_or_default() else {
return Err(ControlError::InvalidCertificateProfile);
};
if connection.protocol_version() != Some(rustls::ProtocolVersion::TLSv1_3)
|| connection.alpn_protocol() != Some(CONTROL_ALPN)
|| certificate_digest(certificate) != publication.coordinator_tls_certificate_sha256
{
return Err(ControlError::InvalidCertificateProfile);
}
Ok(rustls::StreamOwned::new(connection, socket))
ControlClient::connect(socket_path, config, publication).map_err(ControlError::from)
}
/// Writes one length-delimited control record.
@ -543,7 +503,7 @@ impl PublishedControlFace {
stream: &mut impl Write,
request: &Request,
) -> Result<(), ControlError> {
write_frame(stream, request)
ControlClient::send(stream, request).map_err(ControlError::from)
}
/// Reads one length-delimited control record.
@ -553,7 +513,7 @@ impl PublishedControlFace {
pub fn receive<Response: serde::de::DeserializeOwned + serde::Serialize>(
stream: &mut impl Read,
) -> Result<Response, ControlError> {
read_frame(stream, MAX_CONTROL_BYTES)
ControlClient::receive(stream).map_err(ControlError::from)
}
}
@ -696,8 +656,8 @@ fn enrollment_binding(
generation_identity: &GenerationIdentity,
cli_id: ClientId,
) -> Result<String, ControlError> {
let encoded = encode(&(generation_identity, cli_id))?;
Ok(format!("lazaret-client-{}", hex(&sha256(&encoded))))
lazaret_core::protocol::client_enrollment_binding(generation_identity, cli_id)
.map_err(ControlError::from)
}
fn certificate_digest(certificate: impl AsRef<[u8]>) -> CertificateDigest {

900
crates/lazaret-host-coordinator/src/domain_store.rs
View file

@ -1,900 +0,0 @@
use std::fmt;
use std::fs::File;
use std::io;
use std::path::Path;
use lazaret_core::codec::{CodecError, decode, encode};
use lazaret_core::identity::{GenerationId, SessionId};
use lazaret_core::records::{
Domain, DomainStore, DomainStoreMetadata, DurableDecision, Generation, Mailbox, Participant,
ParticipantRole, ParticipantState, RecordTransitionError, Session, SessionPhase,
SessionTerminal, SessionTerminalState,
};
use lazaret_linux::filesystem::ExclusiveFileLock;
use redb::{
CommitError, Database, DatabaseError, ReadableDatabase, ReadableTable, StorageError,
TableDefinition, TableError, TransactionError,
};
use serde::{Deserialize, Serialize};
const METADATA: TableDefinition<u8, &[u8]> = TableDefinition::new("metadata");
const DOMAINS: TableDefinition<u8, &[u8]> = TableDefinition::new("domains");
const GENERATIONS: TableDefinition<u64, &[u8]> = TableDefinition::new("generations");
const PARTICIPANTS: TableDefinition<&[u8; 9], &[u8]> = TableDefinition::new("participants");
const SESSIONS: TableDefinition<&[u8; 40], &[u8]> = TableDefinition::new("sessions");
const SESSION_TERMINALS: TableDefinition<&[u8; 40], &[u8]> =
TableDefinition::new("session_terminals");
const MAILBOXES: TableDefinition<u64, &[u8]> = TableDefinition::new("mailboxes");
const STATE_RESET: TableDefinition<u8, &[u8]> = TableDefinition::new("state_reset");
const SINGLETON_KEY: u8 = 0;
const STATE_RESET_SCHEMA_V1: &str = "lazaret-state-reset-v1";
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum StateResetScopeV1 {
All,
Selected(Vec<u8>),
}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct StateResetIntentV1 {
schema: String,
domain_id: lazaret_core::identity::DomainId,
scope: StateResetScopeV1,
}
impl StateResetIntentV1 {
#[must_use]
pub fn new(domain_id: lazaret_core::identity::DomainId, scope: StateResetScopeV1) -> Self {
Self {
schema: STATE_RESET_SCHEMA_V1.to_owned(),
domain_id,
scope,
}
}
#[must_use]
pub const fn domain_id(&self) -> lazaret_core::identity::DomainId {
self.domain_id
}
#[must_use]
pub const fn scope(&self) -> &StateResetScopeV1 {
&self.scope
}
fn validate(&self) -> Result<(), DomainStoreError> {
if self.schema == STATE_RESET_SCHEMA_V1 {
Ok(())
} else {
Err(DomainStoreError::InvalidRow("State Reset"))
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DomainStoreMutation {
PutGeneration(Generation),
PutParticipant(Participant),
CleanParticipantAfterAbsence(Participant),
PutSession(Session),
PutMailbox(Mailbox),
CleanMailboxAfterAbsence(Mailbox),
PutSessionTerminal(SessionTerminal),
CleanSessionTerminalAfterAbsence(SessionTerminal),
}
#[derive(Debug)]
pub enum DomainStoreError {
Io(io::Error),
Database(DatabaseError),
Transaction(TransactionError),
Table(TableError),
Storage(StorageError),
Commit(CommitError),
Codec(CodecError),
Transition(RecordTransitionError),
MissingMetadata,
MissingRow(&'static str),
DuplicateRow(&'static str),
InvalidRow(&'static str),
DomainMismatch,
GenerationMismatch,
}
impl fmt::Display for DomainStoreError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Io(error) => write!(formatter, "Domain Store I/O failed: {error}"),
Self::Database(error) => write!(formatter, "Domain Store open failed: {error}"),
Self::Transaction(error) => {
write!(formatter, "Domain Store transaction failed: {error}")
}
Self::Table(error) => write!(formatter, "Domain Store table failed: {error}"),
Self::Storage(error) => write!(formatter, "Domain Store storage failed: {error}"),
Self::Commit(error) => write!(formatter, "Domain Store commit failed: {error}"),
Self::Codec(error) => {
write!(formatter, "Domain Store record failed validation: {error}")
}
Self::Transition(error) => {
write!(formatter, "Domain Store transition refused: {error}")
}
Self::MissingMetadata => formatter.write_str("Domain Store metadata is missing"),
Self::MissingRow(kind) => write!(formatter, "referenced {kind} row is missing"),
Self::DuplicateRow(kind) => write!(formatter, "duplicate {kind} row"),
Self::InvalidRow(kind) => write!(formatter, "invalid {kind} row"),
Self::DomainMismatch => formatter.write_str("Domain Store Domain identity mismatch"),
Self::GenerationMismatch => formatter.write_str("Domain Store generation mismatch"),
}
}
}
impl std::error::Error for DomainStoreError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(error) => Some(error),
Self::Database(error) => Some(error),
Self::Transaction(error) => Some(error),
Self::Table(error) => Some(error),
Self::Storage(error) => Some(error),
Self::Commit(error) => Some(error),
Self::Codec(error) => Some(error),
Self::Transition(error) => Some(error),
Self::MissingMetadata
| Self::MissingRow(_)
| Self::DuplicateRow(_)
| Self::InvalidRow(_)
| Self::DomainMismatch
| Self::GenerationMismatch => None,
}
}
}
macro_rules! from_error {
($source:ty, $variant:ident) => {
impl From<$source> for DomainStoreError {
fn from(error: $source) -> Self {
Self::$variant(error)
}
}
};
}
from_error!(io::Error, Io);
from_error!(DatabaseError, Database);
from_error!(TransactionError, Transaction);
from_error!(TableError, Table);
from_error!(StorageError, Storage);
from_error!(CommitError, Commit);
from_error!(CodecError, Codec);
from_error!(RecordTransitionError, Transition);
pub struct HostCoordinatorDomainStore {
database: Database,
_domain_lock: ExclusiveFileLock,
}
impl HostCoordinatorDomainStore {
/// Opens the Domain Store while retaining the caller's exclusive Domain lock.
///
/// # Errors
///
/// Refuses I/O, database, metadata, row, or Domain identity failures.
pub fn open(
domain_lock: ExclusiveFileLock,
database_path: &Path,
expected_domain: &Domain,
) -> Result<Self, DomainStoreError> {
let is_new = !database_path.try_exists()?;
let database = if is_new {
Database::create(database_path)?
} else {
Database::open(database_path)?
};
let store = Self {
database,
_domain_lock: domain_lock,
};
if is_new {
store.initialize(expected_domain)?;
} else {
store.validate_metadata()?;
if store.snapshot()?.domain != *expected_domain {
return Err(DomainStoreError::DomainMismatch);
}
}
Ok(store)
}
/// Opens the Domain Store from a descriptor-relative file while retaining the Domain lock.
///
/// # Errors
/// Refuses I/O, database, metadata, row, or Domain identity failures.
pub fn open_file(
domain_lock: ExclusiveFileLock,
database_file: File,
expected_domain: &Domain,
) -> Result<Self, DomainStoreError> {
let is_new = database_file.metadata()?.len() == 0;
let database = redb::Database::builder().create_file(database_file)?;
let store = Self {
database,
_domain_lock: domain_lock,
};
if is_new {
store.initialize(expected_domain)?;
} else {
store.validate_metadata()?;
if store.snapshot()?.domain != *expected_domain {
return Err(DomainStoreError::DomainMismatch);
}
}
Ok(store)
}
/// Reads and validates every typed row as one consistent redb snapshot.
///
/// # Errors
///
/// Refuses database, table, codec, key, duplicate, or relationship failures.
pub fn snapshot(&self) -> Result<DomainStore, DomainStoreError> {
self.validate_metadata()?;
let read = self.database.begin_read()?;
let domain = read_domain(&read.open_table(DOMAINS)?)?;
let generations = read_generations(&read.open_table(GENERATIONS)?)?;
let participants = read_participants(&read.open_table(PARTICIPANTS)?)?;
let sessions = read_sessions(&read.open_table(SESSIONS)?)?;
let session_terminals = read_session_terminals(&read.open_table(SESSION_TERMINALS)?)?;
let mailboxes = read_mailboxes(&read.open_table(MAILBOXES)?)?;
let snapshot = DomainStore {
metadata: DomainStoreMetadata::v1(),
domain,
generations,
participants,
sessions,
session_terminals,
mailboxes,
};
validate_snapshot(&snapshot)?;
Ok(snapshot)
}
/// Applies the complete mutation batch in one immediate-durability redb transaction.
///
/// # Errors
///
/// Refuses invalid transitions, generation mismatches, malformed relationships, or storage
/// failures without committing any member of the batch.
pub fn apply(&self, mutations: &[DomainStoreMutation]) -> Result<(), DomainStoreError> {
if mutations.is_empty() {
return Ok(());
}
let mut snapshot = self.snapshot()?;
for mutation in mutations {
apply_mutation(&mut snapshot, mutation)?;
}
validate_snapshot(&snapshot)?;
self.write_snapshot(&snapshot)
}
/// Reads the typed reset recovery point from the versioned Domain Store.
///
/// # Errors
/// Refuses malformed, unknown-version, or mismatched-Domain intent.
pub fn state_reset_intent(&self) -> Result<Option<StateResetIntentV1>, DomainStoreError> {
self.validate_metadata()?;
let read = self.database.begin_read()?;
let table = match read.open_table(STATE_RESET) {
Ok(table) => table,
Err(TableError::TableDoesNotExist(_)) => return Ok(None),
Err(error) => return Err(error.into()),
};
let Some(value) = table.get(SINGLETON_KEY)? else {
return Ok(None);
};
let intent = decode::<StateResetIntentV1>(value.value())?;
intent.validate()?;
if intent.domain_id != self.snapshot()?.domain.domain_id {
return Err(DomainStoreError::DomainMismatch);
}
Ok(Some(intent))
}
/// Commits the reset recovery point in one immediate-durability transaction.
///
/// # Errors
/// Refuses an invalid, foreign, or already-present intent.
pub fn begin_state_reset(&self, intent: &StateResetIntentV1) -> Result<(), DomainStoreError> {
intent.validate()?;
if intent.domain_id != self.snapshot()?.domain.domain_id {
return Err(DomainStoreError::DomainMismatch);
}
let write = self.database.begin_write()?;
{
let mut table = write.open_table(STATE_RESET)?;
if table.get(SINGLETON_KEY)?.is_some() {
return Err(DomainStoreError::DuplicateRow("State Reset"));
}
let encoded = encode(intent)?;
table.insert(SINGLETON_KEY, encoded.as_slice())?;
}
write.commit()?;
Ok(())
}
/// Removes the exact completed intent in one immediate-durability transaction.
///
/// # Errors
/// Refuses missing or changed intent and storage failures.
pub fn finish_state_reset(
&self,
expected: &StateResetIntentV1,
) -> Result<(), DomainStoreError> {
if self.state_reset_intent()?.as_ref() != Some(expected) {
return Err(DomainStoreError::InvalidRow("State Reset"));
}
let write = self.database.begin_write()?;
{
let mut table = write.open_table(STATE_RESET)?;
table.remove(SINGLETON_KEY)?;
}
write.commit()?;
Ok(())
}
fn initialize(&self, domain: &Domain) -> Result<(), DomainStoreError> {
let snapshot = DomainStore {
metadata: DomainStoreMetadata::v1(),
domain: domain.clone(),
generations: Vec::new(),
participants: Vec::new(),
sessions: Vec::new(),
session_terminals: Vec::new(),
mailboxes: Vec::new(),
};
self.write_snapshot(&snapshot)
}
fn validate_metadata(&self) -> Result<(), DomainStoreError> {
let read = self.database.begin_read()?;
let table = read.open_table(METADATA).map_err(|error| match error {
TableError::TableDoesNotExist(_) => DomainStoreError::MissingMetadata,
other => DomainStoreError::Table(other),
})?;
read_metadata(&table)?;
Ok(())
}
fn write_snapshot(&self, snapshot: &DomainStore) -> Result<(), DomainStoreError> {
let write = self.database.begin_write()?;
{
let mut table = write.open_table(METADATA)?;
let encoded = encode(&snapshot.metadata)?;
table.insert(SINGLETON_KEY, encoded.as_slice())?;
}
{
let mut table = write.open_table(DOMAINS)?;
let encoded = encode(&snapshot.domain)?;
table.insert(SINGLETON_KEY, encoded.as_slice())?;
}
{
let mut table = write.open_table(GENERATIONS)?;
for record in &snapshot.generations {
let encoded = encode(record)?;
table.insert(record.generation_id.get(), encoded.as_slice())?;
}
}
{
let mut table = write.open_table(PARTICIPANTS)?;
for record in &snapshot.participants {
let encoded = encode(record)?;
table.insert(&participant_key(record), encoded.as_slice())?;
}
}
{
let mut table = write.open_table(SESSIONS)?;
for record in &snapshot.sessions {
let encoded = encode(record)?;
table.insert(
&session_key(record.generation_id, record.session_id),
encoded.as_slice(),
)?;
}
}
{
let mut table = write.open_table(SESSION_TERMINALS)?;
for record in &snapshot.session_terminals {
let encoded = encode(record)?;
table.insert(
&session_key(record.generation_id(), record.session_id()),
encoded.as_slice(),
)?;
}
}
{
let mut table = write.open_table(MAILBOXES)?;
for record in &snapshot.mailboxes {
let encoded = encode(record)?;
table.insert(record.generation_id.get(), encoded.as_slice())?;
}
}
write.commit()?;
Ok(())
}
}
fn read_generations(
table: &redb::ReadOnlyTable<u64, &[u8]>,
) -> Result<Vec<Generation>, DomainStoreError> {
let mut records = Vec::new();
for entry in table.iter()? {
let (key, value) = entry?;
let record: Generation = decode(value.value())?;
if key.value() != record.generation_id.get() {
return Err(DomainStoreError::InvalidRow("Generation key"));
}
records.push(record);
}
Ok(records)
}
fn read_domain(table: &redb::ReadOnlyTable<u8, &[u8]>) -> Result<Domain, DomainStoreError> {
let mut domain = None;
for entry in table.iter()? {
let (key, value) = entry?;
let record: Domain = decode(value.value())?;
if key.value() != SINGLETON_KEY {
return Err(DomainStoreError::InvalidRow("Domain key"));
}
if domain.replace(record).is_some() {
return Err(DomainStoreError::DuplicateRow("Domain"));
}
}
domain.ok_or(DomainStoreError::MissingRow("Domain"))
}
fn read_metadata(
table: &redb::ReadOnlyTable<u8, &[u8]>,
) -> Result<DomainStoreMetadata, DomainStoreError> {
let mut metadata = None;
for entry in table.iter()? {
let (key, value) = entry?;
let record: DomainStoreMetadata = decode(value.value())?;
if key.value() != SINGLETON_KEY {
return Err(DomainStoreError::InvalidRow("StoreMetadata key"));
}
if metadata.replace(record).is_some() {
return Err(DomainStoreError::DuplicateRow("StoreMetadata"));
}
}
metadata.ok_or(DomainStoreError::MissingMetadata)
}
fn read_participants(
table: &redb::ReadOnlyTable<&[u8; 9], &[u8]>,
) -> Result<Vec<Participant>, DomainStoreError> {
let mut records = Vec::new();
for entry in table.iter()? {
let (key, value) = entry?;
let record: Participant = decode(value.value())?;
if key.value() != &participant_key(&record) {
return Err(DomainStoreError::InvalidRow("Participant key"));
}
records.push(record);
}
Ok(records)
}
fn read_sessions(
table: &redb::ReadOnlyTable<&[u8; 40], &[u8]>,
) -> Result<Vec<Session>, DomainStoreError> {
let mut records = Vec::new();
for entry in table.iter()? {
let (key, value) = entry?;
let record: Session = decode(value.value())?;
if key.value() != &session_key(record.generation_id, record.session_id) {
return Err(DomainStoreError::InvalidRow("Session key"));
}
records.push(record);
}
Ok(records)
}
fn read_session_terminals(
table: &redb::ReadOnlyTable<&[u8; 40], &[u8]>,
) -> Result<Vec<SessionTerminal>, DomainStoreError> {
let mut records = Vec::new();
for entry in table.iter()? {
let (key, value) = entry?;
let record: SessionTerminal = decode(value.value())?;
if key.value() != &session_key(record.generation_id(), record.session_id()) {
return Err(DomainStoreError::InvalidRow("Session Terminal key"));
}
records.push(record);
}
Ok(records)
}
fn read_mailboxes(
table: &redb::ReadOnlyTable<u64, &[u8]>,
) -> Result<Vec<Mailbox>, DomainStoreError> {
let mut records = Vec::new();
for entry in table.iter()? {
let (key, value) = entry?;
let record: Mailbox = decode(value.value())?;
if key.value() != record.generation_id.get() {
return Err(DomainStoreError::InvalidRow("Mailbox key"));
}
records.push(record);
}
Ok(records)
}
fn apply_mutation(
snapshot: &mut DomainStore,
mutation: &DomainStoreMutation,
) -> Result<(), DomainStoreError> {
match mutation {
DomainStoreMutation::PutGeneration(next) => put_generation(snapshot, next.clone()),
DomainStoreMutation::PutParticipant(next) => put_participant(snapshot, next.clone(), false),
DomainStoreMutation::CleanParticipantAfterAbsence(next) => {
put_participant(snapshot, next.clone(), true)
}
DomainStoreMutation::PutSession(next) => put_session(snapshot, next.clone()),
DomainStoreMutation::PutMailbox(next) => put_mailbox(snapshot, next.clone(), false),
DomainStoreMutation::CleanMailboxAfterAbsence(next) => {
put_mailbox(snapshot, next.clone(), true)
}
DomainStoreMutation::PutSessionTerminal(next) => {
put_session_terminal(snapshot, next.clone(), false)
}
DomainStoreMutation::CleanSessionTerminalAfterAbsence(next) => {
put_session_terminal(snapshot, next.clone(), true)
}
}
}
fn put_generation(snapshot: &mut DomainStore, next: Generation) -> Result<(), DomainStoreError> {
if next.domain_id != snapshot.domain.domain_id {
return Err(DomainStoreError::DomainMismatch);
}
if let Some(current) = snapshot
.generations
.iter_mut()
.find(|record| record.generation_id == next.generation_id)
{
validate_generation_transition(current, &next)?;
*current = next;
} else {
if next.durable_decision != DurableDecision::None {
return Err(DomainStoreError::InvalidRow("initial Generation"));
}
snapshot.generations.push(next);
}
Ok(())
}
fn put_participant(
snapshot: &mut DomainStore,
next: Participant,
absence_proven: bool,
) -> Result<(), DomainStoreError> {
require_generation(snapshot, next.generation_id())?;
if let Some(current) = snapshot.participants.iter_mut().find(|record| {
record.generation_id() == next.generation_id() && record.role() == next.role()
}) {
if absence_proven {
current.validate_transition_to_clean_after_absence(&next)?;
} else {
current.validate_transition_to(&next)?;
}
*current = next;
} else {
if absence_proven || next.state() != ParticipantState::Intent || next.ownership().is_some()
{
return Err(DomainStoreError::InvalidRow("initial Participant"));
}
snapshot.participants.push(next);
}
Ok(())
}
fn put_session(snapshot: &mut DomainStore, next: Session) -> Result<(), DomainStoreError> {
require_generation(snapshot, next.generation_id)?;
if let Some(current) = snapshot.sessions.iter_mut().find(|record| {
record.generation_id == next.generation_id && record.session_id == next.session_id
}) {
validate_session_transition(current, &next)?;
*current = next;
} else {
if !matches!(
next.session_phase,
SessionPhase::New | SessionPhase::Waiting | SessionPhase::Registering
) {
return Err(DomainStoreError::InvalidRow("initial Session"));
}
snapshot.sessions.push(next);
}
Ok(())
}
fn put_mailbox(
snapshot: &mut DomainStore,
next: Mailbox,
absence_proven: bool,
) -> Result<(), DomainStoreError> {
require_generation(snapshot, next.generation_id)?;
if let Some(current) = snapshot
.mailboxes
.iter_mut()
.find(|record| record.generation_id == next.generation_id)
{
validate_mailbox_transition(current, &next, absence_proven)?;
*current = next;
} else {
if absence_proven || next.state != ParticipantState::Intent {
return Err(DomainStoreError::InvalidRow("initial Mailbox"));
}
snapshot.mailboxes.push(next);
}
Ok(())
}
fn put_session_terminal(
snapshot: &mut DomainStore,
next: SessionTerminal,
absence_proven: bool,
) -> Result<(), DomainStoreError> {
require_session(snapshot, next.generation_id(), next.session_id())?;
if let Some(current) = snapshot.session_terminals.iter_mut().find(|record| {
record.generation_id() == next.generation_id() && record.session_id() == next.session_id()
}) {
if absence_proven {
current.validate_transition_to_clean_after_absence(&next)?;
} else {
current.validate_transition_to(&next)?;
}
*current = next;
} else {
if absence_proven
|| next.state() != SessionTerminalState::Intent
|| next.container_id().is_some()
|| !next.host_coordinator().is_unassigned()
|| next.pty_master().is_some()
|| !next.exec_client().is_unassigned()
|| next.guest_process().is_some()
{
return Err(DomainStoreError::InvalidRow("initial Session Terminal"));
}
snapshot.session_terminals.push(next);
}
Ok(())
}
fn validate_generation_transition(
current: &Generation,
next: &Generation,
) -> Result<(), DomainStoreError> {
if current.domain_id != next.domain_id
|| current.generation_id != next.generation_id
|| current.configuration_fingerprint != next.configuration_fingerprint
|| current.coordinator_id != next.coordinator_id
|| current.network_sidecar_id != next.network_sidecar_id
|| current.sandbox_id != next.sandbox_id
{
return Err(DomainStoreError::InvalidRow("Generation identity"));
}
let valid = current.durable_decision == next.durable_decision
|| matches!(
(current.durable_decision, next.durable_decision),
(
DurableDecision::None,
DurableDecision::Preparing | DurableDecision::Clean
) | (
DurableDecision::Preparing,
DurableDecision::Published | DurableDecision::Clean
) | (DurableDecision::Published, DurableDecision::Clean)
);
if valid {
Ok(())
} else {
Err(DomainStoreError::InvalidRow("Generation transition"))
}
}
fn validate_session_transition(current: &Session, next: &Session) -> Result<(), DomainStoreError> {
if current.generation_id != next.generation_id
|| current.session_id != next.session_id
|| current.cli_id != next.cli_id
|| current.session_anchor_id != next.session_anchor_id
{
return Err(DomainStoreError::InvalidRow("Session identity"));
}
let valid = current.session_phase == next.session_phase
|| matches!(
(current.session_phase, next.session_phase),
(
SessionPhase::New,
SessionPhase::Waiting | SessionPhase::Registering | SessionPhase::Failed
) | (
SessionPhase::Waiting,
SessionPhase::Registering | SessionPhase::Closed | SessionPhase::Failed
) | (
SessionPhase::Registering | SessionPhase::HostExecuting,
SessionPhase::Attached | SessionPhase::Closed | SessionPhase::Failed
) | (
SessionPhase::Attached,
SessionPhase::Prompting
| SessionPhase::HostExecuting
| SessionPhase::Closed
| SessionPhase::Failed
) | (
SessionPhase::Prompting,
SessionPhase::Attached
| SessionPhase::HostExecuting
| SessionPhase::Closed
| SessionPhase::Failed
)
);
if valid {
Ok(())
} else {
Err(DomainStoreError::InvalidRow("Session transition"))
}
}
fn validate_mailbox_transition(
current: &Mailbox,
next: &Mailbox,
absence_proven: bool,
) -> Result<(), DomainStoreError> {
if current.generation_id != next.generation_id
|| current.identity_path != next.identity_path
|| current.inbox_path != next.inbox_path
|| current.outbox_path != next.outbox_path
{
return Err(DomainStoreError::InvalidRow("Mailbox identity"));
}
if absence_proven {
return if matches!(
current.state,
ParticipantState::Intent | ParticipantState::ReleaseIntent
) && next.state == ParticipantState::Clean
{
Ok(())
} else {
Err(DomainStoreError::InvalidRow("Mailbox cleanup"))
};
}
let valid = current.state == next.state
|| matches!(
(current.state, next.state),
(
ParticipantState::Intent,
ParticipantState::LeaseHeld
| ParticipantState::Prepared
| ParticipantState::Committed
| ParticipantState::ReleaseIntent
) | (
ParticipantState::LeaseHeld
| ParticipantState::Prepared
| ParticipantState::Committed,
ParticipantState::ReleaseIntent
)
);
if valid {
Ok(())
} else {
Err(DomainStoreError::InvalidRow("Mailbox transition"))
}
}
fn validate_snapshot(snapshot: &DomainStore) -> Result<(), DomainStoreError> {
for (index, generation) in snapshot.generations.iter().enumerate() {
if generation.domain_id != snapshot.domain.domain_id {
return Err(DomainStoreError::DomainMismatch);
}
if snapshot.generations[..index]
.iter()
.any(|other| other.generation_id == generation.generation_id)
{
return Err(DomainStoreError::DuplicateRow("Generation"));
}
}
for (index, participant) in snapshot.participants.iter().enumerate() {
require_generation(snapshot, participant.generation_id())?;
if snapshot.participants[..index].iter().any(|other| {
other.generation_id() == participant.generation_id()
&& other.role() == participant.role()
}) {
return Err(DomainStoreError::DuplicateRow("Participant"));
}
}
for (index, session) in snapshot.sessions.iter().enumerate() {
require_generation(snapshot, session.generation_id)?;
if snapshot.sessions[..index].iter().any(|other| {
other.generation_id == session.generation_id && other.session_id == session.session_id
}) {
return Err(DomainStoreError::DuplicateRow("Session"));
}
}
for (index, terminal) in snapshot.session_terminals.iter().enumerate() {
require_session(snapshot, terminal.generation_id(), terminal.session_id())?;
if snapshot.session_terminals[..index].iter().any(|other| {
other.generation_id() == terminal.generation_id()
&& other.session_id() == terminal.session_id()
}) {
return Err(DomainStoreError::DuplicateRow("Session Terminal"));
}
}
for (index, mailbox) in snapshot.mailboxes.iter().enumerate() {
require_generation(snapshot, mailbox.generation_id)?;
if snapshot.mailboxes[..index]
.iter()
.any(|other| other.generation_id == mailbox.generation_id)
{
return Err(DomainStoreError::DuplicateRow("Mailbox"));
}
}
Ok(())
}
fn require_generation(
snapshot: &DomainStore,
generation_id: GenerationId,
) -> Result<(), DomainStoreError> {
if snapshot
.generations
.iter()
.any(|record| record.generation_id == generation_id)
{
Ok(())
} else {
Err(DomainStoreError::GenerationMismatch)
}
}
fn require_session(
snapshot: &DomainStore,
generation_id: GenerationId,
session_id: SessionId,
) -> Result<(), DomainStoreError> {
require_generation(snapshot, generation_id)?;
if snapshot
.sessions
.iter()
.any(|record| record.generation_id == generation_id && record.session_id == session_id)
{
Ok(())
} else {
Err(DomainStoreError::MissingRow("Session"))
}
}
fn participant_key(record: &Participant) -> [u8; 9] {
let mut key = [0; 9];
key[..8].copy_from_slice(&record.generation_id().get().to_be_bytes());
key[8] = role_key(record.role());
key
}
fn role_key(role: ParticipantRole) -> u8 {
match role {
ParticipantRole::PreparedLower => 0,
ParticipantRole::StateAnchor => 1,
ParticipantRole::GitRouting => 2,
ParticipantRole::Network => 3,
ParticipantRole::Helper => 4,
ParticipantRole::Sandbox => 5,
ParticipantRole::Publication => 6,
}
}
fn session_key(generation_id: GenerationId, session_id: SessionId) -> [u8; 40] {
let mut key = [0; 40];
key[..8].copy_from_slice(&generation_id.get().to_be_bytes());
key[8..].copy_from_slice(session_id.as_bytes());
key
}
#[cfg(test)]
mod tests;

494
crates/lazaret-host-coordinator/src/domain_store/tests.rs
View file

@ -1,494 +0,0 @@
use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use lazaret_core::codec::encode;
use lazaret_core::identity::{
ClientId, ConfigurationFingerprint, CoordinatorId, DomainId, GenerationId, HostPath,
NetworkSidecarId, SandboxId, SessionAnchorId, SessionId,
};
use lazaret_core::records::{
DurableDecision, ParticipantOwnership, PersistedHostProcessOwnership, PublicationOwnership,
SessionTerminalFields,
};
use lazaret_linux::filesystem::try_exclusive_flock;
use redb::{ReadableDatabase, TableDefinition, TableHandle};
use serde::Serialize;
use super::*;
static NEXT_DIRECTORY: AtomicU64 = AtomicU64::new(0);
const FOREIGN: TableDefinition<u8, u8> = TableDefinition::new("foreign");
struct TestDirectory(PathBuf);
impl TestDirectory {
fn new(label: &str) -> Self {
let sequence = NEXT_DIRECTORY.fetch_add(1, Ordering::Relaxed);
let path = std::env::temp_dir().join(format!(
"lazaret-domain-store-{label}-{}-{sequence}",
std::process::id()
));
std::fs::create_dir(&path).unwrap();
Self(path)
}
fn path(&self, name: &str) -> PathBuf {
self.0.join(name)
}
}
impl Drop for TestDirectory {
fn drop(&mut self) {
std::fs::remove_dir_all(&self.0).unwrap();
}
}
fn bytes(value: u8) -> [u8; 32] {
[value; 32]
}
fn domain() -> Domain {
Domain {
domain_id: DomainId::new(bytes(1)),
profile_name: "trusted".into(),
canonical_workspace: HostPath::new(b"/workspace/project".to_vec()),
}
}
fn generation(decision: DurableDecision) -> Generation {
Generation {
domain_id: domain().domain_id,
generation_id: GenerationId::new(7),
configuration_fingerprint: ConfigurationFingerprint::new(bytes(2)),
coordinator_id: CoordinatorId::new(bytes(3)),
network_sidecar_id: NetworkSidecarId::new(bytes(4)),
sandbox_id: SandboxId::new(bytes(5)),
durable_decision: decision,
}
}
fn session(phase: SessionPhase) -> Session {
Session {
generation_id: GenerationId::new(7),
session_id: SessionId::new(bytes(6)),
cli_id: ClientId::new(bytes(7)),
session_anchor_id: SessionAnchorId::new(bytes(8)),
session_phase: phase,
}
}
fn mailbox(state: ParticipantState) -> Mailbox {
Mailbox {
generation_id: GenerationId::new(7),
identity_path: HostPath::new(b"identity".to_vec()),
inbox_path: HostPath::new(b"inbox".to_vec()),
outbox_path: HostPath::new(b"outbox".to_vec()),
state,
}
}
fn bare_terminal(state: SessionTerminalState) -> SessionTerminal {
SessionTerminal::snapshot(SessionTerminalFields {
generation_id: GenerationId::new(7),
session_id: SessionId::new(bytes(6)),
state,
container_id: None,
host_coordinator: PersistedHostProcessOwnership::unassigned(),
pty_master: None,
exec_client: PersistedHostProcessOwnership::unassigned(),
guest_process: None,
})
.unwrap()
}
fn publication_intent() -> Participant {
Participant::bare_intent(GenerationId::new(7), ParticipantRole::Publication)
}
fn publication_partial() -> Participant {
Participant::snapshot(
GenerationId::new(7),
ParticipantRole::Publication,
ParticipantState::Intent,
Some(ParticipantOwnership::Publication(PublicationOwnership {
staged_socket_path: None,
staged_identity_path: None,
socket_path: Some(HostPath::new(b"control.sock".to_vec())),
identity_path: None,
socket_identity: None,
identity_file_identity: None,
})),
)
.unwrap()
}
fn lock_file(path: &Path) -> File {
OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(path)
.unwrap()
}
fn open_store(directory: &TestDirectory) -> HostCoordinatorDomainStore {
let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &domain()).unwrap()
}
#[test]
fn reset_intent_is_typed_transactional_and_domain_bound() {
let directory = TestDirectory::new("state-reset");
let store = open_store(&directory);
let intent = StateResetIntentV1::new(
domain().domain_id,
StateResetScopeV1::Selected(b"/usr".to_vec()),
);
assert_eq!(store.state_reset_intent().unwrap(), None);
store.begin_state_reset(&intent).unwrap();
assert_eq!(store.state_reset_intent().unwrap(), Some(intent.clone()));
assert!(store.begin_state_reset(&intent).is_err());
store.finish_state_reset(&intent).unwrap();
assert_eq!(store.state_reset_intent().unwrap(), None);
let foreign = StateResetIntentV1::new(DomainId::new(bytes(99)), StateResetScopeV1::All);
assert!(matches!(
store.begin_state_reset(&foreign),
Err(DomainStoreError::DomainMismatch)
));
}
#[test]
fn reset_intent_unknown_version_refuses_before_scope_use() {
let directory = TestDirectory::new("state-reset-version");
let store = open_store(&directory);
let intent = StateResetIntentV1 {
schema: "lazaret-state-reset-v2".into(),
domain_id: domain().domain_id,
scope: StateResetScopeV1::All,
};
assert!(matches!(
store.begin_state_reset(&intent),
Err(DomainStoreError::InvalidRow("State Reset"))
));
}
#[test]
fn malformed_persisted_reset_intent_refuses_without_fallback() {
let directory = TestDirectory::new("state-reset-malformed");
drop(open_store(&directory));
let database = Database::open(directory.path("domain.redb")).unwrap();
let write = database.begin_write().unwrap();
{
let mut table = write.open_table(STATE_RESET).unwrap();
table
.insert(SINGLETON_KEY, b"not-postcard".as_slice())
.unwrap();
}
write.commit().unwrap();
drop(database);
assert!(matches!(
open_store(&directory).state_reset_intent(),
Err(DomainStoreError::Codec(_))
));
}
#[test]
fn descriptor_store_open_cannot_be_redirected_by_path_replacement() {
let directory = TestDirectory::new("descriptor-open");
let database_path = directory.path("domain.redb");
let database_file = lock_file(&database_path);
let retained_path = directory.path("retained.redb");
std::fs::rename(&database_path, &retained_path).unwrap();
drop(lock_file(&database_path));
let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
drop(HostCoordinatorDomainStore::open_file(lock, database_file, &domain()).unwrap());
assert!(Database::open(retained_path).is_ok());
assert_eq!(std::fs::metadata(database_path).unwrap().len(), 0);
}
#[test]
fn transaction_reopen_and_interrupted_write_preserve_complete_documents() {
let directory = TestDirectory::new("transaction");
let store = open_store(&directory);
store
.apply(&[
DomainStoreMutation::PutGeneration(generation(DurableDecision::None)),
DomainStoreMutation::PutParticipant(publication_intent()),
DomainStoreMutation::PutSession(session(SessionPhase::New)),
DomainStoreMutation::PutMailbox(mailbox(ParticipantState::Intent)),
DomainStoreMutation::PutSessionTerminal(bare_terminal(SessionTerminalState::Intent)),
])
.unwrap();
let refused = store.apply(&[
DomainStoreMutation::PutGeneration(generation(DurableDecision::Preparing)),
DomainStoreMutation::PutSession(session(SessionPhase::Attached)),
]);
assert!(matches!(
refused,
Err(DomainStoreError::InvalidRow("Session transition"))
));
assert_eq!(
store.snapshot().unwrap().generations[0].durable_decision,
DurableDecision::None
);
drop(store);
let database = Database::open(directory.path("domain.redb")).unwrap();
let write = database.begin_write().unwrap();
{
let mut table = write.open_table(GENERATIONS).unwrap();
let encoded = encode(&generation(DurableDecision::Preparing)).unwrap();
table.insert(7, encoded.as_slice()).unwrap();
}
drop(write);
drop(database);
let reopened = open_store(&directory);
let snapshot = reopened.snapshot().unwrap();
assert_eq!(
snapshot.generations[0].durable_decision,
DurableDecision::None
);
assert_eq!(snapshot.participants.len(), 1);
assert_eq!(snapshot.sessions.len(), 1);
assert_eq!(snapshot.mailboxes.len(), 1);
assert_eq!(snapshot.session_terminals.len(), 1);
}
#[test]
fn opener_retains_the_domain_lock_for_its_complete_lifetime() {
let directory = TestDirectory::new("opener");
let store = open_store(&directory);
assert!(try_exclusive_flock(lock_file(&directory.path("domain.lock"))).is_err());
assert!(matches!(
Database::open(directory.path("domain.redb")),
Err(DatabaseError::DatabaseAlreadyOpen)
));
drop(store);
assert!(try_exclusive_flock(lock_file(&directory.path("domain.lock"))).is_ok());
}
#[test]
fn metadata_and_typed_row_validation_fail_closed() {
#[derive(Serialize)]
struct RawMetadata<'a> {
schema_id: &'a str,
schema_version: u32,
}
let directory = TestDirectory::new("metadata");
drop(open_store(&directory));
let database = Database::open(directory.path("domain.redb")).unwrap();
let write = database.begin_write().unwrap();
{
let mut table = write.open_table(METADATA).unwrap();
let encoded = encode(&RawMetadata {
schema_id: "lazaret-domain-store-v1",
schema_version: 2,
})
.unwrap();
table.insert(SINGLETON_KEY, encoded.as_slice()).unwrap();
}
write.commit().unwrap();
drop(database);
let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
assert!(matches!(
HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &domain()),
Err(DomainStoreError::Codec(_))
));
let extra_metadata_directory = TestDirectory::new("extra-metadata-row");
drop(open_store(&extra_metadata_directory));
let database = Database::open(extra_metadata_directory.path("domain.redb")).unwrap();
let write = database.begin_write().unwrap();
{
let mut table = write.open_table(METADATA).unwrap();
table.insert(1, &[0xff, 0xff][..]).unwrap();
}
write.commit().unwrap();
drop(database);
let lock =
try_exclusive_flock(lock_file(&extra_metadata_directory.path("domain.lock"))).unwrap();
assert!(matches!(
HostCoordinatorDomainStore::open(
lock,
&extra_metadata_directory.path("domain.redb"),
&domain(),
),
Err(DomainStoreError::Codec(_))
));
let row_directory = TestDirectory::new("row-key");
let store = open_store(&row_directory);
store
.apply(&[DomainStoreMutation::PutGeneration(generation(
DurableDecision::None,
))])
.unwrap();
drop(store);
let database = Database::open(row_directory.path("domain.redb")).unwrap();
let write = database.begin_write().unwrap();
{
let mut table = write.open_table(GENERATIONS).unwrap();
let encoded = encode(&generation(DurableDecision::None)).unwrap();
table.insert(8, encoded.as_slice()).unwrap();
}
write.commit().unwrap();
drop(database);
let lock = try_exclusive_flock(lock_file(&row_directory.path("domain.lock"))).unwrap();
assert!(matches!(
HostCoordinatorDomainStore::open(lock, &row_directory.path("domain.redb"), &domain()),
Err(DomainStoreError::InvalidRow("Generation key"))
));
let domain_directory = TestDirectory::new("extra-domain-row");
drop(open_store(&domain_directory));
let database = Database::open(domain_directory.path("domain.redb")).unwrap();
let write = database.begin_write().unwrap();
{
let mut table = write.open_table(DOMAINS).unwrap();
table.insert(1, &[0xff, 0xff][..]).unwrap();
}
write.commit().unwrap();
drop(database);
let lock = try_exclusive_flock(lock_file(&domain_directory.path("domain.lock"))).unwrap();
assert!(matches!(
HostCoordinatorDomainStore::open(lock, &domain_directory.path("domain.redb"), &domain()),
Err(DomainStoreError::Codec(_))
));
}
#[test]
fn generation_relationships_and_domain_identity_cannot_be_forged() {
let directory = TestDirectory::new("relationships");
let store = open_store(&directory);
assert!(matches!(
store.apply(&[DomainStoreMutation::PutParticipant(publication_intent())]),
Err(DomainStoreError::GenerationMismatch)
));
let mut wrong_domain = generation(DurableDecision::None);
wrong_domain.domain_id = DomainId::new(bytes(99));
assert!(matches!(
store.apply(&[DomainStoreMutation::PutGeneration(wrong_domain)]),
Err(DomainStoreError::DomainMismatch)
));
store
.apply(&[DomainStoreMutation::PutGeneration(generation(
DurableDecision::None,
))])
.unwrap();
drop(store);
let lock = try_exclusive_flock(lock_file(&directory.path("domain.lock"))).unwrap();
let mut other_domain = domain();
other_domain.profile_name = "other".into();
assert!(matches!(
HostCoordinatorDomainStore::open(lock, &directory.path("domain.redb"), &other_domain),
Err(DomainStoreError::DomainMismatch)
));
}
#[test]
fn lifecycle_updates_enforce_transition_and_absence_boundaries() {
let directory = TestDirectory::new("transitions");
let store = open_store(&directory);
store
.apply(&[
DomainStoreMutation::PutGeneration(generation(DurableDecision::None)),
DomainStoreMutation::PutParticipant(publication_intent()),
DomainStoreMutation::PutSession(session(SessionPhase::New)),
DomainStoreMutation::PutMailbox(mailbox(ParticipantState::Intent)),
DomainStoreMutation::PutSessionTerminal(bare_terminal(SessionTerminalState::Intent)),
])
.unwrap();
store
.apply(&[DomainStoreMutation::PutParticipant(publication_partial())])
.unwrap();
assert!(matches!(
store.apply(&[DomainStoreMutation::PutSession(session(
SessionPhase::Attached
))]),
Err(DomainStoreError::InvalidRow("Session transition"))
));
assert!(matches!(
store.apply(&[DomainStoreMutation::PutParticipant(
Participant::snapshot(
GenerationId::new(7),
ParticipantRole::Publication,
ParticipantState::Clean,
Some(ParticipantOwnership::Publication(PublicationOwnership {
staged_socket_path: None,
staged_identity_path: None,
socket_path: None,
identity_path: None,
socket_identity: None,
identity_file_identity: None,
})),
)
.unwrap(),
)]),
Err(DomainStoreError::Transition(
RecordTransitionError::AbsenceProofRequired
))
));
assert!(matches!(
store.apply(&[DomainStoreMutation::PutMailbox(mailbox(
ParticipantState::Clean
))]),
Err(DomainStoreError::InvalidRow("Mailbox transition"))
));
assert!(matches!(
store.apply(&[DomainStoreMutation::PutSessionTerminal(bare_terminal(
SessionTerminalState::Clean,
))]),
Err(DomainStoreError::Transition(
RecordTransitionError::AbsenceProofRequired
))
));
store
.apply(&[DomainStoreMutation::CleanSessionTerminalAfterAbsence(
bare_terminal(SessionTerminalState::Clean),
)])
.unwrap();
let snapshot = store.snapshot().unwrap();
assert_eq!(
snapshot.session_terminals[0].state(),
SessionTerminalState::Clean
);
}
#[test]
fn every_table_is_created_in_the_first_version_document() {
let directory = TestDirectory::new("tables");
drop(open_store(&directory));
let database = Database::open(directory.path("domain.redb")).unwrap();
let read = database.begin_read().unwrap();
for name in [
METADATA.name(),
DOMAINS.name(),
GENERATIONS.name(),
PARTICIPANTS.name(),
SESSIONS.name(),
SESSION_TERMINALS.name(),
MAILBOXES.name(),
] {
assert!(
read.list_tables()
.unwrap()
.any(|handle| handle.name() == name)
);
}
assert!(read.open_table(FOREIGN).is_err());
}

4
crates/lazaret-host-coordinator/src/lib.rs
View file

@ -4,16 +4,16 @@ pub mod bootstrap;
pub mod client_bootstrap;
pub mod cold_start;
pub mod control;
pub mod domain_store;
pub use lazaret_domain_state::{domain_store, state_maintenance};
pub mod escalation;
pub mod git_routing;
pub mod lifecycle;
pub mod network;
pub mod private_roles;
pub mod recovery;
pub mod runtime;
pub mod runtime_selector;
pub mod sandbox_init;
pub mod state_anchor;
pub mod state_maintenance;
pub mod terminal;
pub mod waiter;

47
crates/lazaret-host-coordinator/src/main.rs
View file

@ -1,49 +1,8 @@
fn main() {
let arguments = std::env::args_os().skip(1).collect::<Vec<_>>();
if let Some(result) =
lazaret_host_coordinator::client_bootstrap::dispatch_private_client_bootstrap(&arguments)
{
if let Err(error) = result {
eprintln!("Host Coordinator bootstrap failed: {error}");
std::process::exit(1);
}
return;
}
if let Some(result) =
lazaret_host_coordinator::sandbox_init::dispatch_private_sandbox_init(&arguments)
{
if let Err(error) = result {
eprintln!("sandbox private role failed: {error}");
std::process::exit(1);
}
return;
}
if let Some(result) = lazaret_state_anchor::dispatch_private_state_anchor_recovery(&arguments) {
if let Err(error) = result {
eprintln!("State Anchor recovery failed: {error}");
std::process::exit(1);
}
return;
}
if let Some(result) = lazaret_linux::process::dispatch_private_process_role(&arguments) {
if let Err(error) = result {
eprintln!("private process role failed: {error}");
std::process::exit(1);
}
return;
}
if let Some(result) = lazaret_ignored_workspace::dispatch_private_router(&arguments) {
if let Err(error) = result {
eprintln!("ignored workspace router failed: {error}");
std::process::exit(1);
}
return;
}
if let Some(result) =
lazaret_runtime_supervisor::supervision::dispatch_private_runtime_role(&arguments)
{
if let Err(error) = result {
eprintln!("Runtime Supervisor failed: {error}");
if let Some(outcome) = lazaret_host_coordinator::private_roles::REGISTRY.dispatch(&arguments) {
if let Err(error) = outcome.result {
eprintln!("private role {} failed: {error}", outcome.name);
std::process::exit(1);
}
return;

81
crates/lazaret-host-coordinator/src/private_roles.rs Normal file
View file

@ -0,0 +1,81 @@
use std::ffi::OsString;
use std::io;
use lazaret_core::protocol::PRIVATE_CLIENT_BOOTSTRAP_ROLE;
use lazaret_ignored_workspace::{PRIVATE_ROUTER_ROLE, dispatch_private_router};
use lazaret_linux::private_role::{PrivateRole, PrivateRoleRegistry};
use lazaret_linux::process::{PRIVATE_PROCESS_ROLE, dispatch_private_process_role};
use lazaret_runtime_supervisor::supervision::{
PRIVATE_RUNTIME_ROLE, dispatch_private_runtime_role,
};
use lazaret_state_anchor::{
PRIVATE_STATE_ANCHOR_RECOVERY_ROLE, dispatch_private_state_anchor_recovery,
};
use crate::client_bootstrap::dispatch_private_client_bootstrap;
use crate::sandbox_init::{
PRIVATE_SANDBOX_INIT_ROLE, PRIVATE_SESSION_EXEC_ROLE, PRIVATE_SESSION_IDENTITY_ROLE,
dispatch_private_sandbox_init,
};
fn sandbox_init(arguments: &[OsString]) -> Option<io::Result<()>> {
dispatch_exact(PRIVATE_SANDBOX_INIT_ROLE, arguments)
}
fn client_bootstrap(arguments: &[OsString]) -> Option<io::Result<()>> {
dispatch_private_client_bootstrap(arguments)
.map(|result| result.map_err(|error| io::Error::other(error.to_string())))
}
fn session_exec(arguments: &[OsString]) -> Option<io::Result<()>> {
dispatch_exact(PRIVATE_SESSION_EXEC_ROLE, arguments)
}
fn session_identity(arguments: &[OsString]) -> Option<io::Result<()>> {
dispatch_exact(PRIVATE_SESSION_IDENTITY_ROLE, arguments)
}
fn dispatch_exact(role: &str, arguments: &[OsString]) -> Option<io::Result<()>> {
(arguments.first() == Some(&OsString::from(role))).then(|| {
dispatch_private_sandbox_init(arguments)
.unwrap_or_else(|| Err(io::Error::other("private role arguments are malformed")))
})
}
static ROLES: [PrivateRole; 8] = [
PrivateRole::new(PRIVATE_CLIENT_BOOTSTRAP_ROLE, client_bootstrap),
PrivateRole::new(PRIVATE_SANDBOX_INIT_ROLE, sandbox_init),
PrivateRole::new(PRIVATE_SESSION_EXEC_ROLE, session_exec),
PrivateRole::new(PRIVATE_SESSION_IDENTITY_ROLE, session_identity),
PrivateRole::new(
PRIVATE_STATE_ANCHOR_RECOVERY_ROLE,
dispatch_private_state_anchor_recovery,
),
PrivateRole::new(PRIVATE_PROCESS_ROLE, dispatch_private_process_role),
PrivateRole::new(PRIVATE_ROUTER_ROLE, dispatch_private_router),
PrivateRole::new(PRIVATE_RUNTIME_ROLE, dispatch_private_runtime_role),
];
pub static REGISTRY: PrivateRoleRegistry = PrivateRoleRegistry::new(&ROLES);
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn host_coordinator_declares_exact_ordered_roles() {
assert_eq!(
REGISTRY.names(),
[
PRIVATE_CLIENT_BOOTSTRAP_ROLE,
PRIVATE_SANDBOX_INIT_ROLE,
PRIVATE_SESSION_EXEC_ROLE,
PRIVATE_SESSION_IDENTITY_ROLE,
PRIVATE_STATE_ANCHOR_RECOVERY_ROLE,
PRIVATE_PROCESS_ROLE,
PRIVATE_ROUTER_ROLE,
PRIVATE_RUNTIME_ROLE,
]
);
}
}

498
crates/lazaret-host-coordinator/src/state_maintenance.rs
View file

@ -1,498 +0,0 @@
use std::ffi::OsString;
use std::fmt;
use std::os::unix::ffi::OsStringExt;
use std::path::{Path, PathBuf};
use lazaret_core::configuration::ResolvedDomainConfiguration;
use lazaret_core::records::{
DurableDecision, ParticipantState, SessionPhase, SessionTerminalState,
};
pub use lazaret_state_anchor::StateAnchorEntryKind;
use lazaret_state_anchor::{
StateAnchorEntry, StateAnchorInspection, StateAnchorLayout, StateAnchorResetScope,
require_no_orphan_reset_quarantine, require_state_anchor_absent,
};
use crate::domain_store::{HostCoordinatorDomainStore, StateResetIntentV1, StateResetScopeV1};
#[derive(Debug)]
pub struct StateMaintenanceError(String);
impl fmt::Display for StateMaintenanceError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
impl std::error::Error for StateMaintenanceError {}
fn failure(error: impl fmt::Display) -> StateMaintenanceError {
StateMaintenanceError(error.to_string())
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StateAnchorReport {
pub container_path: String,
pub changes: Vec<StateAnchorEntry>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DomainStateReport {
pub profile: String,
pub workspace: Vec<u8>,
pub domain_id: [u8; 32],
pub anchors: Vec<StateAnchorReport>,
}
/// Returns an empty report only after proving that no physical State Anchor exists.
///
/// # Errors
/// Refuses changed authority, links, inaccessible paths, or any persistent or runtime anchor.
pub fn inspect_absent_domain_state(
configuration: &ResolvedDomainConfiguration,
) -> Result<DomainStateReport, StateMaintenanceError> {
require_state_anchor_absent(configuration).map_err(failure)?;
Ok(report(
configuration,
&StateAnchorInspection {
entries: Vec::new(),
},
))
}
/// Recovers a durable reset before any maintenance observer or runtime mount can expose state.
///
/// # Errors
/// Refuses malformed intent, changed authority, unsupported scope, orphan quarantine, or cleanup
/// failure.
pub fn recover_state_reset(
configuration: &ResolvedDomainConfiguration,
store: &HostCoordinatorDomainStore,
) -> Result<(), StateMaintenanceError> {
let Some(intent) = store.state_reset_intent().map_err(failure)? else {
return require_no_orphan_reset_quarantine(configuration).map_err(failure);
};
require_intent(configuration, &intent)?;
let layout = StateAnchorLayout::create(configuration).map_err(failure)?;
complete_intent(&layout, &intent)?;
store.finish_state_reset(&intent).map_err(failure)
}
/// Returns content-free State Anchor metadata after mandatory recovery and structural checks.
///
/// # Errors
/// Refuses active or malformed Domain state, changed authority, mounts, links in fixed paths, and
/// unsupported upper-layer entries.
pub fn inspect_domain_state(
configuration: &ResolvedDomainConfiguration,
store: &HostCoordinatorDomainStore,
) -> Result<DomainStateReport, StateMaintenanceError> {
recover_state_reset(configuration, store)?;
require_inactive(store)?;
let layout = StateAnchorLayout::create(configuration).map_err(failure)?;
let inspection = layout.inspect_state().map_err(failure)?;
Ok(report(configuration, &inspection))
}
/// Commits and completes one all-or-selected State Anchor reset generation.
///
/// # Errors
/// Refuses active state, an arbitrary target, malformed durable state, changed authority, mounts,
/// links in fixed paths, quarantine collisions, and unsupported upper-layer entries.
pub fn reset_domain_state(
configuration: &ResolvedDomainConfiguration,
store: &HostCoordinatorDomainStore,
scope: StateResetScopeV1,
) -> Result<DomainStateReport, StateMaintenanceError> {
recover_state_reset(configuration, store)?;
require_inactive(store)?;
validate_scope(configuration, &scope)?;
let intent = StateResetIntentV1::new(configuration.domain_id(), scope);
store.begin_state_reset(&intent).map_err(failure)?;
let layout = StateAnchorLayout::create(configuration).map_err(failure)?;
complete_intent(&layout, &intent)?;
store.finish_state_reset(&intent).map_err(failure)?;
inspect_domain_state(configuration, store)
}
fn complete_intent(
layout: &StateAnchorLayout<'_>,
intent: &StateResetIntentV1,
) -> Result<(), StateMaintenanceError> {
match intent.scope() {
StateResetScopeV1::All => layout
.complete_state_reset(StateAnchorResetScope::All)
.map_err(failure),
StateResetScopeV1::Selected(path) => {
let path = PathBuf::from(OsString::from_vec(path.clone()));
layout
.complete_state_reset(StateAnchorResetScope::Selected(&path))
.map_err(failure)
}
}
}
fn require_intent(
configuration: &ResolvedDomainConfiguration,
intent: &StateResetIntentV1,
) -> Result<(), StateMaintenanceError> {
if intent.domain_id() != configuration.domain_id() {
return Err(failure(
"reset intent Domain differs from resolved authority",
));
}
validate_scope(configuration, intent.scope())
}
fn validate_scope(
configuration: &ResolvedDomainConfiguration,
scope: &StateResetScopeV1,
) -> Result<(), StateMaintenanceError> {
let StateResetScopeV1::Selected(bytes) = scope else {
return Ok(());
};
let selected = PathBuf::from(OsString::from_vec(bytes.clone()));
if configuration
.fingerprint_record()
.persistent_dirs
.iter()
.any(|configured| Path::new(configured) == selected)
{
Ok(())
} else {
Err(failure(
"reset target is not an exact configured persistent container path",
))
}
}
fn require_inactive(store: &HostCoordinatorDomainStore) -> Result<(), StateMaintenanceError> {
let snapshot = store.snapshot().map_err(failure)?;
let inactive = snapshot
.generations
.iter()
.all(|generation| generation.durable_decision == DurableDecision::Clean)
&& snapshot
.participants
.iter()
.all(|participant| participant.state() == ParticipantState::Clean)
&& snapshot.sessions.iter().all(|session| {
matches!(
session.session_phase,
SessionPhase::Closed | SessionPhase::Failed
)
})
&& snapshot
.session_terminals
.iter()
.all(|terminal| terminal.state() == SessionTerminalState::Clean)
&& snapshot
.mailboxes
.iter()
.all(|mailbox| mailbox.state == ParticipantState::Clean);
if inactive {
Ok(())
} else {
Err(failure("Domain is active or has unfinished runtime state"))
}
}
fn report(
configuration: &ResolvedDomainConfiguration,
inspection: &StateAnchorInspection,
) -> DomainStateReport {
let anchors = configuration
.fingerprint_record()
.persistent_dirs
.iter()
.map(|container_path| {
let relative = container_path.trim_start_matches('/').as_bytes();
let path = Path::new(container_path.trim_start_matches('/'));
let mut whiteout = path
.parent()
.unwrap_or(Path::new(""))
.as_os_str()
.as_encoded_bytes()
.to_vec();
if !whiteout.is_empty() {
whiteout.push(b'/');
}
whiteout.extend_from_slice(b".wh.");
whiteout.extend_from_slice(
path.file_name()
.expect("validated persistent path has a name")
.as_encoded_bytes(),
);
let changes = inspection
.entries
.iter()
.filter(|entry| {
entry.path == relative
|| entry.path == whiteout
|| entry
.path
.strip_prefix(relative)
.is_some_and(|suffix| suffix.first() == Some(&b'/'))
})
.cloned()
.collect();
StateAnchorReport {
container_path: container_path.clone(),
changes,
}
})
.collect();
DomainStateReport {
profile: configuration.selected_profile().to_owned(),
workspace: configuration
.workspace()
.canonical_label()
.as_bytes()
.to_vec(),
domain_id: *configuration.domain_id().as_bytes(),
anchors,
}
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use std::fs::{self, File, OpenOptions};
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::symlink;
use std::path::{Path, PathBuf};
use lazaret_core::configuration::{
ConfigurationLoadRequest, ResolvedDomainConfiguration, load_trusted_configuration,
};
use lazaret_core::identity::{
ConfigurationFingerprint, CoordinatorId, GenerationId, NetworkSidecarId, SandboxId,
};
use lazaret_core::records::{Domain, Generation};
use lazaret_linux::filesystem::try_exclusive_flock;
use lazaret_runtime_supervisor::sandbox::default_sandbox_layout;
use rustix::fs::{FileType, Mode, XattrFlags, fsetxattr, makedev, mknodat};
use super::*;
#[derive(Clone, Copy)]
enum EntryCase {
Directory,
File,
Symlink,
WhiteoutFile,
WhiteoutDevice,
}
#[test]
fn every_reset_boundary_recovers_to_new_empty() {
for entry_case in [
EntryCase::Directory,
EntryCase::File,
EntryCase::Symlink,
EntryCase::WhiteoutFile,
EntryCase::WhiteoutDevice,
] {
for boundary in 0..9 {
let root = tempfile::tempdir().unwrap();
let configuration = test_configuration(root.path());
let store = open_store(&configuration);
let layout = StateAnchorLayout::create(&configuration).unwrap();
let anchor = domain_root(&configuration).join("state-anchor");
let upper = anchor.join("upper");
fs::create_dir(upper.join("unrelated")).unwrap();
fs::write(upper.join("unrelated/value"), b"unchanged").unwrap();
create_entry(&upper, entry_case);
let intent = StateResetIntentV1::new(
configuration.domain_id(),
StateResetScopeV1::Selected(b"/usr".to_vec()),
);
store.begin_state_reset(&intent).unwrap();
stage_boundary(&anchor, &upper, entry_case, boundary);
if boundary == 8 {
store.finish_state_reset(&intent).unwrap();
}
drop(layout);
recover_state_reset(&configuration, &store).unwrap();
assert!(!upper.join("usr").exists());
assert!(!upper.join("usr").is_symlink());
assert!(!upper.join(".wh.usr").exists());
assert_eq!(
fs::read(upper.join("unrelated/value")).unwrap(),
b"unchanged"
);
assert!(store.state_reset_intent().unwrap().is_none());
assert!(!anchor.join("reset-quarantine").exists());
}
}
}
#[test]
fn corrupt_mismatched_and_orphan_recovery_state_refuses() {
let root = tempfile::tempdir().unwrap();
let configuration = test_configuration(root.path());
let store = open_store(&configuration);
let foreign = StateResetIntentV1::new(
configuration.domain_id(),
StateResetScopeV1::Selected(b"/etc".to_vec()),
);
store.begin_state_reset(&foreign).unwrap();
assert!(recover_state_reset(&configuration, &store).is_err());
let root = tempfile::tempdir().unwrap();
let configuration = test_configuration(root.path());
let store = open_store(&configuration);
StateAnchorLayout::create(&configuration).unwrap();
fs::create_dir(domain_root(&configuration).join("state-anchor/reset-quarantine")).unwrap();
assert!(recover_state_reset(&configuration, &store).is_err());
}
#[test]
fn active_domain_refuses_inspection_and_reset() {
let root = tempfile::tempdir().unwrap();
let configuration = test_configuration(root.path());
let store = open_store(&configuration);
store
.apply(&[crate::domain_store::DomainStoreMutation::PutGeneration(
Generation {
domain_id: configuration.domain_id(),
generation_id: GenerationId::new(1),
configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
coordinator_id: CoordinatorId::new([3; 32]),
network_sidecar_id: NetworkSidecarId::new([4; 32]),
sandbox_id: SandboxId::new([5; 32]),
durable_decision: DurableDecision::None,
},
)])
.unwrap();
assert!(inspect_domain_state(&configuration, &store).is_err());
assert!(reset_domain_state(&configuration, &store, StateResetScopeV1::All).is_err());
assert!(store.state_reset_intent().unwrap().is_none());
}
fn create_entry(upper: &Path, entry_case: EntryCase) {
match entry_case {
EntryCase::Directory => {
fs::create_dir(upper.join("usr")).unwrap();
fs::write(upper.join("usr/value"), b"state").unwrap();
fs::write(upper.join("usr/.wh..wh..opq"), b"").unwrap();
let directory = File::open(upper.join("usr")).unwrap();
fsetxattr(
directory,
"user.fuseoverlayfs.opaque",
b"y",
XattrFlags::empty(),
)
.unwrap();
}
EntryCase::File => fs::write(upper.join("usr"), b"state").unwrap(),
EntryCase::Symlink => symlink("/foreign", upper.join("usr")).unwrap(),
EntryCase::WhiteoutFile => fs::write(upper.join(".wh.usr"), b"").unwrap(),
EntryCase::WhiteoutDevice => mknodat(
rustix::fs::CWD,
upper.join("usr"),
FileType::CharacterDevice,
Mode::empty(),
makedev(0, 0),
)
.unwrap(),
}
}
fn stage_boundary(anchor: &Path, upper: &Path, entry_case: EntryCase, boundary: usize) {
if boundary == 0 {
return;
}
let quarantine = anchor.join("reset-quarantine");
fs::create_dir(&quarantine).unwrap();
if boundary == 1 {
return;
}
let (source, destination) = match entry_case {
EntryCase::WhiteoutFile => (upper.join(".wh.usr"), quarantine.join("whiteout")),
_ => (upper.join("usr"), quarantine.join("entry")),
};
fs::rename(source, &destination).unwrap();
if boundary <= 3 {
return;
}
remove_entry(&destination);
if boundary <= 5 {
return;
}
fs::remove_dir(&quarantine).unwrap();
}
fn remove_entry(path: &Path) {
let metadata = fs::symlink_metadata(path).unwrap();
if metadata.file_type().is_dir() {
fs::remove_dir_all(path).unwrap();
} else {
fs::remove_file(path).unwrap();
}
}
fn open_store(configuration: &ResolvedDomainConfiguration) -> HostCoordinatorDomainStore {
let root = domain_root(configuration);
let lock = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(root.join("domain.lock"))
.unwrap();
HostCoordinatorDomainStore::open(
try_exclusive_flock(lock).unwrap(),
&root.join("domain.redb"),
&Domain {
domain_id: configuration.domain_id(),
profile_name: configuration.selected_profile().to_owned(),
canonical_workspace: configuration.workspace().canonical_label().clone(),
},
)
.unwrap()
}
fn domain_root(configuration: &ResolvedDomainConfiguration) -> PathBuf {
PathBuf::from(std::ffi::OsString::from_vec(
configuration
.persistent_domain_root()
.canonical_label()
.as_bytes()
.to_vec(),
))
}
fn test_configuration(root: &Path) -> ResolvedDomainConfiguration {
let configuration_directory = root.join("configuration");
let workspace = root.join("workspace");
let runtime = root.join("runtime");
fs::create_dir(&configuration_directory).unwrap();
fs::create_dir(&workspace).unwrap();
fs::create_dir(&runtime).unwrap();
let configuration_path = configuration_directory.join("lazaret.toml");
fs::write(
&configuration_path,
format!(
"schema = \"lazaret-trusted-host-config-v1\"\n[storage]\ndata_root = \"{}\"\n[network]\ndefault_allow = [\"192.0.2.10/32\"]\n[profiles.default]\nimage = \"example.invalid/state:v1\"\npersistent_dirs = [\"/usr\"]\n[profiles.default.readiness]\ndenied = [{{ address = \"10.20.30.40\", port = 8443 }}]\npermitted = [{{ address = \"192.0.2.10\", port = 443 }}]\ndns = [{{ resolver = \"198.51.100.53\", query = \"example.invalid\" }}]\n",
root.join("data").display()
),
)
.unwrap();
load_trusted_configuration(ConfigurationLoadRequest {
explicit_configuration_path: Some(configuration_path),
selected_profile: "default".into(),
workspace,
home: root.join("home"),
xdg_config_home: None,
xdg_data_home: None,
xdg_runtime_dir: runtime,
environment: BTreeMap::new(),
sandbox_layout: default_sandbox_layout().unwrap(),
})
.unwrap()
}
}

55
crates/lazaret-host-coordinator/src/waiter.rs
View file

@ -1,6 +1,6 @@
use std::fmt;
use std::fs::File;
use std::io::{self, Read, Write};
use std::io::{self, Read};
use std::os::fd::AsFd;
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
@ -9,37 +9,20 @@ use std::path::PathBuf;
use lazaret_core::codec;
use lazaret_core::configuration::ResolvedDomainConfiguration;
use lazaret_core::identity::{
ClientId, ConfigurationFingerprint, DomainId, GenerationId, SessionAnchorId, SessionId,
use lazaret_core::identity::{ConfigurationFingerprint, DomainId, GenerationId, SessionId};
pub use lazaret_core::protocol::{
BOOTSTRAP_WAITER_SOCKET_NAME, BootstrapWaitOutcome, BootstrapWaitRequest,
};
use lazaret_core::protocol::{read_waiter_frame, write_waiter_frame};
use lazaret_core::records::{ProtocolVersion, Session, SessionPhase};
use rustix::net::sockopt::socket_peercred;
use serde::{Deserialize, Serialize};
use serde::Serialize;
use crate::lifecycle::{HostCoordinatorLifecycle, LifecycleAdapter, SessionLifecycleAdapter};
pub const BOOTSTRAP_WAITER_SOCKET_NAME: &str = "wait.sock";
const MAX_FRAME_BYTES: usize = 4_096;
const MAX_WAITERS: usize = 64;
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub struct BootstrapWaitRequest {
pub protocol_version: ProtocolVersion,
pub domain_id: DomainId,
pub configuration_fingerprint: ConfigurationFingerprint,
pub cli_id: ClientId,
pub session_id: SessionId,
pub session_anchor_id: SessionAnchorId,
}
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
pub enum BootstrapWaitOutcome {
Waiting,
Published,
Failed,
Refused,
}
#[derive(Debug)]
pub struct BootstrapWaiterError(String);
@ -398,14 +381,7 @@ pub fn write_frame<T: Serialize>(
stream: &mut UnixStream,
record: &T,
) -> Result<(), BootstrapWaiterError> {
let encoded = codec::encode(record).map_err(failure)?;
if encoded.len() > MAX_FRAME_BYTES {
return Err(failure("bootstrap waiter frame is oversized"));
}
let length = u32::try_from(encoded.len()).map_err(failure)?;
stream.write_all(&length.to_be_bytes()).map_err(failure)?;
stream.write_all(&encoded).map_err(failure)?;
stream.flush().map_err(failure)
write_waiter_frame(stream, record).map_err(failure)
}
/// # Errors
@ -414,15 +390,7 @@ pub fn read_frame<T>(stream: &mut UnixStream) -> Result<T, BootstrapWaiterError>
where
T: serde::de::DeserializeOwned + Serialize,
{
let mut length = [0_u8; 4];
stream.read_exact(&mut length).map_err(failure)?;
let length = u32::from_be_bytes(length) as usize;
if length > MAX_FRAME_BYTES {
return Err(failure("bootstrap waiter frame is oversized"));
}
let mut encoded = vec![0_u8; length];
stream.read_exact(&mut encoded).map_err(failure)?;
codec::decode(&encoded).map_err(failure)
read_waiter_frame(stream).map_err(failure)
}
#[cfg(test)]
@ -433,10 +401,11 @@ mod tests {
use std::thread;
use lazaret_core::configuration::{ConfigurationLoadRequest, load_trusted_configuration};
use lazaret_core::identity::{CoordinatorId, NetworkSidecarId, SandboxId};
use lazaret_core::identity::{
ClientId, CoordinatorId, NetworkSidecarId, SandboxId, SessionAnchorId,
};
use lazaret_core::records::{Domain, DurableDecision, Generation};
use lazaret_linux::filesystem::try_exclusive_flock;
use lazaret_runtime_supervisor::sandbox::default_sandbox_layout;
use crate::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
use crate::lifecycle::{LifecycleAction, SessionAdmissionIdentity};
@ -757,7 +726,7 @@ mod tests {
xdg_data_home: None,
xdg_runtime_dir: runtime,
environment: BTreeMap::new(),
sandbox_layout: default_sandbox_layout().unwrap(),
sandbox_layout: lazaret_core::configuration::SandboxLayoutV1::default_layout().unwrap(),
})
.unwrap()
}

164
crates/lazaret-host-coordinator/tests/client-control-face.rs Normal file
View file

@ -0,0 +1,164 @@
use std::fs::OpenOptions;
use std::os::unix::fs::PermissionsExt;
use std::path::PathBuf;
use std::sync::Arc;
use lazaret_client::{ControlEndpoint, execute_over_control_face};
use lazaret_core::identity::{
ConfigurationFingerprint, CoordinatorId, DomainId, GenerationId, HostPath, NetworkSidecarId,
SandboxId,
};
use lazaret_core::records::{
Domain, DurableDecision, Generation, GenerationIdentity, NonInteractiveOutput,
NonInteractiveSessionMessage, NonInteractiveSessionRequest, SessionExit,
};
use lazaret_host_coordinator::control::{
ControlError, NonInteractiveSessionExecutor, PublishedControlFace, TrustAnchor,
};
use lazaret_host_coordinator::domain_store::{DomainStoreMutation, HostCoordinatorDomainStore};
use lazaret_linux::filesystem::try_exclusive_flock;
struct TestDirectory(PathBuf);
impl TestDirectory {
fn new() -> Self {
let path = std::env::temp_dir().join(format!(
"lazaret-client-control-{}-{}",
std::process::id(),
random_suffix()
));
std::fs::create_dir(&path).unwrap();
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700)).unwrap();
Self(path)
}
}
impl Drop for TestDirectory {
fn drop(&mut self) {
std::fs::remove_dir_all(&self.0).unwrap();
}
}
struct Executor;
impl NonInteractiveSessionExecutor for Executor {
fn execute(
&self,
request: &NonInteractiveSessionRequest,
emit: &mut dyn FnMut(NonInteractiveSessionMessage) -> Result<(), ControlError>,
) -> Result<SessionExit, ControlError> {
assert_eq!(request.argv, [b"tool".to_vec(), b"argument".to_vec()]);
emit(NonInteractiveSessionMessage::Stdout(NonInteractiveOutput {
session_id: request.session_id,
bytes: b"first".to_vec(),
}))?;
emit(NonInteractiveSessionMessage::Stderr(NonInteractiveOutput {
session_id: request.session_id,
bytes: b"second".to_vec(),
}))?;
Ok(SessionExit::exited(request.session_id, 19))
}
}
#[test]
fn client_enrolls_registers_forwards_both_streams_and_returns_guest_status() {
let directory = TestDirectory::new();
let identity = generation_identity();
let store = open_store(&directory, &identity);
let face = Arc::new(
PublishedControlFace::publish_with_executor(
&directory.0,
rustix::process::getuid().as_raw(),
TrustAnchor::create(identity).unwrap(),
store,
Arc::new(Executor),
)
.unwrap(),
);
let endpoint = ControlEndpoint::new(
face.socket_path().to_path_buf(),
face.publication_identity().clone(),
);
let server = {
let face = Arc::clone(&face);
std::thread::spawn(move || {
face.serve_next().unwrap();
face.serve_next().unwrap();
})
};
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let status = execute_over_control_face(
&endpoint,
&[b"tool".to_vec(), b"argument".to_vec()],
&mut stdout,
&mut stderr,
)
.unwrap();
server.join().unwrap();
assert_eq!(status, 19);
assert_eq!(stdout, b"first");
assert_eq!(stderr, b"second");
}
fn generation_identity() -> GenerationIdentity {
GenerationIdentity {
agent_profile: "trusted".into(),
domain_id: DomainId::new([1; 32]),
generation_id: GenerationId::new(1),
configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
coordinator_id: CoordinatorId::new([3; 32]),
sandbox_id: SandboxId::new([4; 32]),
}
}
fn open_store(
directory: &TestDirectory,
identity: &GenerationIdentity,
) -> Arc<HostCoordinatorDomainStore> {
let lock = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(directory.0.join("domain.lock"))
.unwrap();
let domain = Domain {
domain_id: identity.domain_id,
profile_name: identity.agent_profile.clone(),
canonical_workspace: HostPath::new(b"/workspace/project".to_vec()),
};
let store = Arc::new(
HostCoordinatorDomainStore::open(
try_exclusive_flock(lock).unwrap(),
&directory.0.join("domain.redb"),
&domain,
)
.unwrap(),
);
for durable_decision in [
DurableDecision::None,
DurableDecision::Preparing,
DurableDecision::Published,
] {
store
.apply(&[DomainStoreMutation::PutGeneration(Generation {
domain_id: identity.domain_id,
generation_id: identity.generation_id,
configuration_fingerprint: identity.configuration_fingerprint,
coordinator_id: identity.coordinator_id,
network_sidecar_id: NetworkSidecarId::new([5; 32]),
sandbox_id: identity.sandbox_id,
durable_decision,
})])
.unwrap();
}
store
}
fn random_suffix() -> u64 {
let mut bytes = [0_u8; 8];
getrandom::fill(&mut bytes).unwrap();
u64::from_ne_bytes(bytes)
}

10
crates/lazaret-host-coordinator/tests/escalation.rs
View file

@ -28,9 +28,7 @@ use lazaret_host_coordinator::escalation::{
GenerationMailbox,
};
use lazaret_linux::filesystem::RegularFileMailbox;
use lazaret_linux::process::{
ProcessOwnershipContext, dispatch_private_process_role, recover_process,
};
use lazaret_linux::process::{ProcessOwnershipContext, recover_process};
use sha2::{Digest, Sha256};
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
@ -882,9 +880,9 @@ fn run() -> TestResult {
fn main() {
let arguments = std::env::args_os().skip(1).collect::<Vec<_>>();
if let Some(result) = dispatch_private_process_role(&arguments) {
if let Err(error) = result {
eprintln!("private process role failed: {error}");
if let Some(outcome) = lazaret_host_coordinator::private_roles::REGISTRY.dispatch(&arguments) {
if let Err(error) = outcome.result {
eprintln!("private role {} failed: {error}", outcome.name);
std::process::exit(1);
}
return;

6
crates/lazaret-host-coordinator/tests/recovery-process.rs
View file

@ -17,13 +17,13 @@ use lazaret_core::records::{
use lazaret_host_coordinator::recovery::{
delete_runsc_container, prove_runtime_owners_absent, runtime_recovery_ownership,
};
use lazaret_linux::process::{TargetCommand, child_broker, dispatch_private_process_role};
use lazaret_linux::process::{TargetCommand, child_broker};
use rustix::process::Signal;
fn main() -> ExitCode {
let arguments = std::env::args_os().skip(1).collect::<Vec<_>>();
if let Some(result) = dispatch_private_process_role(&arguments) {
return exit(result);
if let Some(outcome) = lazaret_host_coordinator::private_roles::REGISTRY.dispatch(&arguments) {
return exit(outcome.result);
}
exit(run())
}

13
crates/lazaret-host-coordinator/tests/runtime-composition.rs
View file

@ -46,7 +46,6 @@ use lazaret_runtime_supervisor::sandbox::{
};
use lazaret_runtime_supervisor::supervision::{
AttachedRunscSignal, RuntimeNetworkLaunch, RuntimeSupervisorLaunch,
dispatch_private_runtime_role,
};
use rustix::process::Signal;
use tempfile::TempDir;
@ -64,16 +63,8 @@ const BENCHMARK_ERROR: &str = "benchmark-error";
fn main() -> ExitCode {
let arguments = std::env::args_os().skip(1).collect::<Vec<_>>();
if let Some(result) = lazaret_linux::process::dispatch_private_process_role(&arguments) {
return exit(result);
}
if let Some(result) = dispatch_private_runtime_role(&arguments) {
return exit(result);
}
if let Some(result) =
lazaret_host_coordinator::sandbox_init::dispatch_private_sandbox_init(&arguments)
{
return exit(result);
if let Some(outcome) = lazaret_host_coordinator::private_roles::REGISTRY.dispatch(&arguments) {
return exit(outcome.result);
}
if arguments.first().and_then(|argument| argument.to_str()) == Some(BENCHMARK_GUEST) {
return exit(run_benchmark_guest(&arguments[1..]));