liminal_server/services/
broker.rs

1//! A blind broker: address registrations, lookups, punch instructions and a UDP echo.
2
3use super::keys::{ServicePublisher, service_envelope};
4use super::{ServiceEffectError, ServiceRequest};
5use crate::auth_lys::session::LysSession;
6use crate::auth_lys::{AccessDecisionError, HotAction, SubjectBinding};
7use crate::config::types::LysConfig;
8use crate::link_handshake::MachineIdentity;
9use serde::{Deserialize, Serialize};
10use serde_json::{Value, json};
11use std::collections::BTreeMap;
12use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket};
13use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
14use std::sync::{Arc, Mutex};
15use std::thread::JoinHandle;
16
17/// A control call refused before its effect, or a failed punch publication.
18#[derive(Debug, thiserror::Error)]
19pub enum BrokerRefusal {
20    /// The caller's pass did not grant this call.
21    #[error("{0}")]
22    Decision(#[from] AccessDecisionError),
23    /// A named boundary or state refusal.
24    #[error("{0}")]
25    Named(&'static str),
26    /// The publisher refused the instruction.
27    #[error("{0}")]
28    Publish(#[from] ServiceEffectError),
29}
30
31/// Traffic counters; there is deliberately no mutable data-byte counter.
32#[derive(Debug, Default)]
33pub struct BrokerCounters {
34    registered: AtomicU64,
35    lookups: AtomicU64,
36    punches_told: AtomicU64,
37    echo_answered: AtomicU64,
38    echo_dropped: AtomicU64,
39}
40
41/// Successful registrations and instructions, lookup attempts and echo traffic.
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
43pub struct CounterSnapshot {
44    /// Successful registrations, including replacements.
45    pub registered: u64,
46    /// Authorized, well-formed lookup attempts.
47    pub lookups: u64,
48    /// Instructions accepted by the publisher.
49    pub punches_told: u64,
50    /// Echo replies sent.
51    pub echo_answered: u64,
52    /// Datagrams rejected by the echo protocol.
53    pub echo_dropped: u64,
54    /// The broker has no data-forwarding operation, so this is always zero.
55    pub data_bytes_forwarded: u64,
56}
57
58impl BrokerCounters {
59    /// Reads each traffic counter; data bytes are pinned to zero.
60    #[must_use]
61    pub fn snapshot(&self) -> CounterSnapshot {
62        CounterSnapshot {
63            registered: self.registered.load(Ordering::Relaxed),
64            lookups: self.lookups.load(Ordering::Relaxed),
65            punches_told: self.punches_told.load(Ordering::Relaxed),
66            echo_answered: self.echo_answered.load(Ordering::Relaxed),
67            echo_dropped: self.echo_dropped.load(Ordering::Relaxed),
68            data_bytes_forwarded: 0,
69        }
70    }
71}
72
73#[derive(Clone, Serialize, Deserialize)]
74#[serde(deny_unknown_fields)]
75struct Registration {
76    device: String,
77    addresses: Vec<String>,
78    #[serde(default)]
79    observed: Option<SocketAddr>,
80}
81
82#[derive(Deserialize)]
83#[serde(deny_unknown_fields)]
84struct Lookup {
85    device: String,
86}
87
88#[derive(Deserialize)]
89#[serde(deny_unknown_fields)]
90struct Punch {
91    to: String,
92    nonce: String,
93}
94
95/// In-memory control state, resolved from the caller's verified holder.
96pub struct Broker {
97    network: String,
98    recorded: BTreeMap<String, MachineIdentity>,
99    registrations: Mutex<BTreeMap<[u8; 32], Registration>>,
100    publisher: Arc<dyn ServicePublisher>,
101    counters: Arc<BrokerCounters>,
102}
103
104impl std::fmt::Debug for Broker {
105    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        formatter
107            .debug_struct("Broker")
108            .field("network", &self.network)
109            .finish_non_exhaustive()
110    }
111}
112
113impl Broker {
114    /// Uses the same holder-to-machine directory as the link host.
115    #[must_use]
116    pub fn new(
117        network: String,
118        recorded: BTreeMap<String, MachineIdentity>,
119        publisher: Arc<dyn ServicePublisher>,
120    ) -> Self {
121        Self {
122            network,
123            recorded,
124            registrations: Mutex::new(BTreeMap::new()),
125            publisher,
126            counters: Arc::new(BrokerCounters::default()),
127        }
128    }
129
130    /// The counters shared with this broker's echo worker.
131    #[must_use]
132    pub fn counters(&self) -> Arc<BrokerCounters> {
133        Arc::clone(&self.counters)
134    }
135
136    /// Decides a control call offline before parsing or changing its state.
137    ///
138    /// # Errors
139    /// Missing call grant, unknown holder, malformed input, absent registration,
140    /// poisoned state or a refused publication; each retains its name.
141    pub fn serve(
142        &self,
143        caller: &dyn LysSession,
144        subject: &str,
145        request: &ServiceRequest<'_>,
146        now: u64,
147    ) -> Result<Value, BrokerRefusal> {
148        if !serves_subject(subject) {
149            return Err(BrokerRefusal::Named("not_a_broker_subject"));
150        }
151        let binding =
152            SubjectBinding::new(self.network.clone(), subject.to_owned(), subject.to_owned())
153                .map_err(|_| BrokerRefusal::Named("invalid_subject"))?;
154        caller.decide(&binding, HotAction::Call, now)?;
155        if request.operation.trim().is_empty() || request.operation.len() > 256 {
156            return Err(BrokerRefusal::Named("invalid_operation"));
157        }
158        if request.payload.len() > 16_384 {
159            return Err(BrokerRefusal::Named("request_too_large"));
160        }
161        let device = self
162            .recorded
163            .get(caller.holder())
164            .ok_or(BrokerRefusal::Named("machine_not_recorded"))?;
165        match subject {
166            "net.broker.register" => self.register(device, request.payload),
167            "net.broker.lookup" => self.lookup(request.payload),
168            "net.broker.punch" => self.punch(device, caller.holder(), request),
169            _ => Err(BrokerRefusal::Named("not_a_broker_subject")),
170        }
171    }
172
173    fn register(&self, caller: &MachineIdentity, payload: &[u8]) -> Result<Value, BrokerRefusal> {
174        let mut registration: Registration = parse(payload)?;
175        let device = device_key(&registration.device)?;
176        if &device != caller.ed25519_public() {
177            return Err(BrokerRefusal::Named("device_mismatch"));
178        }
179        if registration.addresses.len() > 64
180            || registration
181                .addresses
182                .iter()
183                .any(|address| !valid_address(address))
184            || registration
185                .observed
186                .is_some_and(|address| address.port() == 0)
187        {
188            return Err(BrokerRefusal::Named("invalid_address"));
189        }
190        registration.device = hex::encode(device);
191        self.registrations
192            .lock()
193            .map_err(|_| BrokerRefusal::Named("registry_poisoned"))?
194            .insert(device, registration);
195        self.counters.registered.fetch_add(1, Ordering::Relaxed);
196        Ok(json!({"ok": true}))
197    }
198
199    fn lookup(&self, payload: &[u8]) -> Result<Value, BrokerRefusal> {
200        let request: Lookup = parse(payload)?;
201        let device = device_key(&request.device)?;
202        self.counters.lookups.fetch_add(1, Ordering::Relaxed);
203        let registration = {
204            let registrations = self
205                .registrations
206                .lock()
207                .map_err(|_| BrokerRefusal::Named("registry_poisoned"))?;
208            registrations
209                .get(&device)
210                .cloned()
211                .ok_or(BrokerRefusal::Named("not_registered"))?
212        };
213        Ok(json!({"addresses": registration.addresses, "observed": registration.observed}))
214    }
215
216    fn punch(
217        &self,
218        caller: &MachineIdentity,
219        holder: &str,
220        request: &ServiceRequest<'_>,
221    ) -> Result<Value, BrokerRefusal> {
222        let punch: Punch = parse(request.payload)?;
223        let to = device_key(&punch.to)?;
224        let nonce = hex::decode(&punch.nonce).map_err(|_| BrokerRefusal::Named("invalid_nonce"))?;
225        if nonce.len() != 16 {
226            return Err(BrokerRefusal::Named("invalid_nonce"));
227        }
228        let observed = {
229            let registrations = self
230                .registrations
231                .lock()
232                .map_err(|_| BrokerRefusal::Named("registry_poisoned"))?;
233            registrations
234                .get(caller.ed25519_public())
235                .ok_or(BrokerRefusal::Named("not_registered"))?
236                .observed
237        };
238        let payload = json!({"from": hex::encode(caller.ed25519_public()),
239            "observed": observed, "nonce": hex::encode(nonce)})
240        .to_string();
241        // A caller's operation cannot suppress another holder's instruction.
242        let operation = blake3::hash(
243            &serde_json::to_vec(&(holder, request.operation, to, &payload))
244                .map_err(|_| BrokerRefusal::Named("instruction_encoding_failed"))?,
245        )
246        .to_hex();
247        self.publisher.publish(
248            &format!("net.punch.{}", hex::encode(to)),
249            &service_envelope(b"liminal.net.punch/1", payload.as_bytes()),
250            operation.as_str(),
251        )?;
252        self.counters.punches_told.fetch_add(1, Ordering::Relaxed);
253        Ok(json!({"ok": true}))
254    }
255}
256
257/// The three specified control subjects; arbitrary payload subjects are refused.
258#[must_use]
259pub fn serves_subject(subject: &str) -> bool {
260    matches!(
261        subject,
262        "net.broker.register" | "net.broker.lookup" | "net.broker.punch"
263    )
264}
265
266fn parse<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T, BrokerRefusal> {
267    serde_json::from_slice(bytes).map_err(|_| BrokerRefusal::Named("invalid_request"))
268}
269
270fn device_key(value: &str) -> Result<[u8; 32], BrokerRefusal> {
271    let mut bytes = [0; 32];
272    hex::decode_to_slice(value, &mut bytes).map_err(|_| BrokerRefusal::Named("invalid_device"))?;
273    MachineIdentity::new(bytes)
274        .x25519_public()
275        .map_err(|_| BrokerRefusal::Named("invalid_device"))?;
276    Ok(bytes)
277}
278
279fn valid_address(address: &str) -> bool {
280    if address
281        .parse::<SocketAddr>()
282        .is_ok_and(|address| address.port() != 0)
283    {
284        return true;
285    }
286    let Some((host, port)) = address.rsplit_once(':') else {
287        return false;
288    };
289    host.len() <= 253
290        && !host.is_empty()
291        && host.split('.').all(|label| {
292            !label.is_empty()
293                && label.len() <= 63
294                && !label.starts_with('-')
295                && !label.ends_with('-')
296                && label
297                    .bytes()
298                    .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
299        })
300        && port.parse::<u16>().is_ok_and(|port| port != 0)
301}
302
303/// Adds broker-specific deterministic configuration refusals.
304pub fn validate_configuration(lys: &LysConfig, errors: &mut Vec<String>) {
305    let Some(broker) = &lys.broker else {
306        return;
307    };
308    if lys.machine.is_none() {
309        errors
310            .push("auth.lys.broker: requires auth.lys.machine for recorded caller keys".to_owned());
311    }
312    if broker.udp_listen.port() == 0 {
313        errors.push("auth.lys.broker.udp_listen: port must be nonzero".to_owned());
314    }
315}
316
317/// One blocking echo worker, stopped by a datagram signal and joined by its owner.
318#[derive(Debug)]
319pub struct ObservedEcho {
320    socket: UdpSocket,
321    address: SocketAddr,
322    stopping: Arc<AtomicBool>,
323    worker: Option<JoinHandle<std::io::Result<()>>>,
324}
325
326impl ObservedEcho {
327    /// Binds before starting the sole blocking worker; no refresh or idle timer.
328    ///
329    /// # Errors
330    /// The socket could not bind or clone, or the worker could not start.
331    pub fn bind(address: SocketAddr, counters: Arc<BrokerCounters>) -> std::io::Result<Self> {
332        let socket = UdpSocket::bind(address)?;
333        let address = socket.local_addr()?;
334        let receive = socket.try_clone()?;
335        let stopping = Arc::new(AtomicBool::new(false));
336        let stop = Arc::clone(&stopping);
337        let worker = std::thread::Builder::new()
338            .name("liminal-broker-echo".to_owned())
339            .spawn(move || {
340                let result = echo_loop(&receive, &counters, &stop);
341                if let Err(error) = &result {
342                    tracing::error!(%error, "broker echo worker failed");
343                }
344                result
345            })?;
346        Ok(Self {
347            socket,
348            address,
349            stopping,
350            worker: Some(worker),
351        })
352    }
353
354    /// The address actually bound, including an allocated ephemeral test port.
355    #[must_use]
356    pub const fn local_addr(&self) -> SocketAddr {
357        self.address
358    }
359
360    /// Signals the blocking receive, then joins and reports the worker's result.
361    ///
362    /// # Errors
363    /// The shutdown signal could not send, the worker failed, or it panicked.
364    pub fn shutdown(&mut self) -> std::io::Result<()> {
365        if self.worker.is_none() {
366            return Ok(());
367        }
368        self.stopping.store(true, Ordering::Release);
369        let mut wake = self.address;
370        if wake.ip().is_unspecified() {
371            wake.set_ip(match wake {
372                SocketAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST),
373                SocketAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST),
374            });
375        }
376        self.socket.send_to(&[], wake)?;
377        if let Some(worker) = self.worker.take() {
378            worker
379                .join()
380                .map_err(|_| std::io::Error::other("broker echo worker panicked"))??;
381        }
382        Ok(())
383    }
384}
385
386impl Drop for ObservedEcho {
387    fn drop(&mut self) {
388        if let Err(error) = self.shutdown() {
389            tracing::error!(%error, "broker echo shutdown failed");
390        }
391    }
392}
393
394fn echo_loop(
395    socket: &UdpSocket,
396    counters: &BrokerCounters,
397    stopping: &AtomicBool,
398) -> std::io::Result<()> {
399    let mut request = [0; 21];
400    loop {
401        let (length, source) = match socket.recv_from(&mut request) {
402            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
403            result => result?,
404        };
405        if stopping.load(Ordering::Acquire) {
406            return Ok(());
407        }
408        if length != 20 || &request[..4] != b"LMOB" {
409            counters.echo_dropped.fetch_add(1, Ordering::Relaxed);
410            continue;
411        }
412        let mut reply = Vec::with_capacity(39);
413        reply.extend_from_slice(b"LMOA");
414        reply.extend_from_slice(&request[4..20]);
415        match source {
416            SocketAddr::V4(address) => {
417                reply.push(4);
418                reply.extend_from_slice(&address.ip().octets());
419            }
420            SocketAddr::V6(address) => {
421                reply.push(6);
422                reply.extend_from_slice(&address.ip().octets());
423            }
424        }
425        reply.extend_from_slice(&source.port().to_be_bytes());
426        if socket.send_to(&reply, source)? != reply.len() {
427            return Err(std::io::Error::other(
428                "broker echo sent an incomplete datagram",
429            ));
430        }
431        counters.echo_answered.fetch_add(1, Ordering::Relaxed);
432    }
433}
liminal_server/services/broker.rs

liminal_server/services/
broker.rs
