tawasuyu/brahman

1//! Listener del bus interno. Vive en PID 1, acepta conexiones de Entes hijos,
2//! extrae credenciales del kernel vía SO_PEERCRED, y enruta cada request al
3//! grafo. Conexión bidireccional: el grafo puede *empujar* requests hacia
4//! una conexión registrada (forwarding de Invoke al proveedor).
5//!
6//! ## Por qué bidireccional
7//!
8//! Un Ente que provee `Capability::Endpoint` debe poder *recibir* invokes
9//! sin abrir más sockets. Después de Announce, el grafo guarda el lado de
10//! escritura de su conexión y lo usa para forwardear.
11
12use crate::events::GraphEvent;
13use ente_bus::{read_frame, write_frame, BusMessage, BusPayload, BusResponse, PeerCreds};
14use nix::sys::socket::{getsockopt, sockopt::PeerCredentials};
15use std::path::PathBuf;
16use tokio::net::{UnixListener, UnixStream};
17use tokio::sync::{mpsc, oneshot};
18use tracing::{error, info, trace, warn};
19use ulid::Ulid;
20
21pub fn default_socket_path() -> PathBuf {
22    if let Ok(p) = std::env::var(ente_bus::ENV_BUS_SOCK) {
23        return p.into();
24    }
25    let runtime = std::env::var("XDG_RUNTIME_DIR")
26        .unwrap_or_else(|_| std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()));
27    let user = std::env::var("USER").unwrap_or_else(|_| "ente".into());
28    format!("{runtime}/ente-bus-{user}.sock").into()
29}
30
31pub fn spawn_bus(path: PathBuf, graph_tx: mpsc::Sender<GraphEvent>) -> anyhow::Result<PathBuf> {
32    let _ = std::fs::remove_file(&path);
33    if let Some(parent) = path.parent() {
34        let _ = std::fs::create_dir_all(parent);
35    }
36    let listener = UnixListener::bind(&path)?;
37    info!(path = %path.display(), "bus interno escuchando");
38
39    let path_returned = path.clone();
40    tokio::spawn(async move {
41        loop {
42            match listener.accept().await {
43                Ok((stream, _addr)) => {
44                    let tx = graph_tx.clone();
45                    tokio::spawn(async move {
46                        if let Err(e) = handle_conn(stream, tx).await {
47                            warn!(?e, "bus connection ended");
48                        }
49                    });
50                }
51                Err(e) => {
52                    error!(?e, "bus accept failed, listener cerrando");
53                    return;
54                }
55            }
56        }
57    });
58
59    Ok(path_returned)
60}
61
62async fn handle_conn(stream: UnixStream, graph_tx: mpsc::Sender<GraphEvent>) -> anyhow::Result<()> {
63    // SO_PEERCRED: el kernel adjunta pid/uid/gid al socket en connect/accept.
64    // No-falsificable desde el cliente.
65    let creds = getsockopt(&stream, PeerCredentials)
66        .map_err(|e| anyhow::anyhow!("getsockopt PEERCRED: {e}"))?;
67    let peer = PeerCreds {
68        pid: creds.pid(),
69        uid: creds.uid(),
70        gid: creds.gid(),
71    };
72    trace!(?peer, "bus conn aceptada");
73
74    let (mut reader, mut writer) = stream.into_split();
75    let (out_tx, mut out_rx) = mpsc::channel::<BusMessage>(64);
76
77    // Writer task: única vía de escritura al socket. Multiplexa entre
78    // respuestas a peticiones del cliente y forwards iniciados por el grafo.
79    let writer_task = tokio::spawn(async move {
80        while let Some(msg) = out_rx.recv().await {
81            if let Err(e) = write_frame(&mut writer, &msg).await {
82                warn!(?e, "bus writer falló, terminando");
83                return;
84            }
85        }
86    });
87
88    let mut announced_id: Option<Ulid> = None;
89    let result: anyhow::Result<()> = (async {
90        loop {
91            let msg = match read_frame(&mut reader).await {
92                Ok(m) => m,
93                Err(e) => {
94                    trace!(?e, "bus conn read terminó");
95                    return Ok(());
96                }
97            };
98            match msg.payload {
99                BusPayload::Request(req) => {
100                    let is_announce = matches!(req, ente_bus::BusRequest::Announce { .. });
101                    let (reply_tx, reply_rx) = oneshot::channel();
102                    if graph_tx.send(GraphEvent::BusRequest {
103                        peer,
104                        from: msg.from,
105                        request: req,
106                        outbound: out_tx.clone(),
107                        reply: reply_tx,
108                    }).await.is_err() {
109                        warn!("graph cerrado, terminando bus connection");
110                        return Ok(());
111                    }
112                    let response = reply_rx.await.unwrap_or_else(|_| {
113                        BusResponse::Error("graph dropped reply channel".into())
114                    });
115                    if is_announce && matches!(response, BusResponse::Ok) {
116                        // Auth del Announce ya fue verificada por el grafo;
117                        // memorizamos para cleanup en cierre.
118                        announced_id = msg.from;
119                    }
120                    let out = BusMessage {
121                        from: None,
122                        seq: msg.seq,
123                        payload: BusPayload::Response(response),
124                    };
125                    if out_tx.send(out).await.is_err() { return Ok(()); }
126                }
127                BusPayload::Response(resp) => {
128                    // Respuesta a un Invoke que el grafo forwardeó a este peer.
129                    let _ = graph_tx.send(GraphEvent::BusResponse {
130                        seq: msg.seq,
131                        response: resp,
132                    }).await;
133                }
134            }
135        }
136    }).await;
137
138    if let Some(id) = announced_id {
139        let _ = graph_tx.send(GraphEvent::BusConnClosed { ente_id: Some(id) }).await;
140    }
141    writer_task.abort();
142    result
143}