liminal_server/link_punch/
recovery.rs

1//! Opening and refreshing a direct path retain distinct responsibilities.
2use super::{PunchError, PunchedLink, Puncher, UdpCarrier, fresh_nonce, require};
3use crate::link_handshake::MachineIdentity;
4use serde_json::json;
5use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
6use std::sync::Arc;
7
8impl Puncher {
9    /// A new open or recovery gets fresh admitted control streams after an outage.
10    /// # Errors
11    /// The replacement broker did not admit the device's own subscription.
12    pub fn rebind_broker(&mut self, mut broker: super::WireBroker) -> Result<(), PunchError> {
13        broker.subscribe(self.local.machine())?;
14        self.broker = broker;
15        Ok(())
16    }
17    /// A later open observes once on the retained socket; idle never calls this.
18    /// # Errors
19    /// The echo or broker refused the new opening's observation.
20    pub fn prepare_open(&mut self) -> Result<(), PunchError> {
21        if self.observed.is_some() {
22            return Ok(());
23        }
24        let nonce = fresh_nonce()?;
25        let mut request = [0; 20];
26        request[..4].copy_from_slice(b"LMOB");
27        request[4..].copy_from_slice(&nonce);
28        let (answer, source) = self
29            .carrier()?
30            .observe_raw(&request, self.echo, self.bound)?;
31        self.observed = Some(observed_endpoint(&nonce, self.echo, source, &answer)?);
32        self.register()
33    }
34    /// Shares the device's one receive worker with its direct paths.
35    /// # Errors
36    /// The device has not observed its socket.
37    pub fn shared_carrier(&self) -> Result<Arc<UdpCarrier>, PunchError> {
38        self.carrier
39            .clone()
40            .ok_or(PunchError::Named("device_not_observed"))
41    }
42
43    /// Reads the configured peer's registered wire addresses once.
44    /// # Errors
45    /// Broker refusal or malformed lookup answer.
46    pub fn lookup_addresses(&mut self, peer: MachineIdentity) -> Result<Vec<String>, PunchError> {
47        self.validate_peer(peer)?;
48        let value = self.broker.call(
49            "net.broker.lookup",
50            &json!({"device": hex::encode(peer.ed25519_public())}),
51        )?;
52        serde_json::from_value(
53            value
54                .get("addresses")
55                .cloned()
56                .ok_or(PunchError::Named("invalid_broker_answer"))?,
57        )
58        .map_err(|_| PunchError::Named("invalid_broker_answer"))
59    }
60
61    /// Reopens the mapping while retaining the keyed link and all message counters.
62    /// # Errors
63    /// The broker or peer refused the refreshed mapping.
64    pub fn refresh(&mut self, peer: MachineIdentity, index: u32) -> Result<(), PunchError> {
65        self.validate_peer(peer)?;
66        let lookup = self.broker.call(
67            "net.broker.lookup",
68            &json!({"device": hex::encode(peer.ed25519_public())}),
69        )?;
70        let endpoint: SocketAddr = serde_json::from_value(
71            lookup
72                .get("observed")
73                .cloned()
74                .ok_or(PunchError::Named("peer_not_observed"))?,
75        )
76        .map_err(|_| PunchError::Named("peer_not_observed"))?;
77        let nonce = punch_nonce(false)?;
78        self.probe(endpoint)?;
79        self.request_punch(peer, &nonce)?;
80        let word = self.broker.word(peer, Some(&nonce))?;
81        require(word.observed == Some(endpoint), "peer_observation_changed")?;
82        self.carrier()?.refresh_endpoint(index, endpoint)?;
83        Ok(())
84    }
85
86    /// Answers a mapping refresh without a second KK handshake.
87    /// # Errors
88    /// Invalid broker word, probe failure, or unknown keyed link.
89    pub fn accept_refresh(&mut self, peer: MachineIdentity, index: u32) -> Result<(), PunchError> {
90        self.validate_peer(peer)?;
91        let word = self.broker.word(peer, None)?;
92        let endpoint = word
93            .observed
94            .ok_or(PunchError::Named("peer_not_observed"))?;
95        self.probe(endpoint)?;
96        self.request_punch(peer, &word.nonce)?;
97        self.carrier()?.refresh_endpoint(index, endpoint)?;
98        Ok(())
99    }
100    /// Opens the NAT mapping, requests the peer, then follows its return word.
101    /// # Errors
102    /// Refuses unavailable observations, broker failures, invalid words or KK proof.
103    pub fn punch(&mut self, peer: MachineIdentity, index: u32) -> Result<PunchedLink, PunchError> {
104        self.validate_peer(peer)?;
105        require(self.observed.take().is_some(), "device_not_observed")?;
106        let nonce = punch_nonce(true)?;
107        let lookup = self.broker.call(
108            "net.broker.lookup",
109            &json!({"device": hex::encode(peer.ed25519_public())}),
110        )?;
111        let endpoint: SocketAddr = serde_json::from_value(
112            lookup
113                .get("observed")
114                .cloned()
115                .ok_or(PunchError::Named("peer_not_observed"))?,
116        )
117        .map_err(|_| PunchError::Named("peer_not_observed"))?;
118        self.probe(endpoint)?;
119        self.request_punch(peer, &nonce)?;
120        let word = self.broker.word(peer, Some(&nonce))?;
121        require(word.observed == Some(endpoint), "peer_observation_changed")?;
122        self.handshake(peer, endpoint, index)
123    }
124
125    /// Answers the broker's word after sending a mapping-opening probe.
126    /// # Errors
127    /// Refuses unexpected words, broker failures or an unauthenticated KK peer.
128    pub fn accept(&mut self, peer: MachineIdentity, index: u32) -> Result<PunchedLink, PunchError> {
129        self.validate_peer(peer)?;
130        require(self.observed.take().is_some(), "device_not_observed")?;
131        let word = self.broker.word(peer, None)?;
132        let endpoint = word
133            .observed
134            .ok_or(PunchError::Named("peer_not_observed"))?;
135        self.probe(endpoint)?;
136        // The return word proves both mapping-opening sends preceded the handshake.
137        self.request_punch(peer, &word.nonce)?;
138        self.handshake(peer, endpoint, index)
139    }
140
141    /// The shared carrier holding completed direct links.
142    /// # Errors
143    /// Refuses use before a successful observation.
144    pub fn carrier(&self) -> Result<&UdpCarrier, PunchError> {
145        self.carrier
146            .as_deref()
147            .ok_or(PunchError::Named("device_not_observed"))
148    }
149}
150
151pub(super) fn observed_endpoint(
152    nonce: &[u8; 16],
153    expected: SocketAddr,
154    source: SocketAddr,
155    answer: &[u8],
156) -> Result<SocketAddr, PunchError> {
157    let length = answer.len();
158    if source != expected || length < 23 || &answer[..4] != b"LMOA" || answer[4..20] != *nonce {
159        return Err(PunchError::Named("invalid_broker_echo"));
160    }
161    let address = match (answer[20], length) {
162        (4, 27) => SocketAddr::from((
163            Ipv4Addr::new(answer[21], answer[22], answer[23], answer[24]),
164            u16::from_be_bytes([answer[25], answer[26]]),
165        )),
166        (6, 39) => SocketAddr::from((
167            Ipv6Addr::from(
168                <[u8; 16]>::try_from(&answer[21..37])
169                    .map_err(|_| PunchError::Named("invalid_broker_echo"))?,
170            ),
171            u16::from_be_bytes([answer[37], answer[38]]),
172        )),
173        _ => return Err(PunchError::Named("invalid_broker_echo")),
174    };
175    require(address.port() != 0, "invalid_broker_echo")?;
176    Ok(address)
177}
178
179/// A validated broker instruction still requires the recorded peer's KK proof.
180#[derive(Debug)]
181pub struct Instruction {
182    pub peer: MachineIdentity,
183    pub endpoint: SocketAddr,
184    pub nonce: String,
185}
186impl Puncher {
187    /// Shares the existing device socket while admitting a separate incoming subscription.
188    /// # Errors
189    /// Subscription admission or socket setup failed.
190    pub fn fork_listener(
191        &self,
192        local: super::LocalMachine,
193        mut broker: super::WireBroker,
194    ) -> Result<Self, PunchError> {
195        broker.subscribe(local.machine())?;
196        broker.words.set_read_timeout(None)?;
197        Ok(Self {
198            local,
199            socket: Arc::clone(&self.socket),
200            broker,
201            echo: self.echo,
202            bound: self.bound,
203            observed: self.observed,
204            carrier: Some(self.shared_carrier()?),
205            attempted: true,
206            requests: Arc::clone(&self.requests),
207        })
208    }
209    /// Parks on a broker instruction with no idle timer.
210    /// # Errors
211    /// The broker ended or supplied an invalid instruction.
212    pub fn next_instruction(&mut self) -> Result<Instruction, PunchError> {
213        loop {
214            let Some(liminal::protocol::Frame::Deliver {
215                stream_id: 1,
216                envelope,
217                ..
218            }) = crate::link_carrier::read_frame(&mut self.broker.words)?
219            else {
220                return Err(PunchError::Named("punch_word_missing"));
221            };
222            require(
223                Some(envelope.schema_id) == self.broker.schema,
224                "invalid_punch_schema",
225            )?;
226            let word: super::Word = serde_json::from_slice(&envelope.payload)
227                .map_err(|_| PunchError::Named("invalid_punch_word"))?;
228            let mut key = [0; 32];
229            hex::decode_to_slice(&word.from, &mut key)
230                .map_err(|_| PunchError::Named("invalid_punch_peer"))?;
231            require(
232                hex::decode(&word.nonce).is_ok_and(|nonce| nonce.len() == 16),
233                "invalid_punch_nonce",
234            )?;
235            if self
236                .requests
237                .lock()
238                .map_err(|_| PunchError::Named("punch_request_state_poisoned"))?
239                .contains(&word.nonce)
240            {
241                continue;
242            }
243            let peer = MachineIdentity::new(key);
244            self.validate_peer(peer)?;
245            let endpoint = word
246                .observed
247                .ok_or(PunchError::Named("peer_not_observed"))?;
248            require(endpoint.port() != 0, "peer_not_observed")?;
249            return Ok(Instruction {
250                peer,
251                endpoint,
252                nonce: word.nonce,
253            });
254        }
255    }
256    /// Answers one recorded peer instruction using the existing socket and counters.
257    /// # Errors
258    /// Broker refusal, probe failure, or the peer's KK proof failed.
259    pub fn answer_instruction(
260        &mut self,
261        word: &Instruction,
262        index: u32,
263        refresh: bool,
264    ) -> Result<(), PunchError> {
265        self.probe(word.endpoint)?;
266        self.request_punch(word.peer, &word.nonce)?;
267        if refresh {
268            self.carrier()?.refresh_peer(word.peer, word.endpoint)?;
269        } else {
270            self.handshake(word.peer, word.endpoint, index)?;
271        }
272        Ok(())
273    }
274}
275
276fn punch_nonce(opening: bool) -> Result<String, PunchError> {
277    let mut nonce = fresh_nonce()?;
278    // The high bit tells the peer whether to keep its keys or start a new handshake.
279    if opening {
280        nonce[0] |= 128;
281    } else {
282        nonce[0] &= 127;
283    }
284    Ok(hex::encode(nonce))
285}
286impl Instruction {
287    pub(crate) fn opening(&self) -> bool {
288        matches!(
289            self.nonce.as_bytes().first(),
290            Some(b'8'..=b'9' | b'a'..=b'f' | b'A'..=b'F')
291        )
292    }
293}
liminal_server/link_punch/recovery.rs

liminal_server/link_punch/
recovery.rs
