lazaret/crates/lazaret-host-coordinator/src/control.rs

use std::collections::BTreeMap;
use std::fmt;
use std::fs::{File, OpenOptions};
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::os::fd::AsFd;
use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};

use ed25519_dalek::SigningKey;
use lazaret_core::codec::{CodecError, decode, encode};
use lazaret_core::identity::{
    CertificateDigest, ClientId, SessionAnchorId, SessionId, TrustManifestDigest,
};
use lazaret_core::protocol::{
    ProtocolValidationError, issue_protocol_credential, issue_session_capability,
    issue_trust_manifest, verify_tls_credential,
};
use lazaret_core::records::{
    ClientEnrollmentRequest, ClientEnrollmentResponse, ControlRequest, ControlResponse,
    DurableDecision, GenerationIdentity, ProtocolCredential, ProtocolCredentialRole,
    ProtocolPublicationIdentity, ProtocolSubjectId, ProtocolVersion, SandboxCredentialBundle,
    Session, SessionBinding, SessionPhase, SessionRegistrationRequest, SessionRegistrationResponse,
    TlsCredential, TlsRole, TlsSubjectId, TrustManifest,
};
use rcgen::{
    BasicConstraints, CertificateParams, CertificateSigningRequestParams, CertifiedIssuer,
    DistinguishedName, DnType, DnValue, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose,
    PKCS_ED25519,
};
use rustix::fs::{Mode, OFlags, fchmod, openat};
use rustix::net::sockopt::socket_peercred;
use rustls::pki_types::{
    CertificateDer, CertificateSigningRequestDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName,
};
use rustls::server::{NoServerSessionStorage, WebPkiClientVerifier};
use rustls::{ClientConfig, ClientConnection, RootCertStore, ServerConfig, ServerConnection};
use sha2::{Digest, Sha256};
use time::{Duration, OffsetDateTime};

use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
use crate::lifecycle::{
    HostCoordinatorLifecycle, LifecycleAction, SessionAdmissionIdentity, SessionLifecycleAdapter,
};

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;
const CONTROL_SOCKET_NAME: &str = "control.sock";
const PUBLICATION_IDENTITY_NAME: &str = "protocol-identity.postcard";
const STAGED_CONTROL_SOCKET_NAME: &str = ".control.sock.staged";
const STAGED_PUBLICATION_IDENTITY_NAME: &str = ".protocol-identity.staged";

#[derive(Debug)]
pub enum ControlError {
    Io(io::Error),
    Codec(CodecError),
    Certificate(rcgen::Error),
    Tls(rustls::Error),
    Protocol(ProtocolValidationError),
    DomainStore(DomainStoreError),
    InvalidPath(&'static str),
    InvalidPublication(&'static str),
    UnauthorizedPeer,
    InvalidCertificateProfile,
    StaleGeneration,
    DuplicateIdentity,
    InvalidRequest,
    OversizeRequest { length: usize, maximum: usize },
    LockPoisoned,
    Lifecycle(String),
}

impl fmt::Display for ControlError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Io(error) => write!(formatter, "control I/O failed: {error}"),
            Self::Codec(error) => write!(formatter, "control record failed validation: {error}"),
            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::DomainStore(error) => write!(formatter, "session persistence failed: {error}"),
            Self::InvalidPath(reason) => write!(formatter, "invalid control path: {reason}"),
            Self::InvalidPublication(reason) => {
                write!(formatter, "invalid control publication: {reason}")
            }
            Self::UnauthorizedPeer => formatter.write_str("control peer UID is not authorized"),
            Self::InvalidCertificateProfile => {
                formatter.write_str("TLS certificate profile is invalid")
            }
            Self::StaleGeneration => formatter.write_str("control generation is stale"),
            Self::DuplicateIdentity => formatter.write_str("session identity is already active"),
            Self::InvalidRequest => {
                formatter.write_str("operation is not allowed on this connection")
            }
            Self::OversizeRequest { length, maximum } => {
                write!(
                    formatter,
                    "control request length {length} exceeds {maximum}"
                )
            }
            Self::LockPoisoned => formatter.write_str("control registry lock is poisoned"),
            Self::Lifecycle(detail) => write!(formatter, "control lifecycle failed: {detail}"),
        }
    }
}

impl std::error::Error for ControlError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Io(error) => Some(error),
            Self::Codec(error) => Some(error),
            Self::Certificate(error) => Some(error),
            Self::Tls(error) => Some(error),
            Self::Protocol(error) => Some(error),
            Self::DomainStore(error) => Some(error),
            _ => None,
        }
    }
}

macro_rules! from_error {
    ($source:ty, $variant:ident) => {
        impl From<$source> for ControlError {
            fn from(error: $source) -> Self {
                Self::$variant(error)
            }
        }
    };
}

from_error!(io::Error, Io);
from_error!(CodecError, Codec);
from_error!(rcgen::Error, Certificate);
from_error!(rustls::Error, Tls);
from_error!(ProtocolValidationError, Protocol);
from_error!(DomainStoreError, DomainStore);

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct FileIdentity {
    device: u64,
    inode: u64,
}

impl FileIdentity {
    fn from_metadata(metadata: &std::fs::Metadata) -> Self {
        Self {
            device: metadata.dev(),
            inode: metadata.ino(),
        }
    }
}

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>,
    tls_ca_der: CertificateDer<'static>,
    coordinator_tls_credential: TlsCredential,
    coordinator_tls_private_key: PrivatePkcs8KeyDer<'static>,
    trust_manifest: TrustManifest,
    trust_manifest_digest: TrustManifestDigest,
    coordinator_protocol_credential: ProtocolCredential,
    sandbox_protocol_credential: ProtocolCredential,
    coordinator_protocol_key: SigningKey,
    sandbox_protocol_key: Option<SigningKey>,
}

impl TrustAnchor {
    #[must_use]
    pub const fn generation_identity(&self) -> &GenerationIdentity {
        &self.generation_identity
    }

