liminal_server/
link_punch.rs

1//! Broker-coordinated direct KK links, with one echo and no carrier fallback.
2
3use crate::link_carrier::{Carrier, CarrierError, read_frame, write_frame};
4use crate::link_handshake::{
5    LinkHandshake, LinkHandshakeError, LocalMachine, MachineIdentity, NOISE_OVERHEAD,
6};
7use crate::link_udp::{Datagram, UdpCarrier, UdpError};
8use liminal::protocol::{
9    CONVERSATION_REPLY_REQUESTED_FLAG, CausalContext, Frame, MessageEnvelope, SchemaId,
10};
11use serde::Deserialize;
12use serde_json::{Value, json};
13use std::io;
14use std::net::{SocketAddr, TcpStream};
15use std::sync::Arc;
16use std::time::Duration;
17
18pub use crate::link_udp::PunchSocket;
19mod recovery;
20pub use recovery::Instruction;
21
22/// A named opening refusal; errors never select another carrier.
23#[derive(Debug, thiserror::Error)]
24pub enum PunchError {
25    #[error("{0}")]
26    Named(&'static str),
27    /// A broker's own refusal, without request or credential bytes.
28    #[error("broker_refused: {0}")]
29    Broker(String),
30    #[error("punch_io: {0}")]
31    Io(#[from] io::Error),
32    #[error(transparent)]
33    Udp(#[from] UdpError),
34    /// The expected machine failed its KK proof.
35    #[error(transparent)]
36    Noise(#[from] LinkHandshakeError),
37    #[error(transparent)]
38    Wire(#[from] CarrierError),
39}
40
41#[derive(Deserialize)]
42#[serde(deny_unknown_fields)]
43struct Word {
44    from: String,
45    observed: Option<SocketAddr>,
46    nonce: String,
47}
48
49/// Two already admitted broker wire connections: control and own-device words.
50/// The caller admits both with its machine pass before handing them over.
51pub struct WireBroker {
52    control: TcpStream,
53    words: TcpStream,
54    next: u32,
55    schema: Option<SchemaId>,
56}
57
58impl WireBroker {
59    /// Takes two connections admitted by the same machine's Lys pass.
60    #[must_use]
61    pub const fn new(control: TcpStream, words: TcpStream) -> Self {
62        Self {
63            control,
64            words,
65            next: 2,
66            schema: None,
67        }
68    }
69
70    fn subscribe(&mut self, device: MachineIdentity) -> Result<(), PunchError> {
71        write_frame(
72            &mut self.words,
73            &Frame::Subscribe {
74                flags: 0,
75                stream_id: 1,
76                channel: format!("net.punch.{}", hex::encode(device.ed25519_public())),
77                accepted_schemas: Vec::new(),
78                max_in_flight: 1,
79            },
80        )?;
81        match read_frame(&mut self.words)? {
82            Some(Frame::SubscribeAck {
83                stream_id: 1,
84                selected_schema,
85                ..
86            }) => {
87                self.schema = Some(selected_schema);
88                Ok(())
89            }
90            _ => Err(PunchError::Named("punch_subscription_refused")),
91        }
92    }
93
94    pub(crate) fn call(&mut self, subject: &str, payload: &Value) -> Result<Value, PunchError> {
95        let id = self.next;
96        self.next = id
97            .checked_add(1)
98            .ok_or(PunchError::Named("broker_streams_exhausted"))?;
99        write_frame(
100            &mut self.control,
101            &Frame::ConversationOpen {
102                flags: 0,
103                stream_id: id,
104                conversation_id: u64::from(id),
105                subject: subject.to_owned(),
106            },
107        )?;
108        write_frame(
109            &mut self.control,
110            &Frame::ConversationMessage {
111                flags: CONVERSATION_REPLY_REQUESTED_FLAG,
112                stream_id: id,
113                conversation_id: u64::from(id),
114                envelope: MessageEnvelope::new(
115                    SchemaId::from_schema_bytes(b"liminal.net.punch-control/1"),
116                    CausalContext::independent(),
117                    json!({"operation": format!("punch-{id}"), "payload": payload})
118                        .to_string()
119                        .into_bytes(),
120                ),
121            },
122        )?;
123        let answer = match read_frame(&mut self.control)? {
124            Some(Frame::ConversationMessage {
125                stream_id,
126                conversation_id,
127                envelope,
128                ..
129            }) if stream_id == id && conversation_id == u64::from(id) => {
130                serde_json::from_slice(&envelope.payload)
131                    .map_err(|_| PunchError::Named("invalid_broker_answer"))
132            }
133            Some(Frame::ConversationError {
134                stream_id,
135                conversation_id,
136                message,
137                ..
138            }) if stream_id == id && conversation_id == u64::from(id) => Err(PunchError::Broker(
139                message.ok_or(PunchError::Named("unnamed_broker_refusal"))?,
140            )),
141            _ => Err(PunchError::Named("unexpected_broker_answer")),
142        };
143        write_frame(
144            &mut self.control,
145            &Frame::ConversationClose {
146                flags: 0,
147                stream_id: id,
148                conversation_id: u64::from(id),
149                reason_code: None,
150                message: None,
151            },
152        )?;
153        answer
154    }
155
156    fn word(&mut self, peer: MachineIdentity, nonce: Option<&str>) -> Result<Word, PunchError> {
157        let Some(Frame::Deliver {
158            stream_id: 1,
159            envelope,
160            ..
161        }) = read_frame(&mut self.words)?
162        else {
163            return Err(PunchError::Named("punch_word_missing"));
164        };
165        require(
166            Some(envelope.schema_id) == self.schema,
167            "invalid_punch_schema",
168        )?;
169        let word: Word = serde_json::from_slice(&envelope.payload)
170            .map_err(|_| PunchError::Named("invalid_punch_word"))?;
171        if word.from != hex::encode(peer.ed25519_public())
172            || nonce.is_some_and(|nonce| nonce != word.nonce)
173            || !hex::decode(&word.nonce).is_ok_and(|bytes| bytes.len() == 16)
174            || word.observed.is_none_or(|address| address.port() == 0)
175        {
176            return Err(PunchError::Named("unexpected_punch_word"));
177        }
178        Ok(word)
179    }
180}
181
182/// The authenticated direct path retained by the device's UDP carrier.
183#[derive(Debug)]
184pub struct PunchedLink {
185    pub peer: MachineIdentity,
186    pub local_index: u32,
187    pub endpoint: SocketAddr,
188    /// Always direct after successful KK completion.
189    pub carrier: Carrier,
190    /// Whether this device sent kind 1.
191    pub initiator: bool,
192}
193
194/// One device's boot subscription and one explicitly observed opening attempt.
195pub struct Puncher {
196    local: LocalMachine,
197    socket: Arc<dyn PunchSocket>,
198    broker: WireBroker,
199    echo: SocketAddr,
200    bound: Duration,
201    observed: Option<SocketAddr>,
202    carrier: Option<Arc<UdpCarrier>>,
203    attempted: bool,
204    requests: Arc<std::sync::Mutex<std::collections::VecDeque<String>>>,
205}
206
207impl Puncher {
208    /// Subscribes before registration or any request can publish a word.
209    /// # Errors
210    /// Refuses invalid bounds/endpoints or an unadmitted subscription.
211    pub fn boot(
212        local: LocalMachine,
213        socket: Arc<dyn PunchSocket>,
214        mut broker: WireBroker,
215        echo: SocketAddr,
216        bound: Duration,
217    ) -> Result<Self, PunchError> {
218        require(
219            echo.port() != 0 && !bound.is_zero(),
220            "invalid_echo_configuration",
221        )?;
222        broker.subscribe(local.machine())?;
223        Ok(Self {
224            local,
225            socket,
226            broker,
227            echo,
228            bound,
229            observed: None,
230            carrier: None,
231            attempted: false,
232            requests: Arc::default(),
233        })
234    }
235
236    /// Sends one echo. A lost answer ends this attempt by name, without retry.
237    /// # Errors
238    /// Refuses a repeated observation, an unanswered/invalid echo or socket failure.
239    pub fn observe(&mut self) -> Result<SocketAddr, PunchError> {
240        require(!self.attempted, "echo_already_attempted")?;
241        self.attempted = true;
242        let nonce = fresh_nonce()?;
243        let mut request = b"LMOB".to_vec();
244        request.extend_from_slice(&nonce);
245        self.socket.prepare(Some(self.bound))?;
246        let received = (|| {
247            require(
248                self.socket.send_to(&request, self.echo)? == request.len(),
249                "incomplete_echo_send",
250            )?;
251            let mut answer = [0; 40];
252            let (length, source) = self.socket.recv_from(&mut answer).map_err(|error| {
253                if matches!(
254                    error.kind(),
255                    io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock
256                ) {
257                    PunchError::Named("broker_echo_unanswered")
258                } else {
259                    PunchError::Io(error)
260                }
261            })?;
262            recovery::observed_endpoint(&nonce, self.echo, source, &answer[..length])
263        })();
264        self.socket.prepare(None)?;
265        let observed = received?;
266        self.carrier = Some(Arc::new(UdpCarrier::from_socket(Arc::clone(&self.socket))?));
267        self.observed = Some(observed);
268        Ok(observed)
269    }
270
271    /// Registers this attempt's echo address, with no invented reachable address.
272    /// # Errors
273    /// Refuses missing observation or a broker registration refusal.
274    pub fn register(&mut self) -> Result<(), PunchError> {
275        let observed = self
276            .observed
277            .ok_or(PunchError::Named("device_not_observed"))?;
278        accepted(&self.broker.call("net.broker.register", &json!({"device": hex::encode(self.local.machine().ed25519_public()), "addresses": [], "observed": observed}))?)
279    }
280
281    /// Publishes one instruction through the admitted broker control path.
282    /// # Errors
283    /// Refuses self-punching, invalid keys/nonces, or a broker refusal.
284    pub fn request_punch(&mut self, peer: MachineIdentity, nonce: &str) -> Result<(), PunchError> {
285        self.validate_peer(peer)?;
286        require(
287            hex::decode(nonce).is_ok_and(|bytes| bytes.len() == 16),
288            "invalid_punch_nonce",
289        )?;
290        let mut requests = self
291            .requests
292            .lock()
293            .map_err(|_| PunchError::Named("punch_request_state_poisoned"))?;
294        if !requests.iter().any(|request| request == nonce) {
295            requests.push_back(nonce.to_owned());
296        }
297        while requests.len() > 2048 {
298            requests.pop_front();
299        }
300        drop(requests);
301        accepted(&self.broker.call(
302            "net.broker.punch",
303            &json!({"to": hex::encode(peer.ed25519_public()), "nonce": nonce}),
304        )?)
305    }
306
307    fn validate_peer(&self, peer: MachineIdentity) -> Result<(), PunchError> {
308        require(peer != self.local.machine(), "cannot_punch_self")?;
309        peer.x25519_public()?;
310        Ok(())
311    }
312
313    fn probe(&self, endpoint: SocketAddr) -> Result<(), PunchError> {
314        require(endpoint.port() != 0, "peer_not_observed")?;
315        self.carrier()?.transmit(
316            endpoint,
317            &Datagram {
318                kind: 1,
319                receiver_index: 0,
320                counter: 0,
321                payload: Vec::new(),
322            },
323        )?;
324        Ok(())
325    }
326
327    fn handshake(
328        &self,
329        peer: MachineIdentity,
330        endpoint: SocketAddr,
331        index: u32,
332    ) -> Result<PunchedLink, PunchError> {
333        require(index != 0, "invalid_punch_index")?;
334        let deadline = std::time::Instant::now()
335            .checked_add(self.bound)
336            .ok_or(PunchError::Named("punch_deadline_overflow"))?;
337        let initiator = self.local.machine().ed25519_public() < peer.ed25519_public();
338        let mut handshake = if initiator {
339            LinkHandshake::initiator(&self.local, peer)?
340        } else {
341            LinkHandshake::responder(&self.local, peer)?
342        };
343        if initiator {
344            self.carrier()?
345                .transmit(endpoint, &handshake_datagram(&mut handshake, 1, 0, index)?)?;
346        }
347        let datagram = loop {
348            let (datagram, source) = self.carrier()?.receive_from_until(deadline)?;
349            require(source == endpoint, "unexpected_punch_endpoint")?;
350            if datagram
351                == (Datagram {
352                    kind: 1,
353                    receiver_index: 0,
354                    counter: 0,
355                    payload: Vec::new(),
356                })
357            {
358                continue;
359            }
360            break datagram;
361        };
362        if datagram.kind != if initiator { 2 } else { 1 }
363            || datagram.receiver_index != if initiator { index } else { 0 }
364        {
365            return Err(PunchError::Named("unexpected_punch_handshake"));
366        }
367        handshake.read(&datagram.payload)?;
368        let remote_index = u32::try_from(datagram.counter)
369            .map_err(|_| PunchError::Named("invalid_punch_index"))?;
370        require(remote_index != 0, "invalid_punch_index")?;
371        let response = if initiator {
372            None
373        } else {
374            Some(handshake_datagram(&mut handshake, 2, remote_index, index)?)
375        };
376        self.carrier()?
377            .add_link(index, remote_index, endpoint, handshake.finish()?)?;
378        let result = (|| {
379            if let Some(response) = response {
380                self.carrier()?.transmit(endpoint, &response)?;
381            }
382            if initiator {
383                self.carrier()?.send(index, b"LMPR")?;
384            }
385            loop {
386                let (ready, source) = self.carrier()?.receive_from_until(deadline)?;
387                require(
388                    source == endpoint && ready.receiver_index == index,
389                    "unexpected_punch_ready",
390                )?;
391                if ready.kind == 4 {
392                    continue;
393                }
394                require(
395                    ready.kind == 3 && ready.payload == b"LMPR",
396                    "unexpected_punch_ready",
397                )?;
398                break;
399            }
400            if !initiator {
401                self.carrier()?.send(index, b"LMPR")?;
402            }
403            Ok(PunchedLink {
404                peer,
405                local_index: index,
406                endpoint,
407                carrier: Carrier::Direct,
408                initiator,
409            })
410        })();
411        if result.is_err()
412            && let Err(error) = self.carrier()?.remove_link(index)
413        {
414            tracing::error!(%error, index, "failed punch slot retirement failed");
415        }
416        result
417    }
418}
419
420fn accepted(answer: &Value) -> Result<(), PunchError> {
421    require(answer == &json!({"ok": true}), "invalid_broker_answer")
422}
423
424fn fresh_nonce() -> Result<[u8; 16], PunchError> {
425    use std::io::Read;
426    let mut bytes = [0; 16];
427    std::fs::File::open("/dev/urandom")?.read_exact(&mut bytes)?;
428    Ok(bytes)
429}
430
431fn handshake_datagram(
432    handshake: &mut LinkHandshake,
433    kind: u8,
434    receiver_index: u32,
435    index: u32,
436) -> Result<Datagram, PunchError> {
437    let mut payload = vec![0; NOISE_OVERHEAD];
438    let length = handshake.write(&mut payload)?;
439    payload.truncate(length);
440    Ok(Datagram {
441        kind,
442        receiver_index,
443        counter: u64::from(index),
444        payload,
445    })
446}
447
448const fn require(accepted: bool, refusal: &'static str) -> Result<(), PunchError> {
449    if accepted {
450        Ok(())
451    } else {
452        Err(PunchError::Named(refusal))
453    }
454}
liminal_server/link_punch.rs

liminal_server/
link_punch.rs
