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

liminal_server/services/
production.rs
