liminal_server/services/
production.rs

1//! The production bindings of the service seams (ACCESS-003 R1), built once
2//! from `[auth.lys.services]` and `[auth.lys.broker]`.
3//!
4//! - [`SeatStarter`](super::SeatStarter): [`LysSeatStarter`], a seat started
5//!   by Lys's start-command route on the configured machine.
6//! - [`KeySender`](super::KeySender): [`ChannelKeySender`], the keys
7//!   published on `sessions.<id>.keys` through the server's own publish path.
8//! - [`BudgetSource`](super::BudgetSource): [`LysBudgets`], budgets and stops
9//!   read from Lys, carried whole.
10//! - [`ServiceAction`](super::ServiceAction): [`PublishingActions`], alarms
11//!   and launches published on their subjects, context readings reported to
12//!   Lys through the [`ContextReporter`] the context watch shares.
13//!
14//! Every Lys request is made as this liminal, a machine Lys knows: the
15//! bearer is the current machine pass held by the boot pass source,
16//! verified offline at start exactly as any presented pass is. Without
17//! `[auth.lys.machine]` the services cannot ask Lys as anyone, and start is
18//! refused by that name, never served with a guessed identity.
19
20use super::actions::PublishingActions;
21use super::broker::{Broker, BrokerRefusal, ObservedEcho};
22use super::budgets::LysBudgets;
23use super::context_watch::ContextReporter;
24use super::keys::{ChannelKeySender, ConnectionPublisher, ServicePublisher};
25use super::lys_http::{HttpLys, LysAnswer, LysHttp};
26use super::seats::LysSeatStarter;
27use super::{
28    BudgetPolicy, BudgetSource, KeySender, SeatStarter, ServiceAction, ServiceEffectError,
29    ServiceKind, ServiceRequest, Services,
30};
31use crate::auth_lys::LysPassConsumer;
32use crate::config::types::{LysConfig, MachineConfig};
33use crate::link_pass::MachinePass;
34use crate::server::connection::LiminalConnectionServices;
35use std::net::SocketAddr;
36use std::sync::Arc;
37mod debug;
38
39/// Administrative effect seams and an optional address broker, bound once.
40pub struct ServiceBindings {
41    administrative: Option<AdministrativeBindings>,
42    context: Arc<ContextReporter>,
43    watch_context: bool,
44    broker: Option<Broker>,
45    echo: Option<ObservedEcho>,
46    relay: Option<Arc<super::relay::Relay>>,
47    relay_topology_local: Option<String>,
48}
49
50#[derive(Debug)]
51struct AdministrativeBindings {
52    seats: LysSeatStarter,
53    keys: ChannelKeySender,
54    budgets: LysBudgets,
55    actions: PublishingActions,
56}
57
58impl std::fmt::Debug for ServiceBindings {
59    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
60        formatter
61            .debug_struct("ServiceBindings")
62            .field("administrative", &self.administrative)
63            .field("watch_context", &self.watch_context)
64            .field("broker", &self.broker)
65            .field("echo", &self.echo)
66            .field("relay", &self.relay)
67            .finish_non_exhaustive()
68    }
69}
70
71impl ServiceBindings {
72    /// Binds every seam: Lys asked through `http` with `bearer` (this
73    /// machine's own pass), seats started on `machine`, messages published
74    /// through `publisher`.
75    #[must_use]
76    pub fn new(
77        http: &Arc<dyn LysHttp>,
78        bearer: &str,
79        machine: String,
80        publisher: &Arc<dyn ServicePublisher>,
81    ) -> Self {
82        let context = Arc::new(ContextReporter::new(Arc::clone(http), bearer.to_owned()));
83        Self {
84            administrative: Some(AdministrativeBindings {
85                seats: LysSeatStarter::new(Arc::clone(http), bearer.to_owned(), machine),
86                keys: ChannelKeySender::new(Arc::clone(publisher)),
87                budgets: LysBudgets::new(Arc::clone(http), bearer.to_owned()),
88                actions: PublishingActions::new(Arc::clone(publisher), Arc::clone(&context)),
89            }),
90            context,
91            watch_context: false,
92            broker: None,
93            echo: None,
94            relay: None,
95            relay_topology_local: None,
96        }
97    }
98
99    /// Whether the operator asked for the context watch to run.
100    #[must_use]
101    pub const fn watches_context(&self) -> bool {
102        self.watch_context
103    }
104
105    /// The seams, lent for one [`serve`](super::serve).
106    #[must_use]
107    pub fn services(&self) -> Services<'_> {
108        Services {
109            seats: self,
110            keys: self,
111            budgets: self,
112            actions: self,
113        }
114    }
115
116    /// The reporter the context watch shares with the `context-watch` call.
117    #[must_use]
118    pub fn context_reporter(&self) -> Arc<ContextReporter> {
119        Arc::clone(&self.context)
120    }
121
122    /// The configured address broker, independent of administrative services.
123    #[must_use]
124    pub const fn broker(&self) -> Option<&Broker> {
125        self.broker.as_ref()
126    }
127
128    /// The address actually bound by the configured broker's UDP echo.
129    #[must_use]
130    pub fn echo_address(&self) -> Option<SocketAddr> {
131        self.echo.as_ref().map(ObservedEcho::local_addr)
132    }
133
134    /// The configured blind relay, independent of administrative services.
135    #[must_use]
136    pub const fn relay(&self) -> Option<&Arc<super::relay::Relay>> {
137        self.relay.as_ref()
138    }
139
140    /// The console's canonical local endpoint for relay observations.
141    #[must_use]
142    pub fn relay_topology_local(&self) -> Option<&str> {
143        self.relay_topology_local.as_deref()
144    }
145
146    /// Serves a broker request on the connection's admitted session.
147    ///
148    /// # Errors
149    /// The broker or admitted session is absent, or the broker refuses the call.
150    pub fn serve_broker(
151        &self,
152        session: Option<&dyn crate::auth_lys::session::LysSession>,
153        subject: &str,
154        request: &ServiceRequest<'_>,
155        now: u64,
156    ) -> Result<serde_json::Value, BrokerRefusal> {
157        let broker = self
158            .broker
159            .as_ref()
160            .ok_or(BrokerRefusal::Named("broker_not_configured"))?;
161        let session = session.ok_or(BrokerRefusal::Named("pass_not_admitted"))?;
162        broker.serve(session, subject, request, now)
163    }
164
165    /// Builds configured services when the boot machine pass is present.
166    /// `Ok(None)` when no services are selected or a dial pass was refused.
167    /// Lys is asked on `handle`'s runtime; messages are published through
168    /// `services` as `liminal.<host>`.
169    ///
170    /// # Errors
171    /// `[auth.lys.machine]` is absent (the services would have no identity to
172    /// ask Lys as), this machine's pass or the issuer's key set cannot be
173    /// read or does not verify, or the issuer is not an HTTP origin.
174    pub fn configure(
175        lys: &LysConfig,
176        services: Arc<LiminalConnectionServices>,
177        handle: tokio::runtime::Handle,
178        now: u64,
179    ) -> Result<Option<Self>, ServiceEffectError> {
180        if lys.services.is_none()
181            && lys.broker.is_none()
182            && !lys.relay.as_ref().is_some_and(|relay| relay.enabled)
183        {
184            return Ok(None);
185        }
186        let machine = lys
187            .machine
188            .as_ref()
189            .ok_or_else(|| ServiceEffectError("auth.lys.machine is absent".into()))?;
190        let Some(pass) = services_pass(machine)? else {
191            return Ok(None);
192        };
193        let bearer = machine_bearer(lys, now)?;
194        let http: Arc<dyn LysHttp> = Arc::new(MachineHttp {
195            http: HttpLys::new(lys.api_url.as_deref().unwrap_or(&lys.issuer), handle)?,
196            pass,
197        });
198        let host = lys
199            .services
200            .as_ref()
201            .map_or("broker", |section| section.host.as_str());
202        let topology_local = services.console().device().to_owned();
203        let publisher: Arc<dyn ServicePublisher> =
204            Arc::new(ConnectionPublisher::new(services, host.to_owned()));
205        let mut bindings = if let Some(section) = &lys.services {
206            let mut bindings = Self::new(&http, &bearer, section.machine.clone(), &publisher);
207            bindings.watch_context = section.context_watch;
208            bindings
209        } else {
210            Self {
211                administrative: None,
212                context: Arc::new(ContextReporter::new(Arc::clone(&http), bearer)),
213                watch_context: false,
214                broker: None,
215                echo: None,
216                relay: None,
217                relay_topology_local: None,
218            }
219        };
220        if let Some(section) = &lys.broker {
221            let machine = lys.machine.as_ref().ok_or_else(|| {
222                ServiceEffectError("auth.lys.broker requires auth.lys.machine".to_owned())
223            })?;
224            let recorded = machine
225                .machines
226                .iter()
227                .map(|(holder, key)| {
228                    let mut bytes = [0; 32];
229                    hex::decode_to_slice(key, &mut bytes).map_err(|_| {
230                        ServiceEffectError(format!(
231                            "auth.lys.machine.machines: invalid key for holder {holder}"
232                        ))
233                    })?;
234                    let identity = crate::link_handshake::MachineIdentity::new(bytes);
235                    identity.x25519_public().map_err(|error| {
236                        ServiceEffectError(format!(
237                            "auth.lys.machine.machines: holder {holder}: {error}"
238                        ))
239                    })?;
240                    Ok((holder.clone(), identity))
241                })
242                .collect::<Result<_, ServiceEffectError>>()?;
243            let broker = Broker::new(lys.network_id.clone(), recorded, publisher);
244            bindings.echo = Some(
245                ObservedEcho::bind(section.udp_listen, broker.counters()).map_err(|error| {
246                    ServiceEffectError(format!("auth.lys.broker.udp_listen: {error}"))
247                })?,
248            );
249            bindings.broker = Some(broker);
250        }
251        if lys.relay.as_ref().is_some_and(|relay| relay.enabled) {
252            let machine = lys.machine.as_ref().ok_or_else(|| {
253                ServiceEffectError("auth.lys.relay requires auth.lys.machine".to_owned())
254            })?;
255            let recorded = machine
256                .machines
257                .iter()
258                .map(|(holder, key)| {
259                    let mut bytes = [0; 32];
260                    hex::decode_to_slice(key, &mut bytes).map_err(|_| {
261                        ServiceEffectError(format!("invalid recorded machine for holder {holder}"))
262                    })?;
263                    let identity = crate::link_handshake::MachineIdentity::new(bytes);
264                    identity
265                        .x25519_public()
266                        .map_err(|error| ServiceEffectError(error.to_string()))?;
267                    Ok((holder.clone(), identity))
268                })
269                .collect::<Result<_, ServiceEffectError>>()?;
270            bindings.relay = Some(super::relay::Relay::new(lys.network_id.clone(), recorded));
271            bindings.relay_topology_local = Some(topology_local);
272        }
273        Ok(Some(bindings))
274    }
275
276    fn administrative(&self) -> Result<&AdministrativeBindings, ServiceEffectError> {
277        self.administrative
278            .as_ref()
279            .ok_or_else(|| ServiceEffectError("auth.lys.services is not configured".to_owned()))
280    }
281}
282
283impl SeatStarter for ServiceBindings {
284    fn start(&self, seat: &str, operation: &str) -> Result<(), ServiceEffectError> {
285        self.administrative()?.seats.start(seat, operation)
286    }
287}
288
289impl KeySender for ServiceBindings {
290    fn send(&self, session: &str, keys: &[u8], operation: &str) -> Result<(), ServiceEffectError> {
291        self.administrative()?.keys.send(session, keys, operation)
292    }
293}
294
295impl BudgetSource for ServiceBindings {
296    fn read(&self, name: &str) -> Result<Option<BudgetPolicy>, ServiceEffectError> {
297        self.administrative()?.budgets.read(name)
298    }
299}
300
301impl ServiceAction for ServiceBindings {
302    fn act(
303        &self,
304        service: ServiceKind,
305        payload: &[u8],
306        operation: &str,
307    ) -> Result<(), ServiceEffectError> {
308        self.administrative()?
309            .actions
310            .act(service, payload, operation)
311    }
312}
313
314/// The pass the services present, or `None` when a dial machine holds none
315/// yet: its ask was refused at start (ACCESS-009 R1, the link asks again at
316/// its next open) or no source is configured. Services bind at start only.
317///
318/// # Errors
319/// The pass holder is poisoned, or the pass cannot be read for a reason
320/// other than an absent source.
321fn services_pass(machine: &MachineConfig) -> Result<Option<MachinePass>, ServiceEffectError> {
322    match MachinePass::configured(machine) {
323        Ok(pass) => {
324            let issued = pass
325                .issued()
326                .map_err(|error| ServiceEffectError(error.to_string()))?;
327            Ok((issued || machine.agent_id.is_empty()).then_some(pass))
328        }
329        Err(error) if !machine.agent_id.is_empty() && error.source_absent() => Ok(None),
330        Err(error) => Err(ServiceEffectError(error.to_string())),
331    }
332}
333
334/// This machine's own Lys pass as the bearer the services ask Lys with.
335///
336/// It is verified offline for audience `liminal` at `now` against the
337/// issuer's key set, so an expired or foreign pass refuses start by its
338/// cause.
339///
340/// # Errors
341/// No `[auth.lys.machine]`, or the pass or key set is unreadable or refused.
342pub fn machine_bearer(lys: &LysConfig, now: u64) -> Result<String, ServiceEffectError> {
343    let machine = lys.machine.as_ref().ok_or_else(|| {
344        ServiceEffectError(
345            "auth.lys.services or auth.lys.broker needs auth.lys.machine: services use this machine's pass"
346                .to_owned(),
347        )
348    })?;
349    let keys_json = std::fs::read_to_string(&lys.keys_path).map_err(|error| {
350        ServiceEffectError(format!(
351            "auth.lys.keys_path {}: {error}",
352            lys.keys_path.display()
353        ))
354    })?;
355    let adapter = crate::auth_lys::lys_pass_adapter::LysPassAdapter::from_keys_json(
356        &keys_json,
357        lys.issuer.clone(),
358    )
359    .map_err(|error| ServiceEffectError(format!("auth.lys key set refused: {error}")))?;
360    let pass = crate::link_pass::MachinePass::configured(machine)
361        .map_err(|error| ServiceEffectError(error.to_string()))?;
362    let pass_bytes = pass
363        .current()
364        .map_err(|error| ServiceEffectError(error.to_string()))?;
365    adapter
366        .verify_offline(
367            pass_bytes.as_bytes(),
368            crate::auth_lys::LIMINAL_PASS_AUDIENCE,
369            now,
370        )
371        .map_err(|error| ServiceEffectError(format!("auth.lys.machine pass refused: {error}")))?;
372    Ok(pass_bytes.trim().to_owned())
373}
374
375// Keep the holder so renewal reaches services that are already bound.
376struct MachineHttp {
377    http: HttpLys,
378    pass: crate::link_pass::MachinePass,
379}
380impl LysHttp for MachineHttp {
381    fn get(&self, _: &str, path: &str) -> Result<LysAnswer, ServiceEffectError> {
382        let token = self
383            .pass
384            .current()
385            .map_err(|error| ServiceEffectError(error.to_string()))?;
386        self.http.get(&token, path)
387    }
388    fn post(
389        &self,
390        _: &str,
391        path: &str,
392        body: &serde_json::Value,
393    ) -> Result<LysAnswer, ServiceEffectError> {
394        let token = self
395            .pass
396            .current()
397            .map_err(|error| ServiceEffectError(error.to_string()))?;
398        self.http.post(&token, path, body)
399    }
400}
401
402#[cfg(test)]
403mod tests;
liminal_server/services/production.rs

liminal_server/services/
production.rs
