liminal-server 0.25.4

//! The inner service wire is validated before creating a far connection.

use super::{CallError, CallPacket, CallReply, CallRequest, Frame, Zeroizing, codec};
use liminal::protocol::CONVERSATION_REPLY_REQUESTED_FLAG;
use liminal_protocol::credential::WireCredential;

// Position zero is never a durable subscription delivery. It reserves an
// extension space without changing the ordinary positive-position crossings.
pub(super) const MAGIC: &[u8; 12] = b"\0\0\0\0\0\0\0\0LMC1";

pub(super) fn encode_frames(
    frames: &[Frame],
    bound: usize,
) -> Result<Zeroizing<Vec<u8>>, CallError> {
    let mut bytes = Zeroizing::new(Vec::new());
    for frame in frames {
        let length = codec::encoded_len(frame).map_err(|_| CallError::Invalid)?;
        let end = bytes
            .len()
            .checked_add(length)
            .filter(|length| *length <= bound)
            .ok_or(CallError::Bound)?;
        let start = bytes.len();
        bytes.resize(end, 0);
        codec::encode(frame, &mut bytes[start..]).map_err(|_| CallError::Invalid)?;
    }
    Ok(bytes)
}

pub(super) fn request_frames(bytes: &[u8], subject: &str) -> Result<Vec<Frame>, CallError> {
    let (connect, length) = codec::decode(bytes).map_err(|_| CallError::Invalid)?;
    let (open, next) = codec::decode(bytes.get(length..).ok_or(CallError::Invalid)?)
        .map_err(|_| CallError::Invalid)?;
    let offset = length.checked_add(next).ok_or(CallError::Invalid)?;
    let (message, next) = codec::decode(bytes.get(offset..).ok_or(CallError::Invalid)?)
        .map_err(|_| CallError::Invalid)?;
    if offset.checked_add(next) != Some(bytes.len()) {
        return Err(CallError::Invalid);
    }
    let Frame::Connect { auth_token, .. } = &connect else {
        return Err(CallError::Invalid);
    };
    if !matches!(
        WireCredential::classify(auth_token),
        Ok(WireCredential::LysPass(_))
    ) {
        return Err(CallError::Invalid);
    }
    let Frame::ConversationOpen {
        stream_id,
        conversation_id,
        subject: placed,
        ..
    } = &open
    else {
        return Err(CallError::Invalid);
    };
    let Frame::ConversationMessage {
        flags,
        stream_id: request_stream,
        conversation_id: request_id,
        ..
    } = &message
    else {
        return Err(CallError::Invalid);
    };
    if *stream_id == 0
        || *conversation_id == 0
        || placed != subject
        || stream_id != request_stream
        || conversation_id != request_id
        || flags & CONVERSATION_REPLY_REQUESTED_FLAG == 0
    {
        return Err(CallError::Invalid);
    }
    Ok(vec![connect, open, message])
}

pub(super) fn encode(packet: &CallPacket, bound: usize) -> Result<Zeroizing<Vec<u8>>, CallError> {
    let (kind, id, subject, body) = match packet {
        CallPacket::Request(request) => (
            1,
            request.id,
            request.subject.as_bytes(),
            request.bytes.clone(),
        ),
        CallPacket::Reply(reply) => (
            2,
            reply.id,
            &[][..],
            encode_frames(std::slice::from_ref(&reply.frame), bound)?,
        ),
        CallPacket::Cancel(id) => (3, *id, &[][..], Zeroizing::new(Vec::new())),
        CallPacket::Topology(bytes) => (4, 0, &[][..], bytes.clone()),
    };
    if kind != 4 && id == 0 {
        return Err(CallError::Invalid);
    }
    let length = MAGIC
        .len()
        .checked_add(17)
        .and_then(|length| length.checked_add(subject.len()))
        .and_then(|length| length.checked_add(body.len()))
        .ok_or(CallError::Bound)?;
    if length > bound
        || length.saturating_add(crate::link_handshake::NOISE_OVERHEAD)
            > crate::link_handshake::NOISE_MAX_MESSAGE
    {
        return Err(CallError::Bound);
    }
    let mut bytes = Zeroizing::new(Vec::with_capacity(length));
    bytes.extend_from_slice(MAGIC);
    bytes.push(kind);
    bytes.extend_from_slice(&id.to_be_bytes());
    for length in [subject.len(), body.len()] {
        bytes.extend_from_slice(
            &u32::try_from(length)
                .map_err(|_| CallError::Bound)?
                .to_be_bytes(),
        );
    }
    bytes.extend_from_slice(subject);
    bytes.extend_from_slice(&body);
    Ok(bytes)
}

pub(super) fn decode(bytes: &[u8], bound: usize) -> Result<CallPacket, CallError> {
    if bytes.len() > bound || !bytes.starts_with(MAGIC) {
        return Err(CallError::Bound);
    }
    let mut remaining = bytes.get(MAGIC.len()..).ok_or(CallError::Invalid)?;
    let kind = take(&mut remaining, 1)?[0];
    let id = u64::from_be_bytes(
        take(&mut remaining, 8)?
            .try_into()
            .map_err(|_| CallError::Invalid)?,
    );
    let subject_len = size(&mut remaining)?;
    let body_len = size(&mut remaining)?;
    let subject =
        std::str::from_utf8(take(&mut remaining, subject_len)?).map_err(|_| CallError::Invalid)?;
    let body = take(&mut remaining, body_len)?;
    if !remaining.is_empty() {
        return Err(CallError::Invalid);
    }
    match (kind, id, subject) {
        (1, id, subject) if id > 0 => Ok(CallPacket::Request(CallRequest::decode(
            id,
            subject.into(),
            Zeroizing::new(body.to_vec()),
            bound,
        )?)),
        (2, id, "") if id > 0 => {
            let (frame, used) = codec::decode(body).map_err(|_| CallError::Invalid)?;
            if used != body.len() {
                return Err(CallError::Invalid);
            }
            Ok(CallPacket::Reply(CallReply::new(id, frame)?))
        }
        (3, id, "") if id > 0 && body.is_empty() => Ok(CallPacket::Cancel(id)),
        (4, 0, "") => Ok(CallPacket::Topology(Zeroizing::new(body.to_vec()))),
        _ => Err(CallError::Invalid),
    }
}

const fn take<'a>(bytes: &mut &'a [u8], length: usize) -> Result<&'a [u8], CallError> {
    if length > bytes.len() {
        return Err(CallError::Invalid);
    }
    let (taken, rest) = bytes.split_at(length);
    *bytes = rest;
    Ok(taken)
}
fn size(bytes: &mut &[u8]) -> Result<usize, CallError> {
    usize::try_from(u32::from_be_bytes(
        take(bytes, 4)?.try_into().map_err(|_| CallError::Invalid)?,
    ))
    .map_err(|_| CallError::Invalid)
}