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.local(),
100                self.peer,
101                config,
102                &token,
103                deadline.expires(),
104            )?;
105            Ok(Box::new(path) as Box<dyn Path>)
106        });
107        self.network
108            .pending
109            .lock()
110            .map_err(named)?
111            .remove(&peer_name);
112        result
113    }
114    fn relay(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
115        self.require_config(config)?;
116        self.network.require_running()?;
117        let choice = &self.network.choice;
118        let holder = choice
119            .relay_holder
120            .as_deref()
121            .ok_or_else(|| OpenError("relay_not_configured".to_owned()))?;
122        let relay = self
123            .network
124            .host
125            .recorded_machine(holder)
126            .ok_or_else(|| OpenError("relay_has_no_recorded_machine".to_owned()))?;
127        let address = choice
128            .relay_address
129            .as_deref()
130            .ok_or_else(|| OpenError("relay_address_not_configured".to_owned()))?;
131        let (mut socket, interrupt) = self.network.connect(address)?;
132        write_frame(
133            &mut socket,
134            &Frame::ConversationOpen {
135                flags: 0,
136                stream_id: 1,
137                conversation_id: 1,
138                subject: format!("net.relay.{}", hex::encode(self.peer.ed25519_public())),
139            },
140        )
141        .map_err(named)?;
142        write_frame(&mut socket, &Frame::Ping { flags: 0 }).map_err(named)?;
143        if !matches!(
144            read_frame(&mut socket).map_err(named)?,
145            Some(Frame::Pong { .. })
146        ) {
147            return Err(OpenError("relay_stream_not_marked".to_owned()));
148        }
149        let carrier = RelayCarrier::new(socket, 1, relay).map_err(named)?;
150        let link = crate::link_relay::open(carrier, self.network.host.local(), self.peer)
151            .map_err(named)?;
152        Ok(Box::new(
153            StreamPath::relay(link, hex::encode(relay.ed25519_public())).with_interrupt(interrupt),
154        ))
155    }
156    fn repunch(
157        &mut self,
158        config: &OutboundLinkConfig,
159        path: &mut dyn Path,
160    ) -> Result<(), OpenError> {
161        self.require_config(config)?;
162        let index = path
163            .direct_index()
164            .ok_or_else(|| OpenError("cannot_repunch_a_stream".to_owned()))?;
165        self.network
166            .with_puncher(|puncher| puncher.refresh(self.peer, index).map_err(named))
167    }
168    fn topology(
169        &mut self,
170        carrier: Option<Carrier>,
171        relay: Option<String>,
172        refusal: Option<&str>,
173    ) -> Result<(), OpenError> {
174        use crate::console::records::{Carrier as ObservedCarrier, Link, LinkCounters, LinkState};
175        if refusal.is_some() {
176            self.refused = self
177                .refused
178                .checked_add(1)
179                .ok_or_else(|| OpenError("carrier_refusal_count_exhausted".to_owned()))?;
180        }
181        let peer = hex::encode(self.peer.ed25519_public());
182        let local = self.services.console().device();
183        let carrier = match carrier {
184            Some(Carrier::Wire) => Some(ObservedCarrier::Wire),
185            Some(Carrier::Direct) => Some(ObservedCarrier::Direct),
186            Some(Carrier::Relay) => {
187                let device = relay.or_else(|| {
188                    self.network
189                        .choice
190                        .relay_holder
191                        .as_deref()
192                        .and_then(|holder| self.network.host.recorded_machine(holder))
193                        .map(|machine| hex::encode(machine.ed25519_public()))
194                });
195                if device.is_none() && refusal.is_none() {
196                    return Err(OpenError("relay_topology_has_no_device".to_owned()));
197                }
198                device.map(ObservedCarrier::Relay)
199            }
200            None => None,
201        };
202        self.services
203            .console()
204            .emit(crate::console::TopologyChange::Link {
205                link_id: Link::to_peer(local, &peer).id,
206                peer,
207                carrier,
208                state: if refusal.is_some() {
209                    LinkState::Down
210                } else {
211                    LinkState::Up
212                },
213                reason: refusal.map(str::to_owned),
214                counters: LinkCounters {
215                    refused: Some(self.refused),
216                    ..LinkCounters::default()
217                },
218            })
219            .map_err(named)
220    }
221}
liminal_server/link_open/network/backend.rs

liminal_server/link_open/network/
backend.rs
