liminal-server 0.25.4

//! Datagram setup carries the same machine admission and link-open frames as the wire.
use super::{DirectPath, OpenError, named};
use crate::config::types::OutboundLinkConfig;
use crate::link_handshake::{LinkHandshake, MachineIdentity, NOISE_OVERHEAD};
use liminal::protocol::{
    Frame, LinkEnd, LinkOffer, LinkOpenOutcome, ProtocolVersion, REVOCATION_CLOSE_VERSION, codec,
};
use liminal_protocol::credential::WireCredential;
use std::time::Instant;

impl DirectPath {
    pub(crate) fn admit(
        &mut self,
        host: &crate::link::LinkHost,
        agent: &str,
        peer: MachineIdentity,
        config: &OutboundLinkConfig,
        token: &str,
        deadline: Instant,
    ) -> Result<(), OpenError> {
        let connect = Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: REVOCATION_CLOSE_VERSION,
            auth_token: WireCredential::lys_record(token.trim().as_bytes()),
        };
        match self.exchange(&connect, deadline)? {
            Frame::ConnectAck {
                selected_version, ..
            } if (ProtocolVersion::new(1, 0)..=REVOCATION_CLOSE_VERSION)
                .contains(&selected_version) => {}
            Frame::ConnectError { reason_code, .. } => {
                return Err(OpenError(format!("link_refused:udp_connect_{reason_code}")));
            }
            _ => return Err(OpenError("unexpected_udp_connect_answer".to_owned())),
        }
        let local = host.local();
        let mut handshake = LinkHandshake::initiator(local, peer).map_err(named)?;
        let mut bytes = [0; NOISE_OVERHEAD];
        let length = handshake.write(&mut bytes).map_err(named)?;
        let offer = LinkOffer {
            link_id: config.link_id.clone(),
            machine: *local.machine().ed25519_public(),
            end: LinkEnd::Export,
            handshake: crate::link::certificate::offer::for_machine(host, agent, &bytes[..length])
                .map_err(named)?,
        };
        match self.exchange(
            &Frame::LinkOpen {
                flags: 0,
                stream_id: 1,
                offer,
            },
            deadline,
        )? {
            Frame::LinkOpenAck {
                stream_id: 1,
                outcome: LinkOpenOutcome::Opened { handshake: answer },
                ..
            } => {
                handshake.read(&answer).map_err(named)?;
                self.transport = Some(handshake.finish().map_err(named)?);
                Ok(())
            }
            Frame::LinkOpenAck {
                stream_id: 1,
                outcome: LinkOpenOutcome::Held { grant, draft },
                ..
            } => Err(OpenError(format!("link_held:{grant}:{draft}"))),
            Frame::LinkOpenAck {
                stream_id: 1,
                outcome: LinkOpenOutcome::Refused { reason },
                ..
            } => Err(OpenError(format!("link_refused:{reason}"))),
            _ => Err(OpenError("unexpected_udp_link_open_answer".to_owned())),
        }
    }
    fn exchange(&self, frame: &Frame, deadline: Instant) -> Result<Frame, OpenError> {
        let length = codec::encoded_len(frame).map_err(named)?;
        if length > crate::link_carrier::WIRE_FRAME_BOUND {
            return Err(OpenError("udp_control_frame_too_large".to_owned()));
        }
        let mut bytes = vec![0; length];
        codec::encode(frame, &mut bytes).map_err(named)?;
        let total = u32::try_from(length).map_err(named)?;
        let mut last = None;
        // Only admission is chunked; application crossings retain one datagram and one deadline.
        for (index, chunk) in bytes.chunks(1000).enumerate() {
            let mut request = b"LMUC".to_vec();
            request.extend_from_slice(&total.to_be_bytes());
            request.extend_from_slice(&u32::try_from(index * 1000).map_err(named)?.to_be_bytes());
            request.extend(chunk);
            let counter = self.carrier.send(self.index, &request).map_err(named)?;
            if !self
                .carrier
                .wait_acknowledged_until(self.index, counter, deadline)
                .map_err(named)?
            {
                return Err(OpenError("udp_open_datagram_unanswered".to_owned()));
            }
            last = Some(counter);
        }
        self.answer(
            last.ok_or_else(|| OpenError("empty_udp_control_frame".to_owned()))?,
            deadline,
        )
    }
    pub(super) fn answer(&self, counter: u64, deadline: Instant) -> Result<Frame, OpenError> {
        let datagram = self.inbox.until(deadline).map_err(named)?;
        let payload = datagram.payload;
        if payload.get(..4) != Some(b"LMUR")
            || payload.get(4..12) != Some(counter.to_be_bytes().as_slice())
        {
            return Err(OpenError("unexpected_udp_answer_counter".to_owned()));
        }
        let bytes = payload
            .get(12..)
            .ok_or_else(|| OpenError("short_udp_answer".to_owned()))?;
        let (frame, read) = codec::decode(bytes).map_err(named)?;
        if read != bytes.len() {
            return Err(OpenError("udp_answer_has_trailing_bytes".to_owned()));
        }
        Ok(frame)
    }
}