tawasuyu/arje

1//! ente-bus: bus de capacidades interno del fractal.
2//!
3//! Wire format: Unix SOCK_STREAM con framing length-prefijo (u32 BE) + payload
4//! postcard. Bidireccional pero por ahora request-response síncrono.
5//!
6//! Identidad: cada Ente hijo recibe `ENTE_BUS_SOCK` y `ENTE_ID` en su entorno.
7//! El cliente lee ambos vía `BusClient::from_env`.
8
9use ente_card::Capability;
10use serde::{Deserialize, Serialize};
11use std::path::Path;
12use std::str::FromStr;
13use tokio::io::{AsyncReadExt, AsyncWriteExt};
14use tokio::net::UnixStream;
15use ulid::Ulid;
16
17pub const ENV_BUS_SOCK: &str = "ENTE_BUS_SOCK";
18pub const ENV_ENTE_ID: &str = "ENTE_ID";
19pub const MAX_FRAME: usize = 1 << 20; // 1 MiB — protección contra OOM
20
21/// Interface UUID para decisiones de policy. Un Ente independiente
22/// (separado de polkit-compat) se anuncia como proveedor de
23/// `Capability::Endpoint { interface: POLKIT_DECISION_IFACE, version: 1 }`
24/// para arbitrar autorizaciones. Recibe blob:
25/// `pid_be | uid_be | action_id_utf8` → responde 1 byte: 1=allow, 0=deny.
26pub const POLKIT_DECISION_IFACE: ente_card::InterfaceId =
27    ente_card::InterfaceId([0xb0; 16]);
28
29/// Interface UUID auto-anunciado por compat-polkit. Diferente al de
30/// decisión para evitar recursión (polkit-compat invoca DECISION pero
31/// no es proveedor de DECISION; se anuncia como SERVICE).
32pub const POLKIT_SERVICE_IFACE: ente_card::InterfaceId =
33    ente_card::InterfaceId([0xa4; 16]);
34
35/// Credenciales del peer extraídas vía SO_PEERCRED al accept del bus.
36/// Imposibles de falsear desde el cliente — el kernel las inyecta.
37/// Definidas aquí (no en ente-zero) porque conceptualmente son atributo
38/// del protocolo del bus, no del init.
39#[derive(Debug, Clone, Copy)]
40pub struct PeerCreds {
41    pub pid: i32,
42    pub uid: u32,
43    pub gid: u32,
44}
45
46#[derive(Debug, Clone, Serialize, Deserialize)]
47pub struct BusMessage {
48    pub from: Option<Ulid>,
49    pub seq: u64,
50    pub payload: BusPayload,
51}
52
53#[derive(Debug, Clone, Serialize, Deserialize)]
54pub enum BusPayload {
55    Request(BusRequest),
56    Response(BusResponse),
57}
58
59#[derive(Debug, Clone, Serialize, Deserialize)]
60pub enum BusRequest {
61    /// Saludo. El Ente anuncia que está vivo y declara sus capacidades.
62    /// El Init usa esto para saber que el child arrancó correctamente,
63    /// independientemente de su exit code.
64    Announce { capabilities: Vec<Capability> },
65
66    /// Listar Entes vivos. Útil para debugging y para Entes-supervisor.
67    ListEntes,
68
69    /// Control de estado del fractal. Traducido desde D-Bus por compat-logind.
70    PowerOff { interactive: bool },
71    Reboot { interactive: bool },
72    Suspend { interactive: bool },
73    Hibernate { interactive: bool },
74
75    /// Invocación genérica de capacidad. `cap` debe estar provista por algún
76    /// Ente del grafo; el blob es el argumento opaco que el proveedor parsea.
77    Invoke { cap: Capability, blob: Vec<u8> },
78
79    /// Actualización dinámica del set de capacidades del Ente que llama.
80    /// Sólo aplicable al `from_authenticated` — un Ente sólo puede modificar
81    /// sus propias caps. La Card original (immutable) no se toca; la mutación
82    /// va al `dynamic_provides` del Incarnated.
83    UpdateCapabilities {
84        adds: Vec<Capability>,
85        removes: Vec<Capability>,
86    },
87}
88
89#[derive(Debug, Clone, Serialize, Deserialize)]
90pub enum BusResponse {
91    Ok,
92    Error(String),
93    Entes(Vec<EnteInfo>),
94    Invoked { result: Vec<u8> },
95}
96
97#[derive(Debug, Clone, Serialize, Deserialize)]
98pub struct EnteInfo {
99    pub id: Ulid,
100    pub label: String,
101    pub provides: Vec<Capability>,
102    pub pid: Option<i32>,
103}
104
105pub async fn write_frame<W: AsyncWriteExt + Unpin>(w: &mut W, msg: &BusMessage) -> anyhow::Result<()> {
106    let bytes = postcard::to_stdvec(msg)?;
107    if bytes.len() > MAX_FRAME {
108        anyhow::bail!("frame too large: {} > {}", bytes.len(), MAX_FRAME);
109    }
110    w.write_u32(bytes.len() as u32).await?;
111    w.write_all(&bytes).await?;
112    Ok(())
113}
114
115pub async fn read_frame<R: AsyncReadExt + Unpin>(r: &mut R) -> anyhow::Result<BusMessage> {
116    let len = r.read_u32().await? as usize;
117    if len > MAX_FRAME {
118        anyhow::bail!("frame oversize: {len}");
119    }
120    let mut buf = vec![0u8; len];
121    r.read_exact(&mut buf).await?;
122    Ok(postcard::from_bytes(&buf)?)
123}
124
125pub struct BusClient {
126    stream: UnixStream,
127    seq: u64,
128    self_id: Option<Ulid>,
129}
130
131/// Trait que un Ente proveedor implementa para servir invokes que el bus le
132/// forwarda. Sync por simplicidad — un handler async se cubre con
133/// `tokio::task::block_in_place` o un canal hacia un task externo.
134pub trait InvokeHandler {
135    fn handle(&mut self, cap: ente_card::Capability, blob: Vec<u8>) -> BusResponse;
136}
137
138/// Conexión long-lived para Entes que proveen capacidades. A diferencia de
139/// `BusClient` (request-response y desconecta), `BusServer`:
140///   1. Anuncia su identidad al bus
141///   2. Mantiene la conexión abierta
142///   3. Atiende invokes forwardeados por el bus en bucle
143pub struct BusServer {
144    stream: UnixStream,
145    self_id: Ulid,
146}
147
148impl BusServer {
149    pub async fn from_env() -> anyhow::Result<Self> {
150        let path = std::env::var(ENV_BUS_SOCK)
151            .map_err(|_| anyhow::anyhow!("{} no definido", ENV_BUS_SOCK))?;
152        let id_s = std::env::var(ENV_ENTE_ID)
153            .map_err(|_| anyhow::anyhow!("{} no definido", ENV_ENTE_ID))?;
154        let self_id = Ulid::from_str(&id_s)
155            .map_err(|_| anyhow::anyhow!("{} no es un Ulid válido: {id_s}", ENV_ENTE_ID))?;
156        let stream = UnixStream::connect(&path).await?;
157        Ok(Self { stream, self_id })
158    }
159
160    pub async fn announce(&mut self, capabilities: Vec<ente_card::Capability>) -> anyhow::Result<()> {
161        let req = BusMessage {
162            from: Some(self.self_id),
163            seq: 1,
164            payload: BusPayload::Request(BusRequest::Announce { capabilities }),
165        };
166        write_frame(&mut self.stream, &req).await?;
167        let resp = read_frame(&mut self.stream).await?;
168        match resp.payload {
169            BusPayload::Response(BusResponse::Ok) => Ok(()),
170            BusPayload::Response(other) => anyhow::bail!("Announce rechazado: {other:?}"),
171            BusPayload::Request(_) => anyhow::bail!("expected Response, got Request"),
172        }
173    }
174
175    /// Bucle principal del proveedor. Atiende invokes hasta que la conexión
176    /// se cierra (el bus muere o el Ente recibe SIGTERM y nosotros dropeamos).
177    pub async fn serve<H: InvokeHandler>(mut self, mut handler: H) -> anyhow::Result<()> {
178        loop {
179            let msg = read_frame(&mut self.stream).await?;
180            let resp = match msg.payload {
181                BusPayload::Request(BusRequest::Invoke { cap, blob }) => {
182                    handler.handle(cap, blob)
183                }
184                BusPayload::Request(other) => {
185                    BusResponse::Error(format!("BusServer no maneja {other:?}"))
186                }
187                BusPayload::Response(_) => continue,
188            };
189            let out = BusMessage {
190                from: Some(self.self_id),
191                seq: msg.seq,
192                payload: BusPayload::Response(resp),
193            };
194            write_frame(&mut self.stream, &out).await?;
195        }
196    }
197}
198
199impl BusClient {
200    pub async fn connect(path: impl AsRef<Path>) -> anyhow::Result<Self> {
201        let stream = UnixStream::connect(path).await?;
202        let self_id = std::env::var(ENV_ENTE_ID)
203            .ok()
204            .and_then(|s| Ulid::from_str(&s).ok());
205        Ok(Self { stream, seq: 0, self_id })
206    }
207
208    pub async fn from_env() -> anyhow::Result<Self> {
209        let path = std::env::var(ENV_BUS_SOCK)
210            .map_err(|_| anyhow::anyhow!("{} no definido", ENV_BUS_SOCK))?;
211        Self::connect(&path).await
212    }
213
214    pub async fn call(&mut self, req: BusRequest) -> anyhow::Result<BusResponse> {
215        self.seq = self.seq.wrapping_add(1);
216        let msg = BusMessage {
217            from: self.self_id,
218            seq: self.seq,
219            payload: BusPayload::Request(req),
220        };
221        write_frame(&mut self.stream, &msg).await?;
222        let resp = read_frame(&mut self.stream).await?;
223        match resp.payload {
224            BusPayload::Response(r) => Ok(r),
225            BusPayload::Request(_) => anyhow::bail!("expected response, got request"),
226        }
227    }
228}