    /// Creates one ephemeral generation Trust Anchor with role-separated Ed25519 keys.
    ///
    /// # Errors
    /// Returns an error when key generation, certificate issuance, or protocol signing fails.
    pub fn create(generation_identity: GenerationIdentity) -> Result<Self, ControlError> {
        let now = OffsetDateTime::now_utc();
        let mut ca_params = CertificateParams::default();
        ca_params.distinguished_name = DistinguishedName::new();
        ca_params
            .distinguished_name
            .push(DnType::CommonName, "Lazaret ephemeral TLS CA");
        ca_params.not_before = now - Duration::minutes(1);
        ca_params.not_after = now + Duration::hours(24);
        ca_params.is_ca = IsCa::Ca(BasicConstraints::Unconstrained);
        ca_params.key_usages = vec![
            KeyUsagePurpose::DigitalSignature,
            KeyUsagePurpose::KeyCertSign,
            KeyUsagePurpose::CrlSign,
        ];
        let ca_key = KeyPair::generate_for(&PKCS_ED25519)?;
        let tls_ca = CertifiedIssuer::self_signed(ca_params, ca_key)?;
        let tls_ca_der = tls_ca.der().clone();
        let tls_ca_digest = certificate_digest(&tls_ca_der);

        let protocol_issuer = random_signing_key()?;
        let (trust_manifest, trust_manifest_digest) =
            issue_trust_manifest(generation_identity.clone(), tls_ca_digest, &protocol_issuer)?;
        let coordinator_protocol_key = random_signing_key()?;
        let sandbox_protocol_key = random_signing_key()?;
        let coordinator_protocol_credential = issue_protocol_credential(
            generation_identity.clone(),
            ProtocolCredentialRole::SessionCoordinator,
            ProtocolSubjectId::SessionCoordinator(generation_identity.coordinator_id),
            &coordinator_protocol_key.verifying_key(),
            &protocol_issuer,
        )?;
        let sandbox_protocol_credential = issue_protocol_credential(
            generation_identity.clone(),
            ProtocolCredentialRole::AgentSandbox,
            ProtocolSubjectId::AgentSandbox(generation_identity.sandbox_id),
            &sandbox_protocol_key.verifying_key(),
            &protocol_issuer,
        )?;

        let coordinator_key = KeyPair::generate_for(&PKCS_ED25519)?;
        let coordinator_certificate = leaf_parameters(
            CONTROL_SERVER_NAME,
            ExtendedKeyUsagePurpose::ServerAuth,
            now,
        )?
        .signed_by(&coordinator_key, &tls_ca)?;
        let coordinator_certificate_der = coordinator_certificate.der().to_vec();
        let coordinator_tls_credential = TlsCredential::session_coordinator(
            generation_identity.clone(),
            generation_identity.coordinator_id,
            coordinator_certificate_der.clone(),
            certificate_digest(coordinator_certificate.der()),
        );

        Ok(Self {
            generation_identity,
            tls_ca,
            tls_ca_der,
            coordinator_tls_credential,
            coordinator_tls_private_key: PrivatePkcs8KeyDer::from(coordinator_key.serialize_der()),
            trust_manifest,
            trust_manifest_digest,
            coordinator_protocol_credential,
            sandbox_protocol_credential,
            coordinator_protocol_key,
            sandbox_protocol_key: Some(sandbox_protocol_key),
        })
    }

    #[must_use]
    pub fn publication_identity(&self) -> ProtocolPublicationIdentity {
        ProtocolPublicationIdentity {
            protocol_version: ProtocolVersion::V1,
            generation_identity: self.generation_identity.clone(),
            tls_ca_certificate_der: self.tls_ca_der.to_vec(),
            trust_manifest_sha256: self.trust_manifest_digest,
            coordinator_tls_certificate_sha256: self.coordinator_tls_credential.certificate_sha256,
        }
    }

    #[must_use]
    pub fn coordinator_protocol_key(&self) -> SigningKey {
        self.coordinator_protocol_key.clone()
    }

    /// Transfers the sandbox credential package once without exposing coordinator or issuer keys.
    ///
    /// # Errors
    /// Returns an error after the sandbox private key has already been transferred.
    pub fn take_sandbox_credential_bundle(
        &mut self,
    ) -> Result<SandboxCredentialBundle, ControlError> {
        let private_key = self
            .sandbox_protocol_key
            .take()
            .ok_or(ControlError::InvalidRequest)?
            .to_bytes()
            .to_vec();
        SandboxCredentialBundle::new(
            self.generation_identity.clone(),
            self.trust_manifest.clone(),
            self.coordinator_protocol_credential.clone(),
            self.sandbox_protocol_credential.clone(),
            private_key,
        )
        .map_err(|_| ControlError::InvalidCertificateProfile)
    }

    fn issue_client_credential(
        &self,
        request: &ClientEnrollmentRequest,
    ) -> Result<TlsCredential, ControlError> {
        if request.protocol_version != ProtocolVersion::V1
            || request.generation_identity != self.generation_identity
        {
            return Err(ControlError::StaleGeneration);
        }
        let csr_der =
            CertificateSigningRequestDer::from(request.certificate_signing_request_der.as_slice());
        let csr = CertificateSigningRequestParams::from_der(&csr_der)?;
        if csr.public_key.algorithm() != &PKCS_ED25519
            || csr.params.distinguished_name.get(&DnType::CommonName)
                != Some(&DnValue::Utf8String(enrollment_binding(
                    &request.generation_identity,
                    request.cli_id,
                )?))
        {
            return Err(ControlError::InvalidCertificateProfile);
        }
        let now = OffsetDateTime::now_utc();
        let params = leaf_parameters(
            &format!("client-{}", short_identity(request.cli_id.as_bytes())),
            ExtendedKeyUsagePurpose::ClientAuth,
            now,
        )?;
        let constrained = CertificateSigningRequestParams {
            params,
            public_key: csr.public_key,
        };
        let certificate = constrained.signed_by(&self.tls_ca)?;
        let certificate_der = certificate.der().to_vec();
        Ok(TlsCredential::client_cli(
            self.generation_identity.clone(),
            request.cli_id,
            certificate_der,
            certificate_digest(certificate.der()),
        ))
    }

