tawasuyu/arje

1//! Bus mediator: integración de `EnteGraph` con el bus interno.
2//!
3//! Responsabilidades:
4//!   - Auth de Announce (verificar identidad reclamada contra SO_PEERCRED)
5//!   - Registro de conexiones (`bus_connections` indexado por Ulid)
6//!   - Forwarding de Invokes a proveedores
7//!   - Tracking de invokes en vuelo (`pending_invokes` por seq)
8//!   - Cleanup en cierre de conexión
9
10use super::{EnteGraph, SERVER_SEQ_FLAG};
11use ente_bus::{BusMessage, BusPayload, BusRequest, BusResponse, EnteInfo, PeerCreds};
12use ente_card::Capability;
13use tokio::sync::{mpsc, oneshot};
14use tracing::{debug, info, warn};
15use ulid::Ulid;
16
17/// Operaciones que requieren identidad verificada en el bus.
18///
19/// Sólo `Announce`: establece la entrada en `bus_connections` para forwarding
20/// y debe ser no-falsificable. Invoke, ListEntes y power-mgmt se aceptan
21/// anonymous — políticas por capacidad se aplican aguas abajo, no aquí.
22fn requires_auth(req: &BusRequest) -> bool {
23    matches!(req, BusRequest::Announce { .. })
24}
25
26impl EnteGraph {
27    pub async fn on_bus_request(
28        &mut self,
29        peer: PeerCreds,
30        from: Option<Ulid>,
31        request: BusRequest,
32        outbound: mpsc::Sender<BusMessage>,
33        reply: oneshot::Sender<BusResponse>,
34    ) {
35        // ---- Auth: kernel-injected SO_PEERCRED vs identidad reclamada ----
36        let from_authenticated = match from {
37            None => None,
38            Some(claimed) => {
39                let expected = self.incarnated.get(&claimed).and_then(|i| i.pid);
40                match expected {
41                    Some(p) if p.as_raw() == peer.pid => Some(claimed),
42                    Some(p) => {
43                        warn!(
44                            claimed = %claimed, expected_pid = p.as_raw(),
45                            actual_pid = peer.pid,
46                            "identity mismatch — rechazando request"
47                        );
48                        let _ = reply.send(BusResponse::Error("identity mismatch".into()));
49                        return;
50                    }
51                    None => {
52                        warn!(?claimed, peer_pid = peer.pid, "Ente desconocido reclamando identidad");
53                        let _ = reply.send(BusResponse::Error("unknown ente claimed".into()));
54                        return;
55                    }
56                }
57            }
58        };
59        if requires_auth(&request) && from_authenticated.is_none() {
60            let _ = reply.send(BusResponse::Error("auth required for this request".into()));
61            return;
62        }
63
64        // ---- Dispatch ----
65        match request {
66            BusRequest::Announce { capabilities } => {
67                let id = from_authenticated.expect("auth-required guarantees Some");
68                let label = self.incarnated.get(&id).map(|i| i.card.label.clone())
69                    .unwrap_or_else(|| "anónimo".into());
70                info!(%id, %label, ?capabilities, peer_pid = peer.pid, "Announce autenticado");
71                self.bus_connections.insert(id, outbound);
72                let _ = reply.send(BusResponse::Ok);
73            }
74            BusRequest::ListEntes => {
75                let entes = self.incarnated.values()
76                    .map(|i| EnteInfo {
77                        id: i.card.id,
78                        label: i.card.label.clone(),
79                        provides: i.card.provides.iter().cloned().collect(),
80                        pid: i.pid.map(|p| p.as_raw()),
81                    })
82                    .collect();
83                let _ = reply.send(BusResponse::Entes(entes));
84            }
85            BusRequest::PowerOff { interactive } => {
86                info!(?from_authenticated, interactive, peer_pid = peer.pid, "PowerOff via bus");
87                let _ = reply.send(BusResponse::Ok);
88            }
89            BusRequest::Reboot { interactive } => {
90                info!(?from_authenticated, interactive, "Reboot via bus");
91                let _ = reply.send(BusResponse::Ok);
92            }
93            BusRequest::Suspend { interactive } => {
94                info!(?from_authenticated, interactive, "Suspend via bus");
95                let _ = reply.send(BusResponse::Ok);
96            }
97            BusRequest::Hibernate { interactive } => {
98                info!(?from_authenticated, interactive, "Hibernate via bus");
99                let _ = reply.send(BusResponse::Ok);
100            }
101            BusRequest::Invoke { cap, blob } => {
102                self.forward_invoke(from_authenticated, cap, blob, reply).await;
103            }
104        }
105    }
106
107    /// Enruta un Invoke al proveedor real de la capacidad. Aloca un seq
108    /// server-side, registra el reply oneshot en `pending_invokes`, y empuja
109    /// el request por la conexión del proveedor.
110    async fn forward_invoke(
111        &mut self,
112        from: Option<Ulid>,
113        cap: Capability,
114        blob: Vec<u8>,
115        reply: oneshot::Sender<BusResponse>,
116    ) {
117        let provider_id = match self.pick_invokable_provider(&cap) {
118            Some(id) => id,
119            None => {
120                let _ = reply.send(BusResponse::Error(format!("sin proveedor invokable para {cap:?}")));
121                return;
122            }
123        };
124        let outbound = match self.bus_connections.get(&provider_id) {
125            Some(o) => o.clone(),
126            None => {
127                let _ = reply.send(BusResponse::Error("proveedor no conectado al bus".into()));
128                return;
129            }
130        };
131        let seq = self.alloc_invoke_seq();
132        self.pending_invokes.insert(seq, reply);
133        debug!(?from, ?cap, ?provider_id, seq, blob_len = blob.len(), "forwardeando Invoke");
134
135        let msg = BusMessage {
136            from: None,
137            seq,
138            payload: BusPayload::Request(BusRequest::Invoke { cap, blob }),
139        };
140        if outbound.send(msg).await.is_err() {
141            if let Some(orig) = self.pending_invokes.remove(&seq) {
142                let _ = orig.send(BusResponse::Error("conn del proveedor cerrada".into()));
143            }
144        }
145    }
146
147    fn pick_invokable_provider(&self, cap: &Capability) -> Option<Ulid> {
148        // Sólo proveedores con conexión al bus pueden recibir forwards.
149        // El propio Ente #0 está en `providers` para varias caps pero no
150        // debe recibir forwards — se filtra implícitamente porque la Semilla
151        // no tiene conexión al bus.
152        self.providers.get(cap)?
153            .iter()
154            .find(|id| self.bus_connections.contains_key(id))
155            .copied()
156    }
157
158    pub(in crate::graph) fn alloc_invoke_seq(&mut self) -> u64 {
159        self.next_invoke_seq = self.next_invoke_seq.wrapping_add(1);
160        SERVER_SEQ_FLAG | self.next_invoke_seq
161    }
162
163    pub async fn on_bus_response(&mut self, seq: u64, response: BusResponse) {
164        if let Some(orig) = self.pending_invokes.remove(&seq) {
165            let _ = orig.send(response);
166        } else {
167            warn!(seq, "Response sin pending invoke");
168        }
169    }
170
171    pub async fn on_bus_conn_closed(&mut self, ente_id: Option<Ulid>) {
172        if let Some(id) = ente_id {
173            self.bus_connections.remove(&id);
174            // No revocamos providers — la capacidad sigue declarada en su
175            // Card. Sólo perdimos el canal de invocación.
176            debug!(%id, "bus connection cerrada");
177        }
178    }
179}