liminal-server 0.25.4

//! The WebSocket connection process's inbound path: each binary message as one
//! canonical frame, the pre-admission bound, and read errors.

use super::{SliceStep, WebSocketConnectionProcess};
use crate::ServerError;
use crate::server::connection::apply::apply_frame;
use crate::server::connection::pre_admission::PreAdmissionOversize;
use crate::server::connection::services::server_error_from_protocol;
use crate::server::connection::state::{FrameAction, ProcessStatus};
use crate::server::connection::supervisor::runtime::ConnectionControl;
use crate::server::connection::websocket::{WsInboundViolation, decode_ws_binary};
use crate::server::participant::ConnectionFateClass;
use beamr::native::native_process::NativeOutcome;
use beamr::process::ExitReason;
use tungstenite::Message;
use tungstenite::protocol::frame::CloseFrame;
use tungstenite::protocol::frame::coding::CloseCode;

impl WebSocketConnectionProcess {
    /// Refuses an unadmitted peer's oversize frame or message by name: the
    /// RFC 6455 "message too big" close code carrying the refusal, then the
    /// same cleanup as a terminal participant rejection. Nothing past the
    /// bound was buffered, and nothing of the message reaches `apply_frame`.
    pub(super) fn refuse_pre_admission(
        &mut self,
        pid: u64,
        refusal: PreAdmissionOversize,
    ) -> SliceStep {
        let violation = WsInboundViolation::PreAdmissionOversize(refusal);
        tracing::warn!(connection_pid = pid, peer_addr = ?self.peer_addr, %violation, "websocket contract violation");
        if let Some(socket) = self.socket.as_mut() {
            if let Err(error) = socket.close(Some(pre_admission_close_frame(refusal))) {
                tracing::debug!(connection_pid = pid, %error, "websocket close frame not sent");
            }
            if let Err(error) = socket.flush() {
                tracing::debug!(connection_pid = pid, %error, "websocket close flush incomplete");
            }
        }
        self.release_conversations();
        self.runtime.finish(pid);
        SliceStep::Stop(ExitReason::Normal)
    }

    /// Replaces the pre-admission bound with the acceptor's message bound the
    /// first time the connection is seen admitted. One flag test per message
    /// afterwards; no allocation.
    pub(super) fn install_admitted_bound_once(&mut self) {
        if self.admitted_bound_installed || !self.state.authenticated {
            return;
        }
        if let Some(socket) = self.socket.as_mut() {
            super::super::install_inbound_bound(socket, self.admitted_bound);
        }
        self.admitted_bound_installed = true;
    }

    /// A dropped peer still has an abnormal exit and the same fate/cleanup
    /// discipline, but is not evidence that our connection process crashed.
    pub(super) fn finish_read_error(&mut self, pid: u64, error: &tungstenite::Error) -> SliceStep {
        let peer_reset = match error {
            tungstenite::Error::Protocol(
                tungstenite::error::ProtocolError::ResetWithoutClosingHandshake,
            ) => true,
            tungstenite::Error::Io(error) => matches!(
                error.kind(),
                std::io::ErrorKind::ConnectionReset
                    | std::io::ErrorKind::ConnectionAborted
                    | std::io::ErrorKind::BrokenPipe
                    | std::io::ErrorKind::UnexpectedEof
            ),
            _ => false,
        };
        let description = if peer_reset {
            "websocket peer reset"
        } else {
            "websocket connection read failed"
        };
        tracing::warn!(connection_pid = pid, peer_addr = ?self.peer_addr, %error, "{description}");
        if let Err(fate_error) = self.complete_connection_fate(ConnectionFateClass::ConnectionLost)
        {
            return self.fail_fate(pid, &fate_error);
        }
        self.close_transport_abnormal();
        self.release_conversations();
        if peer_reset {
            self.runtime.finish(pid);
        } else {
            self.runtime
                .mark_crashed(pid, ExitReason::Error, self.peer_addr);
        }
        SliceStep::Stop(ExitReason::Error)
    }

    pub(super) fn handle_inbound_message(
        &mut self,
        pid: u64,
        message: Message,
    ) -> Option<SliceStep> {
        match message {
            Message::Binary(bytes) => match self.apply_binary(pid, &bytes) {
                Ok(ProcessStatus::Continue) => {
                    self.install_admitted_bound_once();
                    None
                }
                Ok(ProcessStatus::Close) => Some(self.finish_normal_close(pid)),
                Ok(ProcessStatus::CloseWithFate(ConnectionFateClass::ServerShutdown)) => {
                    self.runtime.take_force_close(pid);
                    Some(
                        match self.handle_control(pid, ConnectionControl::ForceClose) {
                            Some(NativeOutcome::Stop(reason)) => SliceStep::Stop(reason),
                            _ => SliceStep::Continue,
                        },
                    )
                }
                Ok(ProcessStatus::CloseWithFate(class)) => Some(self.finish_fate_close(pid, class)),
                Err(error) => {
                    tracing::warn!(connection_pid = pid, %error, "connection process failed");
                    self.close_transport_abnormal();
                    self.release_conversations();
                    self.runtime
                        .mark_crashed(pid, ExitReason::Error, self.peer_addr);
                    Some(SliceStep::Stop(ExitReason::Error))
                }
            },
            Message::Text(_) => {
                let violation = WsInboundViolation::TextMessage;
                tracing::warn!(connection_pid = pid, %violation, "websocket contract violation");
                let bound = match self.bound_protocol_refusal() {
                    Ok(bound) => bound,
                    Err(error) => return Some(self.fail_fate(pid, &error)),
                };
                if bound {
                    return Some(self.finish_fate_close(pid, ConnectionFateClass::ProtocolError));
                }
                self.close_transport_abnormal();
                self.release_conversations();
                self.runtime
                    .mark_crashed(pid, ExitReason::Error, self.peer_addr);
                Some(SliceStep::Stop(ExitReason::Error))
            }
            Message::Ping(_) => {
                // tungstenite queued the pong reply internally (RFC 6455
                // transport control); it rides out with the next flush.
                self.outbound.note_transport_write_pending();
                None
            }
            Message::Pong(_) => None,
            Message::Close(close_frame) => {
                tracing::debug!(
                    connection_pid = pid,
                    close_frame = ?close_frame,
                    "websocket peer initiated close"
                );
                self.outbound.note_transport_write_pending();
                Some(self.finish_fate_close(pid, ConnectionFateClass::ConnectionLost))
            }
            Message::Frame(_) => {
                tracing::warn!(
                    connection_pid = pid,
                    "unexpected raw websocket frame surfaced by the transport library"
                );
                let bound = match self.bound_protocol_refusal() {
                    Ok(bound) => bound,
                    Err(error) => return Some(self.fail_fate(pid, &error)),
                };
                if bound {
                    return Some(self.finish_fate_close(pid, ConnectionFateClass::ProtocolError));
                }
                self.close_transport_abnormal();
                self.release_conversations();
                self.runtime
                    .mark_crashed(pid, ExitReason::Error, self.peer_addr);
                Some(SliceStep::Stop(ExitReason::Error))
            }
        }
    }