    fn server_config(&self) -> Result<ServerConfig, ControlError> {
        let mut roots = RootCertStore::empty();
        roots.add(self.tls_ca_der.clone())?;
        let verifier = WebPkiClientVerifier::builder(Arc::new(roots))
            .allow_unauthenticated()
            .build()
            .map_err(|_| ControlError::InvalidCertificateProfile)?;
        let mut config = ServerConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
            .with_client_cert_verifier(verifier)
            .with_single_cert(
                vec![CertificateDer::from(
                    self.coordinator_tls_credential.certificate_der.clone(),
                )],
                PrivateKeyDer::Pkcs8(self.coordinator_tls_private_key.clone_key()),
            )?;
        config.alpn_protocols = vec![CONTROL_ALPN.to_vec()];
        config.max_early_data_size = 0;
        config.send_tls13_tickets = 0;
        config.session_storage = Arc::new(NoServerSessionStorage {});
        Ok(config)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct RegisteredSession {
    pub binding: SessionBinding,
    pub session_anchor_id: SessionAnchorId,
}

#[derive(Clone)]
pub struct ControlSessionAdmission {
    pub request: SessionRegistrationRequest,
    pub credential: TlsCredential,
}

pub type ControlHostCoordinatorLifecycle<'store, A> =
    HostCoordinatorLifecycle<'store, A, Arc<SessionRegistry>>;

#[derive(Default)]
struct RegistryState {
    enrolled: BTreeMap<CertificateDigest, TlsCredential>,
    connected: BTreeMap<SessionId, RegisteredSession>,
}

pub struct SessionRegistry {
    state: Mutex<RegistryState>,
    store: Arc<HostCoordinatorDomainStore>,
    generation_identity: GenerationIdentity,
}

impl SessionRegistry {
    #[must_use]
    pub fn new(
        store: Arc<HostCoordinatorDomainStore>,
        generation_identity: GenerationIdentity,
    ) -> Self {
        Self {
            state: Mutex::new(RegistryState::default()),
            store,
            generation_identity,
        }
    }

    /// Returns the exact set of connected, registered sessions.
    ///
    /// # Errors
    /// Returns an error if the registry lock is poisoned.
    pub fn connected_sessions(&self) -> Result<Vec<RegisteredSession>, ControlError> {
        Ok(self.lock()?.connected.values().cloned().collect())
    }

    fn lock(&self) -> Result<MutexGuard<'_, RegistryState>, ControlError> {
        self.state.lock().map_err(|_| ControlError::LockPoisoned)
    }

    fn enroll(&self, credential: TlsCredential) -> Result<(), ControlError> {
        self.lock()?
            .enrolled
            .insert(credential.certificate_sha256, credential);
        Ok(())
    }

    fn credential_for_leaf(&self, certificate: &[u8]) -> Result<TlsCredential, ControlError> {
        self.lock()?
            .enrolled
            .get(&certificate_digest(certificate))
            .cloned()
            .ok_or(ControlError::InvalidCertificateProfile)
    }

    fn register(
        &self,
        request: SessionRegistrationRequest,
        credential: &TlsCredential,
    ) -> Result<RegisteredSession, ControlError> {
        let binding = request.session_binding;
        if request.protocol_version != ProtocolVersion::V1
            || binding.generation_identity != self.generation_identity
        {
            return Err(ControlError::StaleGeneration);
        }
        let TlsSubjectId::ClientCli(certificate_cli_id) = credential.subject_id() else {
            return Err(ControlError::InvalidCertificateProfile);
        };
        if credential.role() != TlsRole::ClientCli || certificate_cli_id != binding.cli_id {
            return Err(ControlError::InvalidCertificateProfile);
        }
        verify_tls_credential(
            credential,
            &self.generation_identity,
            TlsRole::ClientCli,
            TlsSubjectId::ClientCli(binding.cli_id),
        )?;
        let anchor_bytes: [u8; 32] = binding
            .session_anchor_public_key
            .as_slice()
            .try_into()
            .map_err(|_| ControlError::InvalidRequest)?;
        ed25519_dalek::VerifyingKey::from_bytes(&anchor_bytes)
            .map_err(|_| ControlError::InvalidRequest)?;
        let session_anchor_id = SessionAnchorId::new(sha256(&anchor_bytes));

        self.require_serviceable_generation()?;
        let snapshot = self.store.snapshot()?;
        let state = self.lock()?;
        let duplicate_in_memory = state.connected.values().any(|registered| {
            registered.binding.session_id == binding.session_id
                || registered.binding.cli_id == binding.cli_id
                || registered.session_anchor_id == session_anchor_id
        });
        let resumable_waiter = snapshot.sessions.iter().any(|session| {
            session.generation_id == binding.generation_identity.generation_id
                && session.session_id == binding.session_id
                && session.cli_id == binding.cli_id
                && session.session_anchor_id == session_anchor_id
                && session.session_phase == SessionPhase::Waiting
        });
        let duplicate_session = snapshot.sessions.iter().any(|session| {
            let same_session_allowed = resumable_waiter
                && session.session_id == binding.session_id
                && (session.session_phase == SessionPhase::Closed
                    || (session.generation_id == binding.generation_identity.generation_id
                        && session.cli_id == binding.cli_id
                        && session.session_anchor_id == session_anchor_id
                        && session.session_phase == SessionPhase::Waiting));
            !same_session_allowed
                && (session.session_id == binding.session_id
                    || (is_live_session(session.session_phase)
                        && (session.cli_id == binding.cli_id
                            || session.session_anchor_id == session_anchor_id)))
        });
        if duplicate_in_memory || duplicate_session {
            return Err(ControlError::DuplicateIdentity);
        }

        let registering = Session {
            generation_id: binding.generation_identity.generation_id,
            session_id: binding.session_id,
            cli_id: binding.cli_id,
            session_anchor_id,
            session_phase: SessionPhase::Registering,
        };
        self.store
            .apply(&[DomainStoreMutation::PutSession(registering.clone())])?;
        let mut attached = registering;
        attached.session_phase = SessionPhase::Attached;
        if let Err(error) = self
            .store
            .apply(&[DomainStoreMutation::PutSession(attached)])
        {
            let mut failed = Session {
                generation_id: binding.generation_identity.generation_id,
                session_id: binding.session_id,
                cli_id: binding.cli_id,
                session_anchor_id,
                session_phase: SessionPhase::Registering,
            };
            failed.session_phase = SessionPhase::Failed;
            let _ = self.store.apply(&[DomainStoreMutation::PutSession(failed)]);
            return Err(error.into());
        }
        let registered = RegisteredSession {
            binding,
            session_anchor_id,
        };
        drop(snapshot);
        drop(state);
        self.lock()?
            .connected
            .insert(registered.binding.session_id, registered.clone());
        Ok(registered)
    }

    fn disconnect(&self, session_id: SessionId) -> Result<(), ControlError> {
        let registered = self.lock()?.connected.remove(&session_id);
        let Some(registered) = registered else {
            return Ok(());
        };
        self.store
            .apply(&[DomainStoreMutation::PutSession(Session {
                generation_id: registered.binding.generation_identity.generation_id,
                session_id,
                cli_id: registered.binding.cli_id,
                session_anchor_id: registered.session_anchor_id,
                session_phase: SessionPhase::Closed,
            })])?;
        Ok(())
    }

    fn require_serviceable_generation(&self) -> Result<(), ControlError> {
        let decision = self.require_generation_authority()?;
        if decision != DurableDecision::Published {
            return Err(ControlError::StaleGeneration);
        }
        Ok(())
    }

    fn require_generation_authority(&self) -> Result<DurableDecision, ControlError> {
        let snapshot = self.store.snapshot()?;
        let generation = snapshot
            .generations
            .iter()
            .find(|generation| generation.generation_id == self.generation_identity.generation_id)
            .ok_or(ControlError::StaleGeneration)?;
        let is_latest = snapshot
            .generations
            .iter()
            .all(|candidate| candidate.generation_id <= self.generation_identity.generation_id);
        if !is_latest
            || snapshot.domain.domain_id != self.generation_identity.domain_id
            || snapshot.domain.profile_name != self.generation_identity.agent_profile
            || generation.domain_id != self.generation_identity.domain_id
            || generation.configuration_fingerprint
                != self.generation_identity.configuration_fingerprint
            || generation.coordinator_id != self.generation_identity.coordinator_id
            || generation.sandbox_id != self.generation_identity.sandbox_id
            || !matches!(
                generation.durable_decision,
                DurableDecision::Preparing | DurableDecision::Published
            )
        {
            return Err(ControlError::StaleGeneration);
        }
        Ok(generation.durable_decision)
    }
}

impl SessionLifecycleAdapter for Arc<SessionRegistry> {
    type Error = ControlError;
    type Admission = ControlSessionAdmission;

