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;
33use crate::server::connection::LiminalConnectionServices;
34use std::net::SocketAddr;
35use std::sync::Arc;
36mod debug;
37
38/// Administrative effect seams and an optional address broker, bound once.
39pub struct ServiceBindings {
40    administrative: Option<AdministrativeBindings>,
41    context: Arc<ContextReporter>,
42    watch_context: bool,
43    broker: Option<Broker>,
44    echo: Option<ObservedEcho>,
45    relay: Option<Arc<super::relay::Relay>>,
46    relay_topology_local: Option<String>,
47}
48
49#[derive(Debug)]
50struct AdministrativeBindings {
51    seats: LysSeatStarter,
52    keys: ChannelKeySender,
53    budgets: LysBudgets,
54    actions: PublishingActions,
55}
56
57impl std::fmt::Debug for ServiceBindings {
58    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
59        formatter
60            .debug_struct("ServiceBindings")
61            .field("administrative", &self.administrative)
62            .field("watch_context", &self.watch_context)
63            .field("broker", &self.broker)
64            .field("echo", &self.echo)
65            .field("relay", &self.relay)
66            .finish_non_exhaustive()
67    }
68}
69
70impl ServiceBindings {
71    /// Binds every seam: Lys asked through `http` with `bearer` (this
72    /// machine's own pass), seats started on `machine`, messages published
73    /// through `publisher`.
74    #[must_use]
75    pub fn new(
76        http: &Arc<dyn LysHttp>,
77        bearer: &str,
78        machine: String,
79        publisher: &Arc<dyn ServicePublisher>,
80    ) -> Self {
81        let context = Arc::new(ContextReporter::new(Arc::clone(http), bearer.to_owned()));
82        Self {
83            administrative: Some(AdministrativeBindings {
84                seats: LysSeatStarter::new(Arc::clone(http), bearer.to_owned(), machine),
85                keys: ChannelKeySender::new(Arc::clone(publisher)),
86                budgets: LysBudgets::new(Arc::clone(http), bearer.to_owned()),
87                actions: PublishingActions::new(Arc::clone(publisher), Arc::clone(&context)),
88            }),
89            context,
90            watch_context: false,
91            broker: None,
92            echo: None,
93            relay: None,
94            relay_topology_local: None,
95        }
96    }
97
98    /// Whether the operator asked for the context watch to run.
99    #[must_use]
100    pub const fn watches_context(&self) -> bool {
101        self.watch_context
102    }
103
104    /// The seams, lent for one [`serve`](super::serve).
105    #[must_use]
106    pub fn services(&self) -> Services<'_> {
107        Services {
108            seats: self,
109            keys: self,
110            budgets: self,
111            actions: self,
112        }
113    }
114
115    /// The reporter the context watch shares with the `context-watch` call.
116    #[must_use]
117    pub fn context_reporter(&self) -> Arc<ContextReporter> {
118        Arc::clone(&self.context)
119    }
120
121    /// The configured address broker, independent of administrative services.
122    #[must_use]
123    pub const fn broker(&self) -> Option<&Broker> {
124        self.broker.as_ref()
125    }
126
127    /// The address actually bound by the configured broker's UDP echo.
128    #[must_use]
129    pub fn echo_address(&self) -> Option<SocketAddr> {
130        self.echo.as_ref().map(ObservedEcho::local_addr)
131    }
132
133    /// The configured blind relay, independent of administrative services.
134    #[must_use]
135    pub const fn relay(&self) -> Option<&Arc<super::relay::Relay>> {
136        self.relay.as_ref()
137    }
138
139    /// The console's canonical local endpoint for relay observations.
140    #[must_use]
141    pub fn relay_topology_local(&self) -> Option<&str> {
142        self.relay_topology_local.as_deref()
143    }
144
145    /// Serves a broker request on the connection's admitted session.
146    ///
147    /// # Errors
148    /// The broker or admitted session is absent, or the broker refuses the call.
149    pub fn serve_broker(
150        &self,
151        session: Option<&dyn crate::auth_lys::session::LysSession>,
152        subject: &str,
153        request: &ServiceRequest<'_>,
154        now: u64,
155    ) -> Result<serde_json::Value, BrokerRefusal> {
156        let broker = self
157            .broker
158            .as_ref()
159            .ok_or(BrokerRefusal::Named("broker_not_configured"))?;
160        let session = session.ok_or(BrokerRefusal::Named("pass_not_admitted"))?;
161        broker.serve(session, subject, request, now)
162    }
163
164    /// Builds configured services when the boot machine pass is present.
165    /// `Ok(None)` when no services are selected or a dial pass was refused.
166    /// Lys is asked on `handle`'s runtime; messages are published through
167    /// `services` as `liminal.<host>`.
168    ///
169    /// # Errors
170    /// `[auth.lys.machine]` is absent (the services would have no identity to
171    /// ask Lys as), this machine's pass or the issuer's key set cannot be
172    /// read or does not verify, or the issuer is not an HTTP origin.
173    pub fn configure(
174        lys: &LysConfig,
175        services: Arc<LiminalConnectionServices>,
176        handle: tokio::runtime::Handle,
177        now: u64,
178    ) -> Result<Option<Self>, ServiceEffectError> {
179        if lys.services.is_none()
180            && lys.broker.is_none()
181            && !lys.relay.as_ref().is_some_and(|relay| relay.enabled)
182        {
183            return Ok(None);
184        }
185        let machine = lys
186            .machine
187            .as_ref()
188            .ok_or_else(|| ServiceEffectError("auth.lys.machine is absent".into()))?;
189        let pass = match crate::link_pass::MachinePass::configured(machine) {
190            Ok(pass) => pass,
191            Err(error) if !machine.agent_id.is_empty() && error.source_absent() => return Ok(None),
192            Err(error) => return Err(ServiceEffectError(error.to_string())),
193        };
194        let bearer = machine_bearer(lys, now)?;
195        let http: Arc<dyn LysHttp> = Arc::new(MachineHttp {
196            http: HttpLys::new(lys.api_url.as_deref().unwrap_or(&lys.issuer), handle)?,
197            pass,
198        });
199        let host = lys
200            .services
201            .as_ref()
202            .map_or("broker", |section| section.host.as_str());
203        let topology_local = services.console().device().to_owned();
204        let publisher: Arc<dyn ServicePublisher> =
205            Arc::new(ConnectionPublisher::new(services, host.to_owned()));
206        let mut bindings = if let Some(section) = &lys.services {
207            let mut bindings = Self::new(&http, &bearer, section.machine.clone(), &publisher);
208            bindings.watch_context = section.context_watch;
209            bindings
210        } else {
211            Self {
212                administrative: None,
213                context: Arc::new(ContextReporter::new(Arc::clone(&http), bearer)),
214                watch_context: false,
215                broker: None,
216                echo: None,
217                relay: None,
218                relay_topology_local: None,
219            }
220        };
221        if let Some(section) = &lys.broker {
222            let machine = lys.machine.as_ref().ok_or_else(|| {
223                ServiceEffectError("auth.lys.broker requires auth.lys.machine".to_owned())
224            })?;
225            let recorded = machine
226                .machines
227                .iter()
228                .map(|(holder, key)| {
229                    let mut bytes = [0; 32];
230                    hex::decode_to_slice(key, &mut bytes).map_err(|_| {
231                        ServiceEffectError(format!(
232                            "auth.lys.machine.machines: invalid key for holder {holder}"
233                        ))
234                    })?;
235                    let identity = crate::link_handshake::MachineIdentity::new(bytes);
236                    identity.x25519_public().map_err(|error| {
237                        ServiceEffectError(format!(
238                            "auth.lys.machine.machines: holder {holder}: {error}"
239                        ))
240                    })?;
241                    Ok((holder.clone(), identity))
242                })
243                .collect::<Result<_, ServiceEffectError>>()?;
244            let broker = Broker::new(lys.network_id.clone(), recorded, publisher);
245            bindings.echo = Some(
246                ObservedEcho::bind(section.udp_listen, broker.counters()).map_err(|error| {
247                    ServiceEffectError(format!("auth.lys.broker.udp_listen: {error}"))
248                })?,
249            );
250            bindings.broker = Some(broker);
251        }
252        if lys.relay.as_ref().is_some_and(|relay| relay.enabled) {
253            let machine = lys.machine.as_ref().ok_or_else(|| {
254                ServiceEffectError("auth.lys.relay requires auth.lys.machine".to_owned())
255            })?;
256            let recorded = machine
257                .machines
258                .iter()
259                .map(|(holder, key)| {
260                    let mut bytes = [0; 32];
261                    hex::decode_to_slice(key, &mut bytes).map_err(|_| {
262                        ServiceEffectError(format!("invalid recorded machine for holder {holder}"))
263                    })?;
264                    let identity = crate::link_handshake::MachineIdentity::new(bytes);
265                    identity
266                        .x25519_public()
267                        .map_err(|error| ServiceEffectError(error.to_string()))?;
268                    Ok((holder.clone(), identity))
269                })
270                .collect::<Result<_, ServiceEffectError>>()?;
271            bindings.relay = Some(super::relay::Relay::new(lys.network_id.clone(), recorded));
272            bindings.relay_topology_local = Some(topology_local);
273        }
274        Ok(Some(bindings))
275    }
276
277    fn administrative(&self) -> Result<&AdministrativeBindings, ServiceEffectError> {
278        self.administrative
279            .as_ref()
280            .ok_or_else(|| ServiceEffectError("auth.lys.services is not configured".to_owned()))
281    }
282}
283
284impl SeatStarter for ServiceBindings {
285    fn start(&self, seat: &str, operation: &str) -> Result<(), ServiceEffectError> {
286        self.administrative()?.seats.start(seat, operation)
287    }
288}
289
290impl KeySender for ServiceBindings {
291    fn send(&self, session: &str, keys: &[u8], operation: &str) -> Result<(), ServiceEffectError> {
292        self.administrative()?.keys.send(session, keys, operation)
293    }
294}
295
296impl BudgetSource for ServiceBindings {
297    fn read(&self, name: &str) -> Result<Option<BudgetPolicy>, ServiceEffectError> {
298        self.administrative()?.budgets.read(name)
299    }
300}
301
302impl ServiceAction for ServiceBindings {
303    fn act(
304        &self,
305        service: ServiceKind,
306        payload: &[u8],
307        operation: &str,
308    ) -> Result<(), ServiceEffectError> {
309        self.administrative()?
310            .actions
311            .act(service, payload, operation)
312    }
313}
314
315/// This machine's own Lys pass as the bearer the services ask Lys with.
316///
317/// It is verified offline for audience `liminal` at `now` against the
318/// issuer's key set, so an expired or foreign pass refuses start by its
319/// cause.
320///
321/// # Errors
322/// No `[auth.lys.machine]`, or the pass or key set is unreadable or refused.
323pub fn machine_bearer(lys: &LysConfig, now: u64) -> Result<String, ServiceEffectError> {
324    let machine = lys.machine.as_ref().ok_or_else(|| {
325        ServiceEffectError(
326            "auth.lys.services or auth.lys.broker needs auth.lys.machine: services use this machine's pass"
327                .to_owned(),
328        )
329    })?;
330    let keys_json = std::fs::read_to_string(&lys.keys_path).map_err(|error| {
331        ServiceEffectError(format!(
332            "auth.lys.keys_path {}: {error}",
333            lys.keys_path.display()
334        ))
335    })?;
336    let adapter = crate::auth_lys::lys_pass_adapter::LysPassAdapter::from_keys_json(
337        &keys_json,
338        lys.issuer.clone(),
339    )
340    .map_err(|error| ServiceEffectError(format!("auth.lys key set refused: {error}")))?;
341    let pass = crate::link_pass::MachinePass::configured(machine)
342        .map_err(|error| ServiceEffectError(error.to_string()))?;
343    let pass_bytes = pass
344        .current()
345        .map_err(|error| ServiceEffectError(error.to_string()))?;
346    adapter
347        .verify_offline(
348            pass_bytes.as_bytes(),
349            crate::auth_lys::LIMINAL_PASS_AUDIENCE,
350            now,
351        )
352        .map_err(|error| ServiceEffectError(format!("auth.lys.machine pass refused: {error}")))?;
353    Ok(pass_bytes.trim().to_owned())
354}
355
356// Keep the holder so renewal reaches services that are already bound.
357struct MachineHttp {
358    http: HttpLys,
359    pass: crate::link_pass::MachinePass,
360}
361impl LysHttp for MachineHttp {
362    fn get(&self, _: &str, path: &str) -> Result<LysAnswer, ServiceEffectError> {
363        let token = self
364            .pass
365            .current()
366            .map_err(|error| ServiceEffectError(error.to_string()))?;
367        self.http.get(&token, path)
368    }
369    fn post(
370        &self,
371        _: &str,
372        path: &str,
373        body: &serde_json::Value,
374    ) -> Result<LysAnswer, ServiceEffectError> {
375        let token = self
376            .pass
377            .current()
378            .map_err(|error| ServiceEffectError(error.to_string()))?;
379        self.http.post(&token, path, body)
380    }
381}
382
383#[cfg(test)]
384mod tests {
385    use super::ServiceBindings;
386    use crate::config::types::LysConfig;
387    use crate::server::connection::LiminalConnectionServices;
388    use crate::services::ServiceRequest;
389    use lys_core::keys::Ed25519Identity;
390    use serde_json::json;
391    use std::net::UdpSocket;
392    use std::sync::Arc;
393    use std::time::Duration;
394
395    fn base64url(bytes: &[u8]) -> String {
396        const ALPHABET: &[u8; 64] =
397            b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
398        let mut encoded = String::new();
399        for chunk in bytes.chunks(3) {
400            let first = chunk[0];
401            encoded.push(char::from(ALPHABET[usize::from(first >> 2)]));
402            let second = chunk.get(1).copied().unwrap_or(0);
403            encoded.push(char::from(
404                ALPHABET[usize::from((first & 3) << 4 | second >> 4)],
405            ));
406            if chunk.len() > 1 {
407                let third = chunk.get(2).copied().unwrap_or(0);
408                encoded.push(char::from(
409                    ALPHABET[usize::from((second & 15) << 2 | third >> 6)],
410                ));
411                if chunk.len() > 2 {
412                    encoded.push(char::from(ALPHABET[usize::from(third & 63)]));
413                }
414            }
415        }
416        encoded
417    }
418
419    fn machine_pass() -> Result<(String, String), serde_json::Error> {
420        let issuer = Ed25519Identity::from_seed(&zeroize::Zeroizing::new([9; 32]));
421        let keys = json!({"keys": [{"kty": "OKP", "crv": "Ed25519", "kid": "broker",
422            "x": base64url(&issuer.public_key_bytes()), "alg": "EdDSA", "use": "sig"}]});
423        let claims = json!({"iss": "https://issuer.example", "sub": "broker", "aud": "liminal",
424            "iat": 0, "exp": 20, "rights": [],
425            "holder": {"id": "broker", "kind": "machine", "responsible": "owner"}});
426        let header = base64url(br#"{"alg":"EdDSA","typ":"JWT","kid":"broker"}"#);
427        let signed = format!("{header}.{}", base64url(&serde_json::to_vec(&claims)?));
428        let token = format!("{signed}.{}", base64url(&issuer.sign(signed.as_bytes())));
429        Ok((keys.to_string(), token))
430    }
431
432    #[test]
433    fn broker_only_configuration_binds_echo_and_refuses_unadmitted_calls()
434    -> Result<(), Box<dyn std::error::Error>> {
435        let root = tempfile::tempdir()?;
436        let keys = root.path().join("issuer-keys.json");
437        let pass = root.path().join("machine.pass");
438        let (keys_json, token) = machine_pass()?;
439        std::fs::write(&keys, keys_json)?;
440        std::fs::write(&pass, token)?;
441        let mut lys: LysConfig = serde_json::from_value(json!({
442            "issuer": "https://issuer.example", "network_id": "network-a",
443            "keys_path": keys, "broker": {"udp_listen": "127.0.0.1:0"},
444            "machine": {"identity_path": "unused", "pass_path": pass}
445        }))?;
446        let runtime = tokio::runtime::Builder::new_current_thread().build()?;
447        let services = Arc::new(LiminalConnectionServices::empty()?);
448        let bindings =
449            ServiceBindings::configure(&lys, services.clone(), runtime.handle().clone(), 10)?
450                .ok_or("broker did not configure")?;
451        let refused = bindings
452            .serve_broker(
453                None,
454                "net.broker.lookup",
455                &ServiceRequest {
456                    operation: "lookup",
457                    payload: b"{}",
458                },
459                10,
460            )
461            .err()
462            .ok_or("unadmitted call accepted")?;
463        assert_eq!(refused.to_string(), "pass_not_admitted");
464        let refused = bindings
465            .services()
466            .seats
467            .start("unused", "operation")
468            .err()
469            .ok_or("unconfigured administrative service acted")?;
470        assert!(
471            refused
472                .to_string()
473                .contains("auth.lys.services is not configured")
474        );
475        let address = bindings.echo_address().ok_or("echo missing")?;
476        assert_ne!(address.port(), 0);
477        let counters = bindings.broker().ok_or("broker missing")?.counters();
478        assert_eq!(counters.snapshot().echo_answered, 0);
479        let socket = UdpSocket::bind("127.0.0.1:0")?;
480        socket.set_read_timeout(Some(Duration::from_secs(3)))?;
481        socket.send_to(&[b"LMOB".as_slice(), &[3; 16]].concat(), address)?;
482        let mut reply = [0; 64];
483        let (length, _) = socket.recv_from(&mut reply)?;
484        assert_eq!(length, 27);
485        assert_eq!(&reply[..20], &[b"LMOA".as_slice(), &[3; 16]].concat());
486        drop(bindings);
487        assert_eq!(counters.snapshot().echo_answered, 1);
488        let rebound = UdpSocket::bind(address)?;
489        assert_eq!(rebound.local_addr()?, address);
490        lys.broker = None;
491        lys.services = Some(toml::from_str("machine = 'unused'\nhost = 'broker'")?);
492        let bindings = ServiceBindings::configure(&lys, services, runtime.handle().clone(), 10)?
493            .ok_or("administrative services did not configure")?;
494        assert_eq!(bindings.echo_address(), None);
495        Ok(())
496    }
497}
liminal_server/services/production.rs

liminal_server/services/
production.rs
