liminal-server 0.25.4

//! Broker-coordinated direct KK links, with one echo and no carrier fallback.

use crate::link_carrier::{Carrier, CarrierError, read_frame, write_frame};
use crate::link_handshake::{
    LinkHandshake, LinkHandshakeError, LocalMachine, MachineIdentity, NOISE_OVERHEAD,
};
use crate::link_udp::{Datagram, UdpCarrier, UdpError};
use liminal::protocol::{
    CONVERSATION_REPLY_REQUESTED_FLAG, CausalContext, Frame, MessageEnvelope, SchemaId,
};
use serde::Deserialize;
use serde_json::{Value, json};
use std::io;
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;
use std::time::Duration;

pub use crate::link_udp::PunchSocket;
mod recovery;
pub use recovery::Instruction;

/// A named opening refusal; errors never select another carrier.
#[derive(Debug, thiserror::Error)]
pub enum PunchError {
    #[error("{0}")]
    Named(&'static str),
    /// A broker's own refusal, without request or credential bytes.
    #[error("broker_refused: {0}")]
    Broker(String),
    #[error("punch_io: {0}")]
    Io(#[from] io::Error),
    #[error(transparent)]
    Udp(#[from] UdpError),
    /// The expected machine failed its KK proof.
    #[error(transparent)]
    Noise(#[from] LinkHandshakeError),
    #[error(transparent)]
    Wire(#[from] CarrierError),
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Word {
    from: String,
    observed: Option<SocketAddr>,
    nonce: String,
}

/// Two already admitted broker wire connections: control and own-device words.
/// The caller admits both with its machine pass before handing them over.
pub struct WireBroker {
    control: TcpStream,
    words: TcpStream,
    next: u32,
    schema: Option<SchemaId>,
}

impl WireBroker {
    /// Takes two connections admitted by the same machine's Lys pass.
    #[must_use]
    pub const fn new(control: TcpStream, words: TcpStream) -> Self {
        Self {
            control,
            words,
            next: 2,
            schema: None,
        }
    }

    fn subscribe(&mut self, device: MachineIdentity) -> Result<(), PunchError> {
        write_frame(
            &mut self.words,
            &Frame::Subscribe {
                flags: 0,
                stream_id: 1,
                channel: format!("net.punch.{}", hex::encode(device.ed25519_public())),
                accepted_schemas: Vec::new(),
                max_in_flight: 1,
            },
        )?;
        match read_frame(&mut self.words)? {
            Some(Frame::SubscribeAck {
                stream_id: 1,
                selected_schema,
                ..
            }) => {
                self.schema = Some(selected_schema);
                Ok(())
            }
            _ => Err(PunchError::Named("punch_subscription_refused")),
        }
    }

    pub(crate) fn call(&mut self, subject: &str, payload: &Value) -> Result<Value, PunchError> {
        let id = self.next;
        self.next = id
            .checked_add(1)
            .ok_or(PunchError::Named("broker_streams_exhausted"))?;
        write_frame(
            &mut self.control,
            &Frame::ConversationOpen {
                flags: 0,
                stream_id: id,
                conversation_id: u64::from(id),
                subject: subject.to_owned(),
            },
        )?;
        write_frame(
            &mut self.control,
            &Frame::ConversationMessage {
                flags: CONVERSATION_REPLY_REQUESTED_FLAG,
                stream_id: id,
                conversation_id: u64::from(id),
                envelope: MessageEnvelope::new(
                    SchemaId::from_schema_bytes(b"liminal.net.punch-control/1"),
                    CausalContext::independent(),
                    json!({"operation": format!("punch-{id}"), "payload": payload})
                        .to_string()
                        .into_bytes(),
                ),
            },
        )?;
        let answer = match read_frame(&mut self.control)? {
            Some(Frame::ConversationMessage {
                stream_id,
                conversation_id,
                envelope,
                ..
            }) if stream_id == id && conversation_id == u64::from(id) => {
                serde_json::from_slice(&envelope.payload)
                    .map_err(|_| PunchError::Named("invalid_broker_answer"))
            }
            Some(Frame::ConversationError {
                stream_id,
                conversation_id,
                message,
                ..
            }) if stream_id == id && conversation_id == u64::from(id) => Err(PunchError::Broker(
                message.ok_or(PunchError::Named("unnamed_broker_refusal"))?,
            )),
            _ => Err(PunchError::Named("unexpected_broker_answer")),
        };
        write_frame(
            &mut self.control,
            &Frame::ConversationClose {
                flags: 0,
                stream_id: id,
                conversation_id: u64::from(id),
                reason_code: None,
                message: None,
            },
        )?;
        answer
    }

    fn word(&mut self, peer: MachineIdentity, nonce: Option<&str>) -> Result<Word, PunchError> {
        let Some(Frame::Deliver {
            stream_id: 1,
            envelope,
            ..
        }) = read_frame(&mut self.words)?
        else {
            return Err(PunchError::Named("punch_word_missing"));
        };
        require(
            Some(envelope.schema_id) == self.schema,
            "invalid_punch_schema",
        )?;
        let word: Word = serde_json::from_slice(&envelope.payload)
            .map_err(|_| PunchError::Named("invalid_punch_word"))?;
        if word.from != hex::encode(peer.ed25519_public())
            || nonce.is_some_and(|nonce| nonce != word.nonce)
            || !hex::decode(&word.nonce).is_ok_and(|bytes| bytes.len() == 16)
            || word.observed.is_none_or(|address| address.port() == 0)
        {
            return Err(PunchError::Named("unexpected_punch_word"));
        }
        Ok(word)
    }
}

/// The authenticated direct path retained by the device's UDP carrier.
#[derive(Debug)]
pub struct PunchedLink {
    pub peer: MachineIdentity,
    pub local_index: u32,
    pub endpoint: SocketAddr,
    /// Always direct after successful KK completion.
    pub carrier: Carrier,
    /// Whether this device sent kind 1.
    pub initiator: bool,
}

/// One device's boot subscription and one explicitly observed opening attempt.
pub struct Puncher {
    local: LocalMachine,
    socket: Arc<dyn PunchSocket>,
    broker: WireBroker,
    echo: SocketAddr,
    bound: Duration,
    observed: Option<SocketAddr>,
    carrier: Option<Arc<UdpCarrier>>,
    attempted: bool,
    requests: Arc<std::sync::Mutex<std::collections::VecDeque<String>>>,
}

impl Puncher {
    /// Subscribes before registration or any request can publish a word.
    /// # Errors
    /// Refuses invalid bounds/endpoints or an unadmitted subscription.
    pub fn boot(
        local: LocalMachine,
        socket: Arc<dyn PunchSocket>,
        mut broker: WireBroker,
        echo: SocketAddr,
        bound: Duration,
    ) -> Result<Self, PunchError> {
        require(
            echo.port() != 0 && !bound.is_zero(),
            "invalid_echo_configuration",
        )?;
        broker.subscribe(local.machine())?;
        Ok(Self {
            local,
            socket,
            broker,
            echo,
            bound,
            observed: None,
            carrier: None,
            attempted: false,
            requests: Arc::default(),
        })
    }

    /// Sends one echo. A lost answer ends this attempt by name, without retry.
    /// # Errors
    /// Refuses a repeated observation, an unanswered/invalid echo or socket failure.
    pub fn observe(&mut self) -> Result<SocketAddr, PunchError> {
        require(!self.attempted, "echo_already_attempted")?;
        self.attempted = true;
        let nonce = fresh_nonce()?;
        let mut request = b"LMOB".to_vec();
        request.extend_from_slice(&nonce);
        self.socket.prepare(Some(self.bound))?;
        let received = (|| {
            require(
                self.socket.send_to(&request, self.echo)? == request.len(),
                "incomplete_echo_send",
            )?;
            let mut answer = [0; 40];
            let (length, source) = self.socket.recv_from(&mut answer).map_err(|error| {
                if matches!(
                    error.kind(),
                    io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
                ) {
                    PunchError::Named("broker_echo_unanswered")
                } else {
                    PunchError::Io(error)
                }
            })?;
            recovery::observed_endpoint(&nonce, self.echo, source, &answer[..length])
        })();
        self.socket.prepare(None)?;
        let observed = received?;
        self.carrier = Some(Arc::new(UdpCarrier::from_socket(Arc::clone(&self.socket))?));
        self.observed = Some(observed);
        Ok(observed)
    }

    /// Registers this attempt's echo address, with no invented reachable address.
    /// # Errors
    /// Refuses missing observation or a broker registration refusal.
    pub fn register(&mut self) -> Result<(), PunchError> {
        let observed = self
            .observed
            .ok_or(PunchError::Named("device_not_observed"))?;
        accepted(&self.broker.call("net.broker.register", &json!({"device": hex::encode(self.local.machine().ed25519_public()), "addresses": [], "observed": observed}))?)
    }

    /// Publishes one instruction through the admitted broker control path.
    /// # Errors
    /// Refuses self-punching, invalid keys/nonces, or a broker refusal.
    pub fn request_punch(&mut self, peer: MachineIdentity, nonce: &str) -> Result<(), PunchError> {
        self.validate_peer(peer)?;
        require(
            hex::decode(nonce).is_ok_and(|bytes| bytes.len() == 16),
            "invalid_punch_nonce",
        )?;
        let mut requests = self
            .requests
            .lock()
            .map_err(|_| PunchError::Named("punch_request_state_poisoned"))?;
        if !requests.iter().any(|request| request == nonce) {
            requests.push_back(nonce.to_owned());
        }
        while requests.len() > 2048 {
            requests.pop_front();
        }
        drop(requests);
        accepted(&self.broker.call(
            "net.broker.punch",
            &json!({"to": hex::encode(peer.ed25519_public()), "nonce": nonce}),
        )?)
    }

    fn validate_peer(&self, peer: MachineIdentity) -> Result<(), PunchError> {
        require(peer != self.local.machine(), "cannot_punch_self")?;
        peer.x25519_public()?;
        Ok(())
    }

    fn probe(&self, endpoint: SocketAddr) -> Result<(), PunchError> {
        require(endpoint.port() != 0, "peer_not_observed")?;
        self.carrier()?.transmit(
            endpoint,
            &Datagram {
                kind: 1,
                receiver_index: 0,
                counter: 0,
                payload: Vec::new(),
            },
        )?;
        Ok(())
    }

    fn handshake(
        &self,
        peer: MachineIdentity,
        endpoint: SocketAddr,
        index: u32,
    ) -> Result<PunchedLink, PunchError> {
        require(index != 0, "invalid_punch_index")?;
        let deadline = std::time::Instant::now()
            .checked_add(self.bound)
            .ok_or(PunchError::Named("punch_deadline_overflow"))?;
        let initiator = self.local.machine().ed25519_public() < peer.ed25519_public();
        let mut handshake = if initiator {
            LinkHandshake::initiator(&self.local, peer)?
        } else {
            LinkHandshake::responder(&self.local, peer)?
        };
        if initiator {
            self.carrier()?
                .transmit(endpoint, &handshake_datagram(&mut handshake, 1, 0, index)?)?;
        }
        let datagram = loop {
            let (datagram, source) = self.carrier()?.receive_from_until(deadline)?;
            require(source == endpoint, "unexpected_punch_endpoint")?;
            if datagram
                == (Datagram {
                    kind: 1,
                    receiver_index: 0,
                    counter: 0,
                    payload: Vec::new(),
                })
            {
                continue;
            }
            break datagram;
        };
        if datagram.kind != if initiator { 2 } else { 1 }
            || datagram.receiver_index != if initiator { index } else { 0 }
        {
            return Err(PunchError::Named("unexpected_punch_handshake"));
        }
        handshake.read(&datagram.payload)?;
        let remote_index = u32::try_from(datagram.counter)
            .map_err(|_| PunchError::Named("invalid_punch_index"))?;
        require(remote_index != 0, "invalid_punch_index")?;
        let response = if initiator {
            None
        } else {
            Some(handshake_datagram(&mut handshake, 2, remote_index, index)?)
        };
        self.carrier()?
            .add_link(index, remote_index, endpoint, handshake.finish()?)?;
        let result = (|| {
            if let Some(response) = response {
                self.carrier()?.transmit(endpoint, &response)?;
            }
            if initiator {
                self.carrier()?.send(index, b"LMPR")?;
            }
            loop {
                let (ready, source) = self.carrier()?.receive_from_until(deadline)?;
                require(
                    source == endpoint && ready.receiver_index == index,
                    "unexpected_punch_ready",
                )?;
                if ready.kind == 4 {
                    continue;
                }
                require(
                    ready.kind == 3 && ready.payload == b"LMPR",
                    "unexpected_punch_ready",
                )?;
                break;
            }
            if !initiator {
                self.carrier()?.send(index, b"LMPR")?;
            }
            Ok(PunchedLink {
                peer,
                local_index: index,
                endpoint,
                carrier: Carrier::Direct,
                initiator,
            })
        })();
        if result.is_err()
            && let Err(error) = self.carrier()?.remove_link(index)
        {
            tracing::error!(%error, index, "failed punch slot retirement failed");
        }
        result
    }
}

fn accepted(answer: &Value) -> Result<(), PunchError> {
    require(answer == &json!({"ok": true}), "invalid_broker_answer")
}

fn fresh_nonce() -> Result<[u8; 16], PunchError> {
    use std::io::Read;
    let mut bytes = [0; 16];
    std::fs::File::open("/dev/urandom")?.read_exact(&mut bytes)?;
    Ok(bytes)
}

fn handshake_datagram(
    handshake: &mut LinkHandshake,
    kind: u8,
    receiver_index: u32,
    index: u32,
) -> Result<Datagram, PunchError> {
    let mut payload = vec![0; NOISE_OVERHEAD];
    let length = handshake.write(&mut payload)?;
    payload.truncate(length);
    Ok(Datagram {
        kind,
        receiver_index,
        counter: u64::from(index),
        payload,
    })
}

const fn require(accepted: bool, refusal: &'static str) -> Result<(), PunchError> {
    if accepted {
        Ok(())
    } else {
        Err(PunchError::Named(refusal))
    }
}