    fn admission_identity(&self, admission: &Self::Admission) -> SessionAdmissionIdentity {
        let binding = &admission.request.session_binding;
        SessionAdmissionIdentity {
            generation_id: binding.generation_identity.generation_id,
            session_id: binding.session_id,
            configuration_fingerprint: binding.generation_identity.configuration_fingerprint,
        }
    }

    fn register_session(
        &self,
        _store: &HostCoordinatorDomainStore,
        entry_action: LifecycleAction,
        admission: Self::Admission,
    ) -> Result<Session, Self::Error> {
        if !matches!(
            entry_action,
            LifecycleAction::StartReadySession | LifecycleAction::AdvanceWaitingSession
        ) {
            return Err(ControlError::InvalidRequest);
        }
        let registered = self.register(admission.request, &admission.credential)?;
        Ok(Session {
            generation_id: registered.binding.generation_identity.generation_id,
            session_id: registered.binding.session_id,
            cli_id: registered.binding.cli_id,
            session_anchor_id: registered.session_anchor_id,
            session_phase: SessionPhase::Attached,
        })
    }

    fn disconnect_session(
        &self,
        _store: &HostCoordinatorDomainStore,
        session: &Session,
    ) -> Result<(), Self::Error> {
        let mut state = self.lock()?;
        let registered = state.connected.get(&session.session_id);
        if let Some(registered) = registered
            && (registered.binding.generation_identity.generation_id != session.generation_id
                || registered.binding.cli_id != session.cli_id
                || registered.session_anchor_id != session.session_anchor_id)
        {
            return Err(ControlError::InvalidRequest);
        }
        if session.session_phase == SessionPhase::Attached && registered.is_none() {
            return Err(ControlError::InvalidRequest);
        }
        state.connected.remove(&session.session_id);
        drop(state);
        self.store
            .apply(&[DomainStoreMutation::PutSession(Session {
                session_phase: SessionPhase::Closed,
                ..session.clone()
            })])?;
        Ok(())
    }
}

struct ControlPaths {
    root: PathBuf,
    root_file: File,
    root_identity: FileIdentity,
    socket: PathBuf,
    identity: PathBuf,
    staged_socket: PathBuf,
    staged_identity: PathBuf,
}

impl ControlPaths {
    fn open(root: &Path, uid: u32) -> Result<Self, ControlError> {
        if !root.is_absolute() {
            return Err(ControlError::InvalidPath(
                "runtime Domain root is not absolute",
            ));
        }
        let canonical = std::fs::canonicalize(root)?;
        if canonical != root {
            return Err(ControlError::InvalidPath(
                "runtime Domain root contains a symlink",
            ));
        }
        let metadata = std::fs::symlink_metadata(root)?;
        if !metadata.file_type().is_dir()
            || metadata.uid() != uid
            || metadata.mode() & 0o7777 != 0o700
        {
            return Err(ControlError::InvalidPath(
                "runtime Domain root is not invoking-user-owned mode 0700",
            ));
        }
        let root_file = OpenOptions::new().read(true).open(root)?;
        let root_identity = FileIdentity::from_metadata(&root_file.metadata()?);
        Ok(Self {
            root: root.to_path_buf(),
            root_file,
            root_identity,
            socket: root.join(CONTROL_SOCKET_NAME),
            identity: root.join(PUBLICATION_IDENTITY_NAME),
            staged_socket: root.join(STAGED_CONTROL_SOCKET_NAME),
            staged_identity: root.join(STAGED_PUBLICATION_IDENTITY_NAME),
        })
    }

    fn validate_root(&self, uid: u32) -> Result<(), ControlError> {
        let descriptor = self.root_file.metadata()?;
        let current = std::fs::symlink_metadata(&self.root)?;
        if FileIdentity::from_metadata(&descriptor) != self.root_identity
            || FileIdentity::from_metadata(&current) != self.root_identity
            || !current.file_type().is_dir()
            || current.uid() != uid
            || current.mode() & 0o7777 != 0o700
        {
            return Err(ControlError::InvalidPath("runtime Domain root changed"));
        }
        Ok(())
    }

