netget/server/dbus/
mod.rs

1//! D-Bus server: NetGet as a small message bus that every call terminates at. A client
2//! authenticates (SASL EXTERNAL on the Unix socket, checked against the peer's uid; ANONYMOUS
3//! where allowed), may say Hello and request names, and every method call that is not the
4//! bus's own goes to the model, which answers with a return value or an error.
5//!
6//! **Fails closed:** a call expecting a reply that gets none from the model — no answer, an
7//! answer that does not encode, a backend failure — is answered with a D-Bus error, never
8//! left to the caller's timeout.
9pub mod actions;
10pub mod wire;
11
12use crate::llm::action_helper::call_llm;
13use crate::logging::emit::Log;
14use crate::protocol::{Event, SpawnContext};
15use crate::server::accept_bounded::{accept_bounded, ConnectionLimiter, DEFAULT_MAX_CONNECTIONS};
16use crate::server::connection::ConnectionId;
17use crate::state::server::{ConnectionState, ConnectionStatus, ProtocolConnectionInfo};
18use actions::DbusProtocol;
19use anyhow::{bail, ensure, Context, Result};
20use serde_json::{json, Value};
21use std::collections::BTreeMap;
22use std::net::SocketAddr;
23use std::sync::atomic::{AtomicU32, Ordering};
24use std::sync::Arc;
25use std::time::Duration;
26use tokio::io::{AsyncBufRead, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader};
27use tokio::sync::Mutex;
28use wire::Message;
29
30pub const DEFAULT_ALLOW_ANONYMOUS: bool = true;
31/// How long an authenticated connection may stay silent, by default.
32pub const IDLE_TIMEOUT: Duration = Duration::from_secs(3600);
33pub const MAX_IDLE_TIMEOUT_SECS: u64 = 86_400;
34/// The SASL handshake, from connect to BEGIN.
35pub const AUTH_TIMEOUT: Duration = Duration::from_secs(10);
36
37const BUS_INTERFACE: &str = "org.freedesktop.DBus";
38const PEER_INTERFACE: &str = "org.freedesktop.DBus.Peer";
39
40/// What every connection of one server shares: the bus's id and who owns which name.
41struct Bus {
42    guid: String,
43    /// Well-known name -> owning unique name; unique names map to themselves.
44    owners: std::sync::Mutex<BTreeMap<String, String>>,
45}
46
47#[derive(Clone)]
48struct Policy {
49    allow_anonymous: bool,
50    idle: Duration,
51}
52
53pub async fn spawn(ctx: SpawnContext) -> Result<SocketAddr> {
54    let params = ctx.startup_params.as_ref();
55    let socket_path = params
56        .map(|p| p.get_optional_string("socket_path"))
57        .transpose()?
58        .flatten();
59    let allow_anonymous = params
60        .map(|p| p.get_optional_bool("allow_anonymous"))
61        .transpose()?
62        .flatten()
63        .unwrap_or(DEFAULT_ALLOW_ANONYMOUS);
64    let idle_secs = params
65        .map(|p| p.get_optional_u64("idle_timeout_secs"))
66        .transpose()?
67        .flatten()
68        .unwrap_or(IDLE_TIMEOUT.as_secs());
69    ensure!(
70        (1..=MAX_IDLE_TIMEOUT_SECS).contains(&idle_secs),
71        "idle_timeout_secs must be between 1 and {MAX_IDLE_TIMEOUT_SECS}"
72    );
73    let policy = Policy {
74        allow_anonymous,
75        idle: Duration::from_secs(idle_secs),
76    };
77    let bus = Arc::new(Bus {
78        guid: wire::random_guid(),
79        owners: Default::default(),
80    });
81    let listener = tokio::net::TcpListener::bind(ctx.legacy_listen_addr())
82        .await
83        .context("D-Bus failed to bind")?;
84    let local = listener.local_addr()?;
85    let log = Log::new(Some(&ctx.status_tx));
86    log.info(format!(
87        "D-Bus server listening on tcp:host={},port={}{}",
88        local.ip(),
89        local.port(),
90        socket_path
91            .as_deref()
92            .map(|p| format!(" and unix:path={p}"))
93            .unwrap_or_default()
94    ));
95    let limiter = Arc::new(ConnectionLimiter::new(DEFAULT_MAX_CONNECTIONS));
96
97    #[cfg(unix)]
98    if let Some(path) = socket_path {
99        let p = std::path::Path::new(&path);
100        if p.exists() {
101            use std::os::unix::fs::FileTypeExt;
102            ensure!(
103                std::fs::symlink_metadata(p)?.file_type().is_socket(),
104                "{path} exists and is not a socket; refusing to replace it"
105            );
106            std::fs::remove_file(p)?;
107        }
108        let unix = tokio::net::UnixListener::bind(&path)
109            .with_context(|| format!("D-Bus failed to bind {path}"))?;
110        let (uctx, bus, policy, limiter) =
111            (ctx.clone(), bus.clone(), policy.clone(), limiter.clone());
112        let task = tokio::spawn(async move {
113            loop {
114                let Ok((stream, permit)) = crate::server::accept_bounded::accept_bounded_unix(
115                    &unix,
116                    &limiter,
117                    b"",
118                    "D-Bus",
119                    Some(&uctx.status_tx),
120                )
121                .await
122                else {
123                    break;
124                };
125                let uid = stream.peer_cred().ok().map(|c| c.uid());
126                let (r, w) = tokio::io::split(stream);
127                spawn_connection(&uctx, &bus, &policy, r, w, local, uid, permit).await;
128            }
129        });
130        ctx.state.register_server_task(ctx.server_id, task).await;
131    }
132    #[cfg(not(unix))]
133    if socket_path.is_some() {
134        bail!("socket_path needs a Unix platform");
135    }
136
137    let state = ctx.state.clone();
138    let server_id = ctx.server_id;
139    let task = tokio::spawn(async move {
140        loop {
141            let Ok((stream, peer, permit)) =
142                accept_bounded(&listener, &limiter, b"", "D-Bus", Some(&ctx.status_tx)).await
143            else {
144                break;
145            };
146            let (r, w) = tokio::io::split(stream);
147            spawn_connection(&ctx, &bus, &policy, r, w, peer, None, permit).await;
148        }
149    });
150    state.register_server_task(server_id, task).await;
151    Ok(local)
152}
153
154#[allow(clippy::too_many_arguments)]
155async fn spawn_connection<R, W, P>(
156    ctx: &SpawnContext,
157    bus: &Arc<Bus>,
158    policy: &Policy,
159    reader: R,
160    writer: W,
161    peer: SocketAddr,
162    peer_uid: Option<u32>,
163    permit: P,
164) where
165    R: AsyncRead + Unpin + Send + 'static,
166    W: AsyncWrite + Unpin + Send + 'static,
167    P: Send + 'static,
168{
169    let id = ConnectionId::new(ctx.state.get_next_unified_id().await);
170    let now = crate::utils::clock::Instant::now();
171    ctx.state
172        .add_connection_to_server(
173            ctx.server_id,
174            ConnectionState {
175                id,
176                remote_addr: peer,
177                local_addr: peer,
178                bytes_sent: 0,
179                bytes_received: 0,
180                packets_sent: 0,
181                packets_received: 0,
182                last_activity: now,
183                status: ConnectionStatus::Active,
184                status_changed_at: now,
185                protocol_info: ProtocolConnectionInfo::empty(),
186            },
187        )
188        .await;
189    let writer = Arc::new(Mutex::new(writer));
190    let peer_rx =
191        crate::server::peer_support::register_peer_channel(&ctx.state, ctx.server_id, id.as_u32())
192            .await;
193    crate::server::peer_support::spawn_peer_command_task(
194        peer_rx,
195        Arc::new(DbusProtocol::new()),
196        ctx.state.clone(),
197        ctx.server_id,
198        id.as_u32(),
199        writer.clone(),
200        ctx.status_tx.clone(),
201    );
202    let (child, bus, policy) = (ctx.clone(), bus.clone(), policy.clone());
203    ctx.state
204        .spawn_server_task(ctx.server_id, async move {
205            let _permit = permit;
206            let mut reader = BufReader::new(reader);
207            let mut conn = Connection {
208                ctx: &child,
209                bus: &bus,
210                writer: &writer,
211                id,
212                serials: Arc::new(AtomicU32::new(1)),
213                unique: None,
214            };
215            let result = async {
216                tokio::time::timeout(
217                    AUTH_TIMEOUT,
218                    authenticate(
219                        &mut reader,
220                        &writer,
221                        peer_uid,
222                        policy.allow_anonymous,
223                        &bus.guid,
224                    ),
225                )
226                .await
227                .context("authentication did not finish within 10 s")??;
228                conn.serve(&mut reader, policy.idle).await
229            }
230            .await;
231            if let Err(e) = result {
232                Log::new(Some(&child.status_tx))
233                    .warn(format!("D-Bus connection {id} from {peer} ended: {e:#}"));
234            }
235            if let Some(unique) = &conn.unique {
236                bus.owners
237                    .lock()
238                    .unwrap_or_else(|e| e.into_inner())
239                    .retain(|_, owner| owner != unique);
240            }
241            child
242                .state
243                .remove_peer_handle(child.server_id, id.as_u32())
244                .await;
245            let _ = writer.lock().await.shutdown().await;
246            child
247                .state
248                .update_connection_status(child.server_id, id, ConnectionStatus::Closed)
249                .await;
250            let _ = child.status_tx.send("__UPDATE_UI__".into());
251        })
252        .await;
253}
254
255/// The server's half of SASL: a NUL byte, then lines until BEGIN. EXTERNAL is offered only
256/// where the peer's uid is known (the Unix socket) and succeeds only for that uid.
257pub async fn authenticate<B, W>(
258    r: &mut B,
259    w: &Arc<Mutex<W>>,
260    peer_uid: Option<u32>,
261    allow_anonymous: bool,
262    guid: &str,
263) -> Result<()>
264where
265    B: AsyncBufRead + Unpin,
266    W: AsyncWrite + Unpin,
267{
268    let mut nul = [0u8; 1];
269    r.read_exact(&mut nul).await?;
270    ensure!(nul[0] == 0, "the peer did not start with the SASL NUL byte");
271    let mut mechanisms = Vec::new();
272    if peer_uid.is_some() {
273        mechanisms.push("EXTERNAL");
274    }
275    if allow_anonymous {
276        mechanisms.push("ANONYMOUS");
277    }
278    let rejected = format!("REJECTED {}\r\n", mechanisms.join(" "));
279    let ok = format!("OK {guid}\r\n");
280    let send = |line: String| async move { w.lock().await.write_all(line.as_bytes()).await };
281    let external_ok = |hex_identity: &str| -> bool {
282        let claimed = hex::decode(hex_identity)
283            .ok()
284            .and_then(|b| String::from_utf8(b).ok());
285        match (peer_uid, claimed) {
286            (Some(uid), Some(c)) => c.is_empty() || c == uid.to_string(),
287            _ => false,
288        }
289    };
290    let mut authed = false;
291    let mut waiting_external_data = false;
292    for _ in 0..wire::MAX_AUTH_LINES {
293        let line = wire::read_auth_line(r).await?;
294        let mut words = line.split_whitespace();
295        let command = words.next().unwrap_or_default();
296        let arg = words.next();
297        let reply = match command {
298            "AUTH" => match arg {
299                Some("EXTERNAL") if mechanisms.contains(&"EXTERNAL") => match words.next() {
300                    Some(identity) if external_ok(identity) => {
301                        authed = true;
302                        ok.clone()
303                    }
304                    Some(_) => rejected.clone(),
305                    None => {
306                        waiting_external_data = true;
307                        "DATA\r\n".into()
308                    }
309                },
310                Some("ANONYMOUS") if allow_anonymous => {
311                    authed = true;
312                    ok.clone()
313                }
314                _ => rejected.clone(),
315            },
316            "DATA" if waiting_external_data => {
317                waiting_external_data = false;
318                if external_ok(arg.unwrap_or_default()) {
319                    authed = true;
320                    ok.clone()
321                } else {
322                    rejected.clone()
323                }
324            }
325            "BEGIN" if authed => return Ok(()),
326            "NEGOTIATE_UNIX_FD" if authed => {
327                "ERROR Unix file descriptors are not supported\r\n".into()
328            }
329            "CANCEL" | "ERROR" => {
330                authed = false;
331                rejected.clone()
332            }
333            _ => "ERROR unexpected command\r\n".into(),
334        };
335        send(reply).await?;
336    }
337    bail!(
338        "authentication took more than {} lines",
339        wire::MAX_AUTH_LINES
340    )
341}
342
343struct Connection<'a, W> {
344    ctx: &'a SpawnContext,
345    bus: &'a Arc<Bus>,
346    writer: &'a Arc<Mutex<W>>,
347    id: ConnectionId,
348    serials: Arc<AtomicU32>,
349    unique: Option<String>,
350}
351
352impl<W: AsyncWrite + Unpin> Connection<'_, W> {
353    async fn send(&self, mut m: Message) -> Result<()> {
354        m.serial = self.serials.fetch_add(1, Ordering::Relaxed);
355        if m.destination.is_none() {
356            m.destination = self.unique.clone();
357        }
358        self.write(&m.encode()?).await
359    }
360
361    async fn write(&self, bytes: &[u8]) -> Result<()> {
362        self.writer.lock().await.write_all(bytes).await?;
363        self.ctx
364            .state
365            .update_connection_stats(
366                self.ctx.server_id,
367                self.id,
368                None,
369                Some(bytes.len() as u64),
370                None,
371                Some(1),
372            )
373            .await;
374        Ok(())
375    }
376
377    async fn serve<B: AsyncBufRead + Unpin>(&mut self, r: &mut B, idle: Duration) -> Result<()> {
378        loop {
379            let message = match tokio::time::timeout(idle, wire::read_message(r)).await {
380                Err(_) => return Ok(()),
381                Ok(m) => match m? {
382                    Some(m) => m,
383                    None => return Ok(()),
384                },
385            };
386            self.ctx
387                .state
388                .update_connection_stats(self.ctx.server_id, self.id, None, None, Some(1), None)
389                .await;
390            match message.kind {
391                wire::METHOD_CALL if is_bus_call(&message) => self.bus_method(&message).await?,
392                wire::METHOD_CALL if message.interface.as_deref() == Some(PEER_INTERFACE) => {
393                    self.peer_method(&message).await?
394                }
395                wire::METHOD_CALL | wire::SIGNAL => self.ask_model(message).await?,
396                other => tracing::debug!(
397                    "D-Bus ignored a {} from connection {}",
398                    wire::type_name(other),
399                    self.id
400                ),
401            }
402        }
403    }
404
405    async fn reply(&self, call: &Message, signature: &str, body: Vec<Value>) -> Result<()> {
406        if call.expects_reply() {
407            self.send(Message::reply_to(call, signature, body)).await?;
408        }
409        Ok(())
410    }
411
412    async fn error(&self, call: &Message, name: &str, text: &str) -> Result<()> {
413        if call.expects_reply() {
414            self.send(Message::error_to(call, name, text)).await?;
415        }
416        Ok(())
417    }
418
419    fn owners(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, String>> {
420        self.bus.owners.lock().unwrap_or_else(|e| e.into_inner())
421    }
422
423    fn bus_signal(&self, member: &str, signature: &str, body: Vec<Value>) -> Message {
424        Message {
425            kind: wire::SIGNAL,
426            path: Some(wire::BUS_PATH.into()),
427            interface: Some(BUS_INTERFACE.into()),
428            member: Some(member.into()),
429            sender: Some(wire::BUS_NAME.into()),
430            signature: signature.into(),
431            body,
432            ..Default::default()
433        }
434    }
435
436    /// The methods a message bus answers itself.
437    async fn bus_method(&mut self, call: &Message) -> Result<()> {
438        let member = call.member.as_deref().unwrap_or_default();
439        let arg = call.first_string().unwrap_or_default().to_string();
440        if member != "Hello" && self.unique.is_none() && call.destination.is_some() {
441            return self
442                .error(
443                    call,
444                    "org.freedesktop.DBus.Error.AccessDenied",
445                    "Client tried to send a message other than Hello without being registered",
446                )
447                .await;
448        }
449        match member {
450            "Hello" => {
451                if self.unique.is_some() {
452                    return self
453                        .error(
454                            call,
455                            "org.freedesktop.DBus.Error.Failed",
456                            "Already handled an Hello message",
457                        )
458                        .await;
459                }
460                let unique = format!(":1.{}", self.id.as_u32());
461                self.owners().insert(unique.clone(), unique.clone());
462                self.unique = Some(unique.clone());
463                self.reply(call, "s", vec![json!(unique)]).await?;
464                self.send(self.bus_signal("NameAcquired", "s", vec![json!(unique)]))
465                    .await
466            }
467            "RequestName" => {
468                if !wire::valid_bus_name(&arg) || arg.starts_with(':') {
469                    return self
470                        .error(
471                            call,
472                            "org.freedesktop.DBus.Error.InvalidArgs",
473                            &format!("{arg:?} is not a well-known bus name"),
474                        )
475                        .await;
476                }
477                let me = self
478                    .unique
479                    .clone()
480                    .unwrap_or_else(|| format!(":1.{}", self.id.as_u32()));
481                let code = {
482                    let mut owners = self.owners();
483                    match owners.get(&arg) {
484                        Some(owner) if *owner == me => 4, // ALREADY_OWNER
485                        Some(_) => 3,                     // EXISTS
486                        None => {
487                            owners.insert(arg.clone(), me);
488                            1 // PRIMARY_OWNER
489                        }
490                    }
491                };
492                self.reply(call, "u", vec![json!(code)]).await?;
493                if code == 1 {
494                    self.send(self.bus_signal("NameAcquired", "s", vec![json!(arg)]))
495                        .await?;
496                }
497                Ok(())
498            }
499            "ReleaseName" => {
500                let me = self.unique.clone().unwrap_or_default();
501                let code = {
502                    let mut owners = self.owners();
503                    match owners.get(&arg) {
504                        Some(owner) if *owner == me && !arg.starts_with(':') => {
505                            owners.remove(&arg);
506                            1
507                        }
508                        Some(_) => 3,
509                        None => 2,
510                    }
511                };
512                self.reply(call, "u", vec![json!(code)]).await
513            }
514            "GetId" => self.reply(call, "s", vec![json!(self.bus.guid)]).await,
515            "ListNames" => {
516                let mut names = vec![json!(wire::BUS_NAME)];
517                names.extend(self.owners().keys().map(|n| json!(n)));
518                self.reply(call, "as", vec![Value::Array(names)]).await
519            }
520            "ListActivatableNames" => self.reply(call, "as", vec![json!([wire::BUS_NAME])]).await,
521            "NameHasOwner" => {
522                let has = arg == wire::BUS_NAME || self.owners().contains_key(&arg);
523                self.reply(call, "b", vec![json!(has)]).await
524            }
525            "GetNameOwner" => {
526                if arg == wire::BUS_NAME {
527                    return self.reply(call, "s", vec![json!(wire::BUS_NAME)]).await;
528                }
529                let owner = self.owners().get(&arg).cloned();
530                match owner {
531                    Some(o) => self.reply(call, "s", vec![json!(o)]).await,
532                    None => {
533                        self.error(
534                            call,
535                            "org.freedesktop.DBus.Error.NameHasNoOwner",
536                            &format!("Could not get owner of name '{arg}': no such name"),
537                        )
538                        .await
539                    }
540                }
541            }
542            "AddMatch" | "RemoveMatch" => self.reply(call, "", vec![]).await,
543            _ => {
544                self.error(
545                    call,
546                    "org.freedesktop.DBus.Error.UnknownMethod",
547                    &format!("NetGet's bus does not implement {member}"),
548                )
549                .await
550            }
551        }
552    }
553
554    async fn peer_method(&self, call: &Message) -> Result<()> {
555        match call.member.as_deref() {
556            Some("Ping") => self.reply(call, "", vec![]).await,
557            Some("GetMachineId") => self.reply(call, "s", vec![json!(self.bus.guid)]).await,
558            _ => {
559                self.error(
560                    call,
561                    "org.freedesktop.DBus.Error.UnknownMethod",
562                    "org.freedesktop.DBus.Peer has Ping and GetMachineId",
563                )
564                .await
565            }
566        }
567    }
568
569    /// A call or signal for the model, with the fail-closed rule for calls.
570    async fn ask_model(&self, mut m: Message) -> Result<()> {
571        if m.sender.is_none() {
572            m.sender = self.unique.clone();
573        }
574        let args = Value::Array(m.body.clone());
575        let event = if m.kind == wire::METHOD_CALL {
576            Event::new(
577                &actions::METHOD_CALL_EVENT,
578                json!({
579                    "path": m.path, "interface": m.interface, "member": m.member,
580                    "destination": m.destination, "sender": m.sender,
581                    "signature": m.signature, "args": args, "no_reply_expected": !m.expects_reply(),
582                }),
583            )
584        } else {
585            Event::new(
586                &actions::SIGNAL_EVENT,
587                json!({"path": m.path, "interface": m.interface, "member": m.member, "sender": m.sender, "signature": m.signature, "args": args}),
588            )
589        };
590        let protocol = DbusProtocol::for_call(m.clone(), self.serials.clone());
591        let outcome = call_llm(
592            &self.ctx.llm_client,
593            &self.ctx.state,
594            self.ctx.server_id,
595            Some(self.id),
596            &event,
597            &protocol,
598        )
599        .await;
600        let log = Log::new(Some(&self.ctx.status_tx));
601        let summary = format!(
602            "D-Bus {} {}.{} from connection {}",
603            wire::type_name(m.kind),
604            m.interface.as_deref().unwrap_or("-"),
605            m.member.as_deref().unwrap_or("-"),
606            self.id
607        );
608        let execution = match outcome {
609            Ok(e) => e,
610            Err(e) => {
611                let (name, category) = match crate::utils::wire_failure::WireFailure::classify(&e) {
612                    crate::utils::wire_failure::WireFailure::Overloaded => {
613                        ("org.freedesktop.DBus.Error.LimitsExceeded", "overloaded")
614                    }
615                    crate::utils::wire_failure::WireFailure::Unavailable => {
616                        ("org.freedesktop.DBus.Error.Failed", "unavailable")
617                    }
618                };
619                log.error(format!(
620                    "{summary} decision=fail_closed_llm_error category={category}: {e}"
621                ));
622                return self
623                    .error(
624                        &m,
625                        name,
626                        crate::utils::wire_failure::WireFailure::classify(&e).text(),
627                    )
628                    .await;
629            }
630        };
631        let mut replied = false;
632        for result in &execution.protocol_results {
633            for output in result.get_all_output() {
634                replied |= matches!(
635                    output.get(1),
636                    Some(&wire::METHOD_RETURN) | Some(&wire::ERROR)
637                );
638                self.write(&output).await?;
639            }
640        }
641        if !m.expects_reply() {
642            log.info(format!("{summary} decision=model_answered"));
643            return Ok(());
644        }
645        if replied {
646            log.info(format!("{summary} decision=model_reply"));
647            return Ok(());
648        }
649        log.error(format!(
650            "{summary} decision=fail_closed_no_answer (no usable dbus_return or dbus_error)"
651        ));
652        self.error(
653            &m,
654            "org.freedesktop.DBus.Error.Failed",
655            "No answer was produced for this call",
656        )
657        .await
658    }
659}
660
661fn is_bus_call(m: &Message) -> bool {
662    match m.destination.as_deref() {
663        Some(d) => d == wire::BUS_NAME,
664        None => {
665            m.path.as_deref() == Some(wire::BUS_PATH)
666                && m.interface.as_deref() == Some(BUS_INTERFACE)
667        }
668    }
669}
netget/server/dbus/mod.rs

netget/server/dbus/
mod.rs
