liminal_server/link_open/network/
backend.rs

1//! Each configured link binds its peer before trying a carrier.
2use super::NetworkBackend;
3use crate::config::types::OutboundLinkConfig;
4use crate::link_carrier::{Carrier, RelayCarrier, read_frame, write_frame};
5use crate::link_dial::{LinkDialOutcome, LinkDialer};
6use crate::link_open::carriers::{DirectPath, StreamPath, named};
7use crate::link_open::{Backend, OpenError, Path, Stop};
8use liminal::protocol::Frame;
9use std::sync::Arc;
10impl Backend for NetworkBackend {
11    fn interrupt_handle(&self) -> Option<Stop> {
12        let network = Arc::clone(&self.network);
13        Some(Arc::new(move || network.stop()))
14    }
15    fn wire(
16        &mut self,
17        config: &OutboundLinkConfig,
18        address: &str,
19    ) -> Result<Box<dyn Path>, OpenError> {
20        self.require_config(config)?;
21        self.network.require_running()?;
22        let mut entry = config.clone();
23        entry.peer_address = Some(address.to_owned());
24        match LinkDialer::new(&self.network.host, &self.network.machine)
25            .with_network(&self.network)
26            .open(&entry)
27            .map_err(named)?
28        {
29            LinkDialOutcome::Opened(link) => {
30                let interrupt = self
31                    .network
32                    .retain(link.carrier.interrupt_handle().map_err(named)?)?;
33                Ok(Box::new(StreamPath::wire(*link).with_interrupt(interrupt)))
34            }
35            LinkDialOutcome::Held { grant, draft } => {
36                Err(OpenError(format!("link_held:{grant}:{draft}")))
37            }
38            LinkDialOutcome::Refused { reason } => Err(OpenError(format!("link_refused:{reason}"))),
39        }
40    }
41    fn lookup(&mut self, config: &OutboundLinkConfig) -> Result<Vec<String>, OpenError> {
42        self.require_config(config)?;
43        self.network.require_running()?;
44        let (mut broker, connections) = self.network.broker()?;
45        let result = broker
46            .call(
47                "net.broker.lookup",
48                &serde_json::json!({"device": hex::encode(self.peer.ed25519_public())}),
49            )
50            .map_err(named);
51        drop(connections);
52        let value = result?;
53        let addresses: Vec<String> = serde_json::from_value(
54            value
55                .get("addresses")
56                .cloned()
57                .ok_or_else(|| OpenError("invalid_broker_answer".to_owned()))?,
58        )
59        .map_err(|_| OpenError("invalid_broker_answer".to_owned()))?;
60        if addresses.len() > 64
61            || addresses
62                .iter()
63                .any(|address| address.is_empty() || address.len() > 2048)
64        {
65            return Err(OpenError("invalid_broker_addresses".to_owned()));
66        }
67        Ok(addresses)
68    }
69    fn punch(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
70        self.require_config(config)?;
71        let index = {
72            let mut direct = self.network.direct.lock().map_err(named)?;
73            let index = direct.next;
74            direct.next = index
75                .checked_add(1)
76                .ok_or_else(|| OpenError("direct_indices_exhausted".to_owned()))?;
77            index
78        };
79        let peer_name = hex::encode(self.peer.ed25519_public());
80        self.network
81            .pending
82            .lock()
83            .map_err(named)?
84            .insert(peer_name.clone());
85        let result = self.network.with_puncher(|puncher| {
86            puncher.prepare_open().map_err(named)?;
87            let handshake_guard = self.network.handshakes.lock().map_err(named)?;
88            let link = puncher.punch(self.peer, index).map_err(named)?;
89            drop(handshake_guard);
90            let mut path =
91                DirectPath::new(puncher.shared_carrier().map_err(named)?, link.local_index)?;
92            let pass =
93                crate::link_pass::MachinePass::configured(&self.network.machine).map_err(named)?;
94            let token = pass.for_open().map_err(named)?;
95            let deadline = crate::link_open::Deadline::after(std::time::Duration::from_millis(
96                self.network.choice.crossing_deadline_ms,
97            ))?;
98            path.admit(
99                &self.network.host,
100                &self.network.machine.agent_id,
101                self.peer,
102                config,
103                &token,
104                deadline.expires(),
105            )?;
106            Ok(Box::new(path) as Box<dyn Path>)
107        });
108        self.network
109            .pending
110            .lock()
111            .map_err(named)?
112            .remove(&peer_name);
113        result
114    }
115    fn relay(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
116        self.require_config(config)?;
117        self.network.require_running()?;
118        let choice = &self.network.choice;
119        let holder = choice
120            .relay_holder
121            .as_deref()
122            .ok_or_else(|| OpenError("relay_not_configured".to_owned()))?;
123        let relay = self
124            .network
125            .host
126            .recorded_machine(holder)
127            .ok_or_else(|| OpenError("relay_has_no_recorded_machine".to_owned()))?;
128        let address = choice
129            .relay_address
130            .as_deref()
131            .ok_or_else(|| OpenError("relay_address_not_configured".to_owned()))?;
132        let (mut socket, interrupt) = self.network.connect(address)?;
133        write_frame(
134            &mut socket,
135            &Frame::ConversationOpen {
136                flags: 0,
137                stream_id: 1,
138                conversation_id: 1,
139                subject: format!("net.relay.{}", hex::encode(self.peer.ed25519_public())),
140            },
141        )
142        .map_err(named)?;
143        write_frame(&mut socket, &Frame::Ping { flags: 0 }).map_err(named)?;
144        if !matches!(
145            read_frame(&mut socket).map_err(named)?,
146            Some(Frame::Pong { .. })
147        ) {
148            return Err(OpenError("relay_stream_not_marked".to_owned()));
149        }
150        let carrier = RelayCarrier::new(socket, 1, relay).map_err(named)?;
151        let link = crate::link_relay::open(carrier, self.network.host.local(), self.peer)
152            .map_err(named)?;
153        Ok(Box::new(
154            StreamPath::relay(link, hex::encode(relay.ed25519_public())).with_interrupt(interrupt),
155        ))
156    }
157    fn repunch(
158        &mut self,
159        config: &OutboundLinkConfig,
160        path: &mut dyn Path,
161    ) -> Result<(), OpenError> {
162        self.require_config(config)?;
163        let index = path
164            .direct_index()
165            .ok_or_else(|| OpenError("cannot_repunch_a_stream".to_owned()))?;
166        self.network
167            .with_puncher(|puncher| puncher.refresh(self.peer, index).map_err(named))
168    }
169    fn topology(
170        &mut self,
171        carrier: Option<Carrier>,
172        relay: Option<String>,
173        refusal: Option<&str>,
174    ) -> Result<(), OpenError> {
175        use crate::console::records::{Carrier as ObservedCarrier, Link, LinkCounters, LinkState};
176        if refusal.is_some() {
177            self.refused = self
178                .refused
179                .checked_add(1)
180                .ok_or_else(|| OpenError("carrier_refusal_count_exhausted".to_owned()))?;
181        }
182        let peer = hex::encode(self.peer.ed25519_public());
183        let local = self.services.console().device();
184        let carrier = match carrier {
185            Some(Carrier::Wire) => Some(ObservedCarrier::Wire),
186            Some(Carrier::Direct) => Some(ObservedCarrier::Direct),
187            Some(Carrier::Relay) => {
188                let device = relay.or_else(|| {
189                    self.network
190                        .choice
191                        .relay_holder
192                        .as_deref()
193                        .and_then(|holder| self.network.host.recorded_machine(holder))
194                        .map(|machine| hex::encode(machine.ed25519_public()))
195                });
196                if device.is_none() && refusal.is_none() {
197                    return Err(OpenError("relay_topology_has_no_device".to_owned()));
198                }
199                device.map(ObservedCarrier::Relay)
200            }
201            None => None,
202        };
203        self.services
204            .console()
205            .emit(crate::console::TopologyChange::Link {
206                link_id: Link::to_peer(local, &peer).id,
207                peer,
208                carrier,
209                state: if refusal.is_some() {
210                    LinkState::Down
211                } else {
212                    LinkState::Up
213                },
214                reason: refusal.map(str::to_owned),
215                counters: LinkCounters {
216                    refused: Some(self.refused),
217                    ..LinkCounters::default()
218                },
219            })
220            .map_err(named)
221    }
222}
liminal_server/link_open/network/backend.rs

liminal_server/link_open/network/
backend.rs