    fn create_named_identity(
        &self,
        name: &str,
        bytes: &[u8],
        uid: u32,
    ) -> Result<(File, FileIdentity), ControlError> {
        let fd = openat(
            self.root_file.as_fd(),
            name,
            OFlags::CREATE | OFlags::EXCL | OFlags::RDWR | OFlags::CLOEXEC | OFlags::NOFOLLOW,
            Mode::RUSR | Mode::WUSR,
        )
        .map_err(io::Error::from)?;
        let mut file = File::from(fd);
        let identity = FileIdentity::from_metadata(&file.metadata()?);
        let result = (|| {
            fchmod(file.as_fd(), Mode::RUSR | Mode::WUSR).map_err(io::Error::from)?;
            file.write_all(bytes)
                .map_err(|_| ControlError::InvalidPublication("identity write failed"))?;
            file.sync_all()
                .map_err(|_| ControlError::InvalidPublication("identity sync failed"))?;
            self.root_file
                .sync_all()
                .map_err(|_| ControlError::InvalidPublication("directory sync failed"))?;
            self.validate_identity_at(&self.root.join(name), &file, identity, uid, bytes.len())
        })();
        if let Err(error) = result {
            remove_if_identity_matches(&self.root.join(name), identity);
            return Err(error);
        }
        Ok((file, identity))
    }

    fn validate_identity(
        &self,
        file: &File,
        expected: FileIdentity,
        uid: u32,
        expected_length: usize,
    ) -> Result<(), ControlError> {
        self.validate_identity_at(&self.identity, file, expected, uid, expected_length)
    }

    fn validate_identity_at(
        &self,
        path: &Path,
        file: &File,
        expected: FileIdentity,
        uid: u32,
        expected_length: usize,
    ) -> Result<(), ControlError> {
        self.validate_root(uid)?;
        let descriptor = file.metadata()?;
        let current = std::fs::symlink_metadata(path)?;
        if FileIdentity::from_metadata(&descriptor) != expected
            || FileIdentity::from_metadata(&current) != expected
            || !descriptor.file_type().is_file()
            || !current.file_type().is_file()
            || descriptor.uid() != uid
            || current.uid() != uid
            || descriptor.mode() & 0o7777 != 0o600
            || current.mode() & 0o7777 != 0o600
            || usize::try_from(descriptor.len()).ok() != Some(expected_length)
            || current.len() != descriptor.len()
        {
            return Err(ControlError::InvalidPublication(
                "publication identity changed type, ownership, mode, or content length",
            ));
        }
        Ok(())
    }
}

pub struct PublishedControlFace {
    listener: std::os::unix::net::UnixListener,
    paths: ControlPaths,
    socket_identity: FileIdentity,
    identity_file: File,
    identity_file_identity: FileIdentity,
    publication_identity: ProtocolPublicationIdentity,
    publication_bytes: Vec<u8>,
    uid: u32,
    anchor: TrustAnchor,
    server_config: Arc<ServerConfig>,
    registry: Arc<SessionRegistry>,
}

pub struct StagedControlFace {
    listener: Option<std::os::unix::net::UnixListener>,
    paths: ControlPaths,
    staged_socket_identity: FileIdentity,
    identity_file: File,
    identity_file_identity: FileIdentity,
    publication_identity: ProtocolPublicationIdentity,
    publication_bytes: Vec<u8>,
    uid: u32,
    anchor: Option<TrustAnchor>,
    server_config: Arc<ServerConfig>,
    registry: Arc<SessionRegistry>,
}

impl StagedControlFace {
    /// Stages an undiscoverable private control face before durable publication.
    ///
    /// # Errors
    /// Refuses existing artifacts, changed paths, ownership/mode errors, or TLS failure.
    pub fn stage(
        runtime_domain_root: &Path,
        uid: u32,
        anchor: TrustAnchor,
        registry: Arc<SessionRegistry>,
    ) -> Result<Self, ControlError> {
        if registry.generation_identity != anchor.generation_identity {
            return Err(ControlError::StaleGeneration);
        }
        let paths = ControlPaths::open(runtime_domain_root, uid)?;
        if [
            &paths.socket,
            &paths.identity,
            &paths.staged_socket,
            &paths.staged_identity,
        ]
        .into_iter()
        .any(|path| path.try_exists().unwrap_or(true))
        {
            return Err(ControlError::InvalidPublication(
                "publication path already exists",
            ));
        }
        let publication_identity = anchor.publication_identity();
        let publication_bytes = encode(&publication_identity)?;
        let server_config = Arc::new(anchor.server_config()?);
        let (identity_file, identity_file_identity) = paths.create_named_identity(
            STAGED_PUBLICATION_IDENTITY_NAME,
            &publication_bytes,
            uid,
        )?;
        let listener = std::os::unix::net::UnixListener::bind(&paths.staged_socket)
            .map_err(|_| ControlError::InvalidPublication("staged control socket bind failed"))?;
        std::fs::set_permissions(&paths.staged_socket, std::fs::Permissions::from_mode(0o600))?;
        let socket_metadata = std::fs::symlink_metadata(&paths.staged_socket)?;
        let staged_socket_identity = FileIdentity::from_metadata(&socket_metadata);
        if !socket_metadata.file_type().is_socket()
            || socket_metadata.uid() != uid
            || socket_metadata.mode() & 0o7777 != 0o600
        {
            return Err(ControlError::InvalidPublication(
                "staged control socket ownership or mode is invalid",
            ));
        }
        paths.root_file.sync_all()?;
        Ok(Self {
            listener: Some(listener),
            paths,
            staged_socket_identity,
            identity_file,
            identity_file_identity,
            publication_identity,
            publication_bytes,
            uid,
            anchor: Some(anchor),
            server_config,
            registry,
        })
    }

