Decompose oversized Rust files by responsibility

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

@@ -0,0 +1,772 @@
+impl PublishedControlFace {
+    pub(crate) fn recovery_ownership(&self) -> Result<PublicationOwnership, ControlError> {
+        publication_recovery_ownership(&self.paths)
+    }
+
+    /// Selects blocking or polling admission for the publication listener.
+    ///
+    /// # Errors
+    /// Returns an operating-system failure changing the listener mode.
+    pub fn set_nonblocking(&self, nonblocking: bool) -> Result<(), ControlError> {
+        self.listener.set_nonblocking(nonblocking)?;
+        Ok(())
+    }
+
+    /// Discovers one publication through the retained runtime Domain authority.
+    ///
+    /// # Errors
+    /// Refuses changed root authority or absent, replaced, non-private, oversized, or malformed
+    /// publication objects.
+    pub fn discover_confined(
+        runtime_domain_root: &ConfinedPathPlan,
+        uid: u32,
+    ) -> Result<(File, PathBuf, ProtocolPublicationIdentity), ControlError> {
+        let root = File::from(
+            runtime_domain_root
+                .open_authority()
+                .map_err(|_| ControlError::InvalidPath("runtime Domain authority changed"))?,
+        );
+        let root_metadata = root.metadata()?;
+        if !root_metadata.is_dir()
+            || root_metadata.uid() != uid
+            || root_metadata.mode() & 0o7777 != 0o700
+        {
+            return Err(ControlError::InvalidPath(
+                "runtime Domain root is not invoking-user-owned mode 0700",
+            ));
+        }
+        let descriptor = openat(
+            root.as_fd(),
+            PUBLICATION_IDENTITY_NAME,
+            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
+            Mode::empty(),
+        )
+        .map_err(io::Error::from)?;
+        let mut identity = File::from(descriptor);
+        let identity_metadata = identity.metadata()?;
+        if !identity_metadata.file_type().is_file()
+            || identity_metadata.uid() != uid
+            || identity_metadata.mode() & 0o7777 != 0o600
+            || identity_metadata.nlink() != 1
+            || identity_metadata.len() > MAX_CONTROL_BYTES as u64
+        {
+            return Err(ControlError::InvalidPublication(
+                "publication identity is not an exact private regular file",
+            ));
+        }
+        let socket = File::from(
+            openat(
+                root.as_fd(),
+                CONTROL_SOCKET_NAME,
+                OFlags::PATH | OFlags::CLOEXEC | OFlags::NOFOLLOW,
+                Mode::empty(),
+            )
+            .map_err(io::Error::from)?,
+        );
+        let socket_metadata = socket.metadata()?;
+        if !socket_metadata.file_type().is_socket()
+            || socket_metadata.uid() != uid
+            || socket_metadata.mode() & 0o7777 != 0o600
+        {
+            return Err(ControlError::InvalidPublication(
+                "control socket is not an invoking-user-private socket",
+            ));
+        }
+        let mut bytes = Vec::new();
+        identity.read_to_end(&mut bytes)?;
+        let socket_path = PathBuf::from(format!(
+            "/proc/self/fd/{}/{}",
+            root.as_raw_fd(),
+            CONTROL_SOCKET_NAME
+        ));
+        Ok((root, socket_path, decode(&bytes)?))
+    }
+
+    /// Discovers one exact current control publication beneath a validated runtime Domain root.
+    ///
+    /// # Errors
+    /// Refuses absent, replaced, non-private, oversized, or malformed publication objects.
+    pub fn discover(
+        runtime_domain_root: &Path,
+        uid: u32,
+    ) -> Result<(PathBuf, ProtocolPublicationIdentity), ControlError> {
+        let paths = ControlPaths::open(runtime_domain_root, uid)?;
+        let descriptor = openat(
+            paths.root_file.as_fd(),
+            PUBLICATION_IDENTITY_NAME,
+            OFlags::RDONLY | OFlags::CLOEXEC | OFlags::NOFOLLOW,
+            Mode::empty(),
+        )
+        .map_err(io::Error::from)?;
+        let mut identity = File::from(descriptor);
+        let descriptor_metadata = identity.metadata()?;
+        let path_metadata = std::fs::symlink_metadata(&paths.identity)?;
+        let expected = FileIdentity::from_metadata(&descriptor_metadata);
+        if FileIdentity::from_metadata(&path_metadata) != expected
+            || !descriptor_metadata.file_type().is_file()
+            || descriptor_metadata.uid() != uid
+            || descriptor_metadata.mode() & 0o7777 != 0o600
+            || descriptor_metadata.nlink() != 1
+            || descriptor_metadata.len() > MAX_CONTROL_BYTES as u64
+        {
+            return Err(ControlError::InvalidPublication(
+                "publication identity is not an exact private regular file",
+            ));
+        }
+        let socket = std::fs::symlink_metadata(&paths.socket)?;
+        if !socket.file_type().is_socket() || socket.uid() != uid || socket.mode() & 0o7777 != 0o600
+        {
+            return Err(ControlError::InvalidPublication(
+                "control socket is not an invoking-user-private socket",
+            ));
+        }
+        let mut bytes = Vec::new();
+        identity.read_to_end(&mut bytes)?;
+        Ok((paths.socket, decode(&bytes)?))
+    }
+
+    /// 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> {
+        Self::publish_with_executor(
+            runtime_domain_root,
+            uid,
+            anchor,
+            store,
+            Arc::new(ExecutionUnavailable),
+        )
+    }
+
+    /// Publishes one generation control face with non-interactive session execution.
+    ///
+    /// # Errors
+    /// Refuses the same invalid publication conditions as [`Self::publish`].
+    pub fn publish_with_executor(
+        runtime_domain_root: &Path,
+        uid: u32,
+        anchor: TrustAnchor,
+        store: Arc<HostCoordinatorDomainStore>,
+        executor: Arc<dyn NonInteractiveSessionExecutor>,
+    ) -> Result<Self, ControlError> {
+        let registry = Arc::new(SessionRegistry::new(
+            store,
+            anchor.generation_identity.clone(),
+        ));
+        StagedControlFace::stage_with_executor(
+            runtime_domain_root,
+            uid,
+            anchor,
+            registry,
+            executor,
+        )?
+        .expose()
+    }
+
+    /// Publishes one generation control face with command and terminal execution.
+    ///
+    /// # Errors
+    /// Refuses the same invalid publication conditions as [`Self::publish`].
+    pub fn publish_with_executors(
+        runtime_domain_root: &Path,
+        uid: u32,
+        anchor: TrustAnchor,
+        store: Arc<HostCoordinatorDomainStore>,
+        executor: Arc<dyn NonInteractiveSessionExecutor>,
+        interactive_executor: Arc<dyn InteractiveSessionExecutor>,
+    ) -> Result<Self, ControlError> {
+        let registry = Arc::new(SessionRegistry::new(
+            store,
+            anchor.generation_identity.clone(),
+        ));
+        StagedControlFace::stage_with_executors(
+            runtime_domain_root,
+            uid,
+            anchor,
+            registry,
+            executor,
+            interactive_executor,
+        )?
+        .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,
+        admit: Admit,
+        disconnect: Disconnect,
+    ) -> Result<(), ControlError>
+    where
+        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
+        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
+    {
+        self.serve_next_coordinated_with_registration(admit, |_, _| Ok(()), disconnect)
+    }
+
+    pub(crate) fn serve_next_coordinated_with_registration<Admit, Registered, Disconnect>(
+        &self,
+        admit: Admit,
+        registered: Registered,
+        disconnect: Disconnect,
+    ) -> Result<(), ControlError>
+    where
+        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
+        Registered: FnMut(SessionBinding, SessionRegistrationResponse) -> Result<(), ControlError>,
+        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
+    {
+        let connection = self.accept_connection()?;
+        self.serve_connection_coordinated(connection, admit, registered, disconnect)
+    }
+
+    pub(crate) fn accept_connection(&self) -> Result<AcceptedControlConnection, ControlError> {
+        let (socket, _) = self.listener.accept()?;
+        Ok(AcceptedControlConnection { socket })
+    }
+
+    pub(crate) fn serve_connection_coordinated<Admit, Registered, Disconnect>(
+        &self,
+        connection: AcceptedControlConnection,
+        mut admit: Admit,
+        mut registered: Registered,
+        mut disconnect: Disconnect,
+    ) -> Result<(), ControlError>
+    where
+        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
+        Registered: FnMut(SessionBinding, SessionRegistrationResponse) -> Result<(), ControlError>,
+        Disconnect: FnMut(SessionId) -> Result<(), ControlError>,
+    {
+        let socket = connection.socket;
+        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 registered,
+                &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, Registered, Disconnect>(
+        &self,
+        stream: &mut rustls::StreamOwned<ServerConnection, std::os::unix::net::UnixStream>,
+        certificates: &[Vec<u8>],
+        admit: &mut Admit,
+        registered: &mut Registered,
+        disconnect: &mut Disconnect,
+    ) -> Result<(), ControlError>
+    where
+        Admit: FnMut(ControlSessionAdmission) -> Result<(), ControlError>,
+        Registered: FnMut(SessionBinding, SessionRegistrationResponse) -> 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;
+        let binding = request.session_binding.clone();
+        admit(ControlSessionAdmission {
+            request,
+            credential,
+        })?;
+        let result = (|| {
+            let registration = SessionRegistrationResponse {
+                protocol_version: ProtocolVersion::V1,
+                session_capability: capability,
+            };
+            registered(binding, registration.clone())?;
+            write_frame(stream, &ControlResponse::Registered(registration))?;
+            let request: ControlRequest = match read_frame(stream, MAX_CONTROL_BYTES) {
+                Ok(request) => request,
+                Err(ControlError::Io(error))
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::UnexpectedEof | io::ErrorKind::ConnectionReset
+                    ) =>
+                {
+                    return Ok(());
+                }
+                Err(error) => return Err(error),
+            };
+            match request {
+                ControlRequest::Execute(request) => {
+                    validate_command(request.session_id, session_id, &request.argv)?;
+                    let mut send_message = |message: NonInteractiveSessionMessage| {
+                        if matches!(message, NonInteractiveSessionMessage::Exit(_)) {
+                            return Err(ControlError::InvalidRequest);
+                        }
+                        require_non_interactive_session(&message, request.session_id)?;
+                        write_frame(stream, &message)
+                    };
+                    let outcome = self.executor.execute(&request, &mut send_message)?;
+                    require_session_id(outcome.session_id(), request.session_id)?;
+                    write_frame(stream, &NonInteractiveSessionMessage::Exit(outcome))
+                }
+                ControlRequest::ExecuteInteractive(request) => {
+                    validate_command(request.session_id, session_id, &request.argv)?;
+                    if request.initial_size.session_id != session_id {
+                        return Err(ControlError::InvalidRequest);
+                    }
+                    let mut terminal = self.interactive_executor.start(&request)?;
+                    let mut event = InteractiveSessionEvent::Poll { session_id };
+                    loop {
+                        let batch = terminal.exchange(event)?;
+                        require_interactive_batch(&batch, session_id)?;
+                        let complete = batch.messages.iter().any(|message| {
+                            matches!(
+                                message,
+                                InteractiveSessionMessage::Terminal(
+                                    lazaret_core::records::SessionTerminalMessage::Exit(_)
+                                )
+                            )
+                        });
+                        write_frame(stream, &batch)?;
+                        if complete {
+                            break Ok(());
+                        }
+                        event = read_frame(stream, MAX_CONTROL_BYTES)?;
+                        require_interactive_event(&event, session_id)?;
+                    }
+                }
+                ControlRequest::Enroll(_) | ControlRequest::Register(_) => {
+                    Err(ControlError::InvalidRequest)
+                }
+            }
+        })();
+        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 bytes = vec![0; self.publication_bytes.len()];
+        self.identity_file.read_exact_at(&mut bytes, 0)?;
+        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 require_non_interactive_session(
+    message: &NonInteractiveSessionMessage,
+    expected: SessionId,
+) -> Result<(), ControlError> {
+    let actual = match message {
+        NonInteractiveSessionMessage::Stdout(output)
+        | NonInteractiveSessionMessage::Stderr(output) => output.session_id,
+        NonInteractiveSessionMessage::Exit(exit) => exit.session_id(),
+    };
+    if actual == expected {
+        Ok(())
+    } else {
+        Err(ControlError::InvalidRequest)
+    }
+}
+
+fn validate_command(
+    actual: SessionId,
+    expected: SessionId,
+    argv: &[Vec<u8>],
+) -> Result<(), ControlError> {
+    require_session_id(actual, expected)?;
+    if argv.is_empty() || argv[0].is_empty() || argv.iter().any(|argument| argument.contains(&0)) {
+        return Err(ControlError::InvalidRequest);
+    }
+    Ok(())
+}
+
+fn require_session_id(actual: SessionId, expected: SessionId) -> Result<(), ControlError> {
+    if actual == expected {
+        Ok(())
+    } else {
+        Err(ControlError::InvalidRequest)
+    }
+}
+
+fn require_interactive_event(
+    event: &InteractiveSessionEvent,
+    expected: SessionId,
+) -> Result<(), ControlError> {
+    let actual = match event {
+        InteractiveSessionEvent::Terminal(event) => match event {
+            lazaret_core::records::SessionTerminalEvent::Input(event) => event.session_id,
+            lazaret_core::records::SessionTerminalEvent::Resize(event) => event.session_id,
+            lazaret_core::records::SessionTerminalEvent::Signal(event) => event.session_id,
+            lazaret_core::records::SessionTerminalEvent::Close(event) => event.session_id,
+        },
+        InteractiveSessionEvent::PromptDecision { session_id, .. }
+        | InteractiveSessionEvent::Poll { session_id } => *session_id,
+    };
+    require_session_id(actual, expected)
+}
+
+fn require_interactive_batch(
+    batch: &InteractiveSessionBatch,
+    expected: SessionId,
+) -> Result<(), ControlError> {
+    require_session_id(batch.session_id, expected)?;
+    for message in &batch.messages {
+        let actual = match message {
+            InteractiveSessionMessage::Terminal(message) => match message {
+                lazaret_core::records::SessionTerminalMessage::Output(output) => output.session_id,
+                lazaret_core::records::SessionTerminalMessage::Exit(exit) => exit.session_id(),
+            },
+            InteractiveSessionMessage::Prompt(prompt) => prompt.session_id,
+        };
+        require_session_id(actual, expected)?;
+    }
+    Ok(())
+}
+
+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(())
+}