    /// Applies one complete binary message: the shared participant frame-limit
    /// preflight, the one-message-one-canonical-frame check, then the SAME
    /// `apply_frame` seam and `FrameAction` handling as the TCP process.
    pub(super) fn apply_binary(
        &mut self,
        pid: u64,
        bytes: &[u8],
    ) -> Result<ProcessStatus, ServerError> {
        if self.runtime.force_close_pending(pid) {
            return Ok(ProcessStatus::CloseWithFate(
                ConnectionFateClass::ServerShutdown,
            ));
        }
        if let Some(rejection) = crate::server::participant::preflight_generic_bytes(
            bytes,
            self.state.authenticated,
            self.state.participant_session,
        ) {
            let response = crate::server::participant::encode_server_value(
                liminal_protocol::wire::ServerValue::ParticipantTransportRejected(rejection),
            )
            .map_err(|error| ServerError::ListenerAccept {
                message: format!("failed to encode participant frame-limit rejection: {error:?}"),
            })?;
            self.outbound.enqueue_frame(&response).map_err(|error| {
                ServerError::ListenerAccept {
                    message: format!(
                        "failed to enqueue participant frame-limit rejection: {error}"
                    ),
                }
            })?;
            return Ok(ProcessStatus::Close);
        }
        let frame = match decode_ws_binary(bytes) {
            Ok(frame) => frame,
            Err(violation)
                if self.runtime.connection_has_bound_participant(
                    self.state.connection_incarnation,
                    &self.state.participant_conversations.tracked_conversations(),
                )? =>
            {
                tracing::warn!(connection_pid = pid, %violation, "bound websocket protocol refusal");
                return Ok(ProcessStatus::CloseWithFate(
                    ConnectionFateClass::ProtocolError,
                ));
            }
            Err(WsInboundViolation::MalformedFrame(
                error @ (liminal::protocol::ProtocolError::IncompleteHeader { .. }
                | liminal::protocol::ProtocolError::TruncatedPayload { .. }),
            )) => {
                return Err(ServerError::ListenerAccept {
                    message: format!(
                        "websocket contract violation: binary message is not one complete canonical liminal frame: {error}"
                    ),
                });
            }
            Err(WsInboundViolation::MalformedFrame(error)) => {
                return Err(server_error_from_protocol(&error));
            }
            Err(other) => {
                return Err(ServerError::ListenerAccept {
                    message: format!("websocket contract violation: {other}"),
                });
            }
        };
        match apply_frame(pid, &self.runtime, &mut self.state, frame) {
            FrameAction::Respond(response) => {
                self.outbound.enqueue_frame(&response).map_err(|error| {
                    ServerError::ListenerAccept {
                        message: format!("failed to enqueue connection response: {error}"),
                    }
                })?;
                Ok(ProcessStatus::Continue)
            }
            FrameAction::NoResponse => Ok(ProcessStatus::Continue),
            FrameAction::RespondThenClose(response) => {
                if let Err(error) = self.outbound.enqueue_frame(&response) {
                    tracing::warn!(
                        connection_pid = pid,
                        %error,
                        "auth-rejection frame could not be enqueued; closing anyway"
                    );
                }
                Ok(ProcessStatus::Close)
            }
            FrameAction::Close => Ok(ProcessStatus::Close),
            FrameAction::CloseWithFate(class) => Ok(ProcessStatus::CloseWithFate(class)),
        }
    }

    pub(super) fn bound_protocol_refusal(&self) -> Result<bool, ServerError> {
        let conversations = self.state.participant_conversations.tracked_conversations();
        self.runtime
            .connection_has_bound_participant(self.state.connection_incarnation, &conversations)
    }
}

/// The WebSocket transport's answer to a pre-admission refusal: RFC 6455
/// "message too big" (1009) with the refusal by name as its reason, the same
/// text the byte-stream transport's `ConnectError` carries.
pub(super) fn pre_admission_close_frame(refusal: PreAdmissionOversize) -> CloseFrame {
    CloseFrame {
        code: CloseCode::Size,
        reason: refusal.to_string().into(),
    }
}