    /// Exposes the staged face after the exact generation is durably Published.
    ///
    /// # Errors
    /// Refuses stale durable state, changed staged artifacts, or exposure failure.
    pub fn expose(mut self) -> Result<PublishedControlFace, ControlError> {
        self.registry.require_serviceable_generation()?;
        self.paths.validate_root(self.uid)?;
        let staged_socket = std::fs::symlink_metadata(&self.paths.staged_socket)?;
        if FileIdentity::from_metadata(&staged_socket) != self.staged_socket_identity {
            return Err(ControlError::InvalidPublication(
                "staged control socket changed",
            ));
        }
        self.paths.validate_identity_at(
            &self.paths.staged_identity,
            &self.identity_file,
            self.identity_file_identity,
            self.uid,
            self.publication_bytes.len(),
        )?;
        std::fs::rename(&self.paths.staged_socket, &self.paths.socket)?;
        if let Err(error) = std::fs::rename(&self.paths.staged_identity, &self.paths.identity) {
            remove_if_identity_matches(&self.paths.socket, self.staged_socket_identity);
            return Err(error.into());
        }
        self.paths.root_file.sync_all()?;
        let socket_identity =
            FileIdentity::from_metadata(&std::fs::symlink_metadata(&self.paths.socket)?);
        self.paths.validate_identity(
            &self.identity_file,
            self.identity_file_identity,
            self.uid,
            self.publication_bytes.len(),
        )?;
        let paths = ControlPaths::open(&self.paths.root, self.uid)?;
        Ok(PublishedControlFace {
            listener: self
                .listener
                .take()
                .ok_or(ControlError::InvalidPublication(
                    "staged listener is absent",
                ))?,
            paths,
            socket_identity,
            identity_file: self.identity_file.try_clone()?,
            identity_file_identity: self.identity_file_identity,
            publication_identity: self.publication_identity.clone(),
            publication_bytes: self.publication_bytes.clone(),
            uid: self.uid,
            anchor: self.anchor.take().ok_or(ControlError::InvalidPublication(
                "staged Trust Anchor is absent",
            ))?,
            server_config: Arc::clone(&self.server_config),
            registry: Arc::clone(&self.registry),
        })
    }
}

impl Drop for StagedControlFace {
    fn drop(&mut self) {
        remove_if_identity_matches(&self.paths.staged_socket, self.staged_socket_identity);
        remove_if_identity_matches(&self.paths.staged_identity, self.identity_file_identity);
    }
}

impl PublishedControlFace {
    /// Publishes one generation control face in the fixed host-private runtime path.
    ///
    /// # Errors
    /// Refuses stale generations, existing publication objects, path/ownership/mode errors, and
    /// TLS configuration failures.
    pub fn publish(
        runtime_domain_root: &Path,
        uid: u32,
        anchor: TrustAnchor,
        store: Arc<HostCoordinatorDomainStore>,
    ) -> Result<Self, ControlError> {
        let registry = Arc::new(SessionRegistry::new(
            store,
            anchor.generation_identity.clone(),
        ));
        StagedControlFace::stage(runtime_domain_root, uid, anchor, registry)?.expose()
    }

    #[must_use]
    pub fn publication_identity(&self) -> &ProtocolPublicationIdentity {
        &self.publication_identity
    }

    #[must_use]
    pub fn registry(&self) -> Arc<SessionRegistry> {
        Arc::clone(&self.registry)
    }

    #[must_use]
    pub fn socket_path(&self) -> &Path {
        &self.paths.socket
    }

    #[must_use]
    pub fn identity_path(&self) -> &Path {
        &self.paths.identity
    }

    /// Withdraws exact discovery and connection paths while retaining live accepted streams.
    ///
    /// # Errors
    /// Refuses changed artifacts, removal failures, or artifacts that remain after removal.
    pub fn withdraw(&self) -> Result<(), ControlError> {
        remove_exact_if_identity_matches(&self.paths.socket, self.socket_identity)?;
        remove_exact_if_identity_matches(&self.paths.identity, self.identity_file_identity)
    }

    /// Accepts and serves one TLS connection. Registered connections remain live until EOF.
    ///
    /// # Errors
    /// Refuses peer UID, publication, TLS, ALPN, certificate, framing, tuple, duplicate, and
    /// persistence failures.
    pub fn serve_next(&self) -> Result<(), ControlError> {
        self.serve_next_coordinated(
            |admission| {
                self.registry
                    .register(admission.request, &admission.credential)
                    .map(|_| ())
            },
            |session_id| self.registry.disconnect(session_id),
        )
    }

    /// Serves one connection while routing authenticated session state through its lifecycle.
    ///
    /// # Errors
    /// Refuses the same conditions as [`Self::serve_next`] plus lifecycle admission or disconnect
    /// failure.
    pub fn serve_next_coordinated<Admit, Disconnect>(
        &self,
        mut admit: Admit,
        mut disconnect: Disconnect,
    ) -> Result<(), ControlError>
    where
        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
    {
        let (socket, _) = self.listener.accept()?;
        let peer = socket_peercred(&socket).map_err(io::Error::from)?;
        require_peer_uid(peer.uid.as_raw(), self.uid)?;
        self.validate_publication()?;
        let mut connection = ServerConnection::new(Arc::clone(&self.server_config))?;
        let mut socket = socket;
        while connection.is_handshaking() {
            connection.complete_io(&mut socket)?;
        }
        if connection.alpn_protocol() != Some(CONTROL_ALPN) {
            return Err(ControlError::InvalidRequest);
        }
        let peer_certificate = connection.peer_certificates().map(|certificates| {
            certificates
                .iter()
                .map(|certificate| certificate.to_vec())
                .collect::<Vec<_>>()
        });
        let mut stream = rustls::StreamOwned::new(connection, socket);
        match peer_certificate {
            None => self.serve_enrollment(&mut stream),
            Some(certificates) => {
                self.serve_registration(&mut stream, &certificates, &mut admit, &mut disconnect)
            }
        }
    }

    fn serve_enrollment(
        &self,
        stream: &mut rustls::StreamOwned<ServerConnection, std::os::unix::net::UnixStream>,
    ) -> Result<(), ControlError> {
        let request: ControlRequest = read_frame(stream, MAX_ENROLLMENT_BYTES)?;
        let ControlRequest::Enroll(request) = request else {
            return Err(ControlError::InvalidRequest);
        };
        let credential = self.anchor.issue_client_credential(&request)?;
        self.registry.enroll(credential.clone())?;
        write_frame(
            stream,
            &ControlResponse::Enrolled(ClientEnrollmentResponse {
                protocol_version: ProtocolVersion::V1,
                credential,
            }),
        )
    }

    fn serve_registration<Admit, Disconnect>(
        &self,
        stream: &mut rustls::StreamOwned<ServerConnection, std::os::unix::net::UnixStream>,
        certificates: &[Vec<u8>],
        admit: &mut Admit,
        disconnect: &mut Disconnect,
    ) -> Result<(), ControlError>
    where
        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
    {
        let [leaf] = certificates else {
            return Err(ControlError::InvalidCertificateProfile);
        };
        let credential = self.registry.credential_for_leaf(leaf)?;
        let request: ControlRequest = read_frame(stream, MAX_CONTROL_BYTES)?;
        let ControlRequest::Register(request) = request else {
            return Err(ControlError::InvalidRequest);
        };
        let capability = issue_session_capability(
            request.session_binding.clone(),
            self.anchor.coordinator_protocol_credential.credential_id,
            self.anchor.sandbox_protocol_credential.credential_id,
            &self.anchor.coordinator_protocol_key,
        )?;
        let session_id = request.session_binding.session_id;
        admit(ControlSessionAdmission {
            request,
            credential,
        })?;
        let result = (|| {
            write_frame(
                stream,
                &ControlResponse::Registered(SessionRegistrationResponse {
                    protocol_version: ProtocolVersion::V1,
                    session_capability: capability,
                }),
            )?;
            let mut extra = [0_u8; 1];
            match stream.read(&mut extra) {
                Ok(0) => Ok(()),
                Ok(_) => Err(ControlError::InvalidRequest),
                Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => Ok(()),
                Err(error) => Err(error.into()),
            }
        })();
        let disconnected = disconnect(session_id);
        result.and(disconnected)
    }

    fn validate_publication(&self) -> Result<(), ControlError> {
        self.paths.validate_root(self.uid)?;
        let socket = std::fs::symlink_metadata(&self.paths.socket)?;
        let identity = std::fs::symlink_metadata(&self.paths.identity)?;
        if FileIdentity::from_metadata(&socket) != self.socket_identity
            || !socket.file_type().is_socket()
            || socket.uid() != self.uid
            || socket.mode() & 0o7777 != 0o600
            || FileIdentity::from_metadata(&identity) != self.identity_file_identity
            || !identity.file_type().is_file()
            || identity.uid() != self.uid
            || identity.mode() & 0o7777 != 0o600
            || usize::try_from(self.identity_file.metadata()?.len()).ok()
                != Some(self.publication_bytes.len())
        {
            return Err(ControlError::InvalidPublication(
                "publication objects changed identity, ownership, or mode",
            ));
        }
        let mut exact_identity = self.identity_file.try_clone()?;
        exact_identity.seek(SeekFrom::Start(0))?;
        let mut bytes = Vec::new();
        exact_identity.read_to_end(&mut bytes)?;
        if bytes != self.publication_bytes {
            return Err(ControlError::InvalidPublication(
                "publication identity bytes changed",
            ));
        }
        self.registry.require_serviceable_generation()
    }

    /// Creates a TLS client configuration pinned to this publication identity.
    ///
    /// # Errors
    /// Refuses malformed CA or client certificate/key material.
    pub fn client_config(
        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)
    }

    /// Connects a rustls stream to the fixed control socket using an existing configuration.
    ///
    /// # Errors
    /// Returns path, connection, or TLS initialization failures.
    pub fn connect(
        socket_path: &Path,
        config: ClientConfig,
        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))
    }

    /// Writes one length-delimited control record.
    ///
    /// # Errors
    /// Returns encoding or TLS I/O failures.
    pub fn send<Request: serde::Serialize>(
        stream: &mut impl Write,
        request: &Request,
    ) -> Result<(), ControlError> {
        write_frame(stream, request)
    }

    /// Reads one length-delimited control record.
    ///
    /// # Errors
    /// Returns framing, size, decoding, or TLS I/O failures.
    pub fn receive<Response: serde::de::DeserializeOwned + serde::Serialize>(
        stream: &mut impl Read,
    ) -> Result<Response, ControlError> {
        read_frame(stream, MAX_CONTROL_BYTES)
    }
}

impl Drop for PublishedControlFace {
    fn drop(&mut self) {
        remove_if_identity_matches(&self.paths.socket, self.socket_identity);
        remove_if_identity_matches(&self.paths.identity, self.identity_file_identity);
    }
}

fn remove_exact_if_identity_matches(
    path: &Path,
    expected: FileIdentity,
) -> Result<(), ControlError> {
    let metadata = match std::fs::symlink_metadata(path) {
        Ok(metadata) => metadata,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
        Err(error) => return Err(error.into()),
    };
    if FileIdentity::from_metadata(&metadata) != expected {
        return Err(ControlError::InvalidPublication(
            "publication artifact identity changed before withdrawal",
        ));
    }
    std::fs::remove_file(path)?;
    match std::fs::symlink_metadata(path) {
        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error.into()),
        Ok(_) => Err(ControlError::InvalidPublication(
            "publication artifact remained after withdrawal",
        )),
    }
}

fn remove_if_identity_matches(path: &Path, expected: FileIdentity) {
    if std::fs::symlink_metadata(path)
        .ok()
        .is_some_and(|metadata| FileIdentity::from_metadata(&metadata) == expected)
    {
        let _ = std::fs::remove_file(path);
    }
}

fn leaf_parameters(
    common_name: &str,
    usage: ExtendedKeyUsagePurpose,
    now: OffsetDateTime,
) -> Result<CertificateParams, ControlError> {
    let mut params = CertificateParams::new(vec![common_name.to_owned()])?;
    params.distinguished_name = DistinguishedName::new();
    params
        .distinguished_name
        .push(DnType::CommonName, common_name);
    params.not_before = now - Duration::minutes(1);
    params.not_after = now + Duration::hours(1);
    params.is_ca = IsCa::NoCa;
    params.key_usages = vec![KeyUsagePurpose::DigitalSignature];
    params.extended_key_usages = vec![usage];
    Ok(params)
}

fn random_signing_key() -> Result<SigningKey, ControlError> {
    let mut bytes = [0_u8; 32];
    getrandom::fill(&mut bytes).map_err(|_| ControlError::InvalidCertificateProfile)?;
    Ok(SigningKey::from_bytes(&bytes))
}

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))))
}

fn certificate_digest(certificate: impl AsRef<[u8]>) -> CertificateDigest {
    CertificateDigest::new(sha256(certificate.as_ref()))
}

fn sha256(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

fn short_identity(bytes: &[u8; 32]) -> String {
    hex(&bytes[..8])
}

fn hex(bytes: &[u8]) -> String {
    let mut encoded = String::with_capacity(bytes.len() * 2);
    for byte in bytes {
        use fmt::Write as _;
        let _ = write!(encoded, "{byte:02x}");
    }
    encoded
}

fn require_peer_uid(actual: u32, expected: u32) -> Result<(), ControlError> {
    if actual == expected {
        Ok(())
    } else {
        Err(ControlError::UnauthorizedPeer)
    }
}

fn is_live_session(phase: SessionPhase) -> bool {
    matches!(
        phase,
        SessionPhase::Registering
            | SessionPhase::Attached
            | SessionPhase::Prompting
            | SessionPhase::HostExecuting
    )
}

fn read_frame<Record: serde::de::DeserializeOwned + serde::Serialize>(
    reader: &mut impl Read,
    maximum: usize,
) -> Result<Record, ControlError> {
    let mut length = [0_u8; 4];
    reader.read_exact(&mut length)?;
    let length = u32::from_be_bytes(length) as usize;
    if length > maximum {
        return Err(ControlError::OversizeRequest { length, maximum });
    }
    let mut bytes = vec![0_u8; length];
    reader.read_exact(&mut bytes)?;
    decode(&bytes).map_err(Into::into)
}

fn write_frame<Record: serde::Serialize>(
    writer: &mut impl Write,
    record: &Record,
) -> Result<(), ControlError> {
    let bytes = encode(record)?;
    let length = u32::try_from(bytes.len())
        .map_err(|_| ControlError::OversizeRequest {
            length: bytes.len(),
            maximum: u32::MAX as usize,
        })?
        .to_be_bytes();
    writer.write_all(&length)?;
    writer.write_all(&bytes)?;
    writer.flush()?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::thread;

    use lazaret_core::identity::{
        ConfigurationFingerprint, CoordinatorId, DomainId, GenerationId, SandboxId,
    };
    use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName};

    use super::*;

    fn generation() -> GenerationIdentity {
        GenerationIdentity {
            agent_profile: "trusted".into(),
            domain_id: DomainId::new([1; 32]),
            generation_id: GenerationId::new(7),
            configuration_fingerprint: ConfigurationFingerprint::new([2; 32]),
            coordinator_id: CoordinatorId::new([3; 32]),
            sandbox_id: SandboxId::new([4; 32]),
        }
    }

    fn client_config_with_certificate(
        anchor: &TrustAnchor,
        certificate: CertificateDer<'static>,
        private_key: PrivatePkcs8KeyDer<'static>,
    ) -> ClientConfig {
        let mut roots = RootCertStore::empty();
        roots.add(anchor.tls_ca_der.clone()).unwrap();
        let mut config = ClientConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
            .with_root_certificates(roots)
            .with_client_auth_cert(vec![certificate], PrivateKeyDer::Pkcs8(private_key))
            .unwrap();
        config.alpn_protocols = vec![CONTROL_ALPN.to_vec()];
        config.resumption = rustls::client::Resumption::disabled();
        config
    }

    fn handshake_pair(
        server_config: ServerConfig,
        client_config: ClientConfig,
    ) -> (Result<(), rustls::Error>, Result<(), rustls::Error>) {
        let (mut server_socket, mut client_socket) =
            std::os::unix::net::UnixStream::pair().unwrap();
        let server = thread::spawn(move || {
            let mut connection = ServerConnection::new(Arc::new(server_config)).unwrap();
            while connection.is_handshaking() {
                connection
                    .complete_io(&mut server_socket)
                    .map_err(|error| rustls::Error::General(error.to_string()))?;
            }
            Ok(())
        });
        let mut connection = ClientConnection::new(
            Arc::new(client_config),
            ServerName::try_from(CONTROL_SERVER_NAME).unwrap(),
        )
        .unwrap();
        let client = (|| {
            while connection.is_handshaking() {
                connection
                    .complete_io(&mut client_socket)
                    .map_err(|error| rustls::Error::General(error.to_string()))?;
            }
            Ok(())
        })();
        (server.join().unwrap(), client)
    }

    #[test]
    fn exact_peer_uid_is_required() {
        assert!(require_peer_uid(1_000, 1_000).is_ok());
        assert!(matches!(
            require_peer_uid(1_001, 1_000),
            Err(ControlError::UnauthorizedPeer)
        ));
    }

    #[test]
    fn rustls_refuses_client_leaf_with_server_role() {
        let anchor = TrustAnchor::create(generation()).unwrap();
        let key = KeyPair::generate_for(&PKCS_ED25519).unwrap();
        let certificate = leaf_parameters(
            "wrong-client-role.invalid",
            ExtendedKeyUsagePurpose::ServerAuth,
            OffsetDateTime::now_utc(),
        )
        .unwrap()
        .signed_by(&key, &anchor.tls_ca)
        .unwrap();
        let client_config = client_config_with_certificate(
            &anchor,
            certificate.der().clone(),
            PrivatePkcs8KeyDer::from(key.serialize_der()),
        );
        let (server, client) = handshake_pair(anchor.server_config().unwrap(), client_config);
        assert!(server.is_err() || client.is_err());
    }

    #[test]
    fn rustls_refuses_server_leaf_with_client_role() {
        let anchor = TrustAnchor::create(generation()).unwrap();
        let key = KeyPair::generate_for(&PKCS_ED25519).unwrap();
        let certificate = leaf_parameters(
            CONTROL_SERVER_NAME,
            ExtendedKeyUsagePurpose::ClientAuth,
            OffsetDateTime::now_utc(),
        )
        .unwrap()
        .signed_by(&key, &anchor.tls_ca)
        .unwrap();
        let server_config =
            ServerConfig::builder_with_protocol_versions(&[&rustls::version::TLS13])
                .with_no_client_auth()
                .with_single_cert(
                    vec![certificate.der().clone()],
                    PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key.serialize_der())),
                )
                .unwrap();
        let client_config =
            PublishedControlFace::client_config(&anchor.publication_identity(), None).unwrap();
        let (server, client) = handshake_pair(server_config, client_config);
        assert!(server.is_err() || client.is_err());
    }
}