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/// - `Announce`: establece bus_connections para forwarding.
20/// - `UpdateCapabilities`: muta dynamic_provides del Ente — sólo el dueño.
21///
22/// Invoke, ListEntes y power-mgmt se aceptan anonymous — políticas por
23/// capacidad se aplican aguas abajo, no aquí.
24fn requires_auth(req: &BusRequest) -> bool {
25    matches!(
26        req,
27        BusRequest::Announce { .. } | BusRequest::UpdateCapabilities { .. }
28    )
29}
30
31impl EnteGraph {
32    pub async fn on_bus_request(
33        &mut self,
34        peer: PeerCreds,
35        from: Option<Ulid>,
36        request: BusRequest,
37        outbound: mpsc::Sender<BusMessage>,
38        reply: oneshot::Sender<BusResponse>,
39    ) {
40        // ---- Auth: kernel-injected SO_PEERCRED vs identidad reclamada ----
41        let from_authenticated = match from {
42            None => None,
43            Some(claimed) => {
44                let expected = self.incarnated.get(&claimed).and_then(|i| i.pid);
45                match expected {
46                    Some(p) if p.as_raw() == peer.pid => Some(claimed),
47                    Some(p) => {
48                        warn!(
49                            claimed = %claimed, expected_pid = p.as_raw(),
50                            actual_pid = peer.pid,
51                            "identity mismatch — rechazando request"
52                        );
53                        let _ = reply.send(BusResponse::Error("identity mismatch".into()));
54                        return;
55                    }
56                    None => {
57                        warn!(?claimed, peer_pid = peer.pid, "Ente desconocido reclamando identidad");
58                        let _ = reply.send(BusResponse::Error("unknown ente claimed".into()));
59                        return;
60                    }
61                }
62            }
63        };
64        if requires_auth(&request) && from_authenticated.is_none() {
65            let _ = reply.send(BusResponse::Error("auth required for this request".into()));
66            return;
67        }
68
69        // ---- Dispatch ----
70        match request {
71            BusRequest::Announce { capabilities } => {
72                let id = from_authenticated.expect("auth-required guarantees Some");
73                let label = self.incarnated.get(&id).map(|i| i.card.label.clone())
74                    .unwrap_or_else(|| "anónimo".into());
75                info!(%id, %label, ?capabilities, peer_pid = peer.pid, "Announce autenticado");
76                self.bus_connections.insert(id, outbound);
77                let _ = reply.send(BusResponse::Ok);
78            }
79            BusRequest::ListEntes => {
80                let entes = self.incarnated.values()
81                    .map(|i| EnteInfo {
82                        id: i.card.id,
83                        label: i.card.label.clone(),
84                        provides: i.card.provides.iter().cloned().collect(),
85                        pid: i.pid.map(|p| p.as_raw()),
86                    })
87                    .collect();
88                let _ = reply.send(BusResponse::Entes(entes));
89            }
90            BusRequest::PowerOff { interactive } => {
91                info!(?from_authenticated, interactive, peer_pid = peer.pid, "PowerOff via bus");
92                let _ = reply.send(BusResponse::Ok);
93            }
94            BusRequest::Reboot { interactive } => {
95                info!(?from_authenticated, interactive, "Reboot via bus");
96                let _ = reply.send(BusResponse::Ok);
97            }
98            BusRequest::Suspend { interactive } => {
99                info!(?from_authenticated, interactive, "Suspend via bus");
100                let _ = reply.send(BusResponse::Ok);
101            }
102            BusRequest::Hibernate { interactive } => {
103                info!(?from_authenticated, interactive, "Hibernate via bus");
104                let _ = reply.send(BusResponse::Ok);
105            }
106            BusRequest::Invoke { cap, blob } => {
107                self.forward_invoke(from_authenticated, cap, blob, reply).await;
108            }
109            BusRequest::UpdateCapabilities { adds, removes } => {
110                let id = from_authenticated.expect("auth-required guarantees Some");
111                self.apply_capability_update(id, adds, removes);
112                let _ = reply.send(BusResponse::Ok);
113            }
114        }
115    }
116
117    /// Muta `dynamic_provides` del Ente y actualiza el índice global de
118    /// providers. La Card original (immutable) no se toca.
119    fn apply_capability_update(
120        &mut self,
121        ente_id: Ulid,
122        adds: Vec<Capability>,
123        removes: Vec<Capability>,
124    ) {
125        // Adiciones: dedupe contra Card.provides + dynamic_provides existentes.
126        let mut added = Vec::new();
127        let mut removed = Vec::new();
128        if let Some(inc) = self.incarnated.get_mut(&ente_id) {
129            for cap in adds {
130                if inc.card.provides.contains(&cap) || inc.dynamic_provides.contains(&cap) {
131                    continue; // ya provista, no-op
132                }
133                inc.dynamic_provides.insert(cap.clone());
134                added.push(cap);
135            }
136            for cap in removes {
137                if inc.dynamic_provides.remove(&cap) {
138                    removed.push(cap);
139                }
140                // Caps de la Card original no se pueden quitar — silenciosamente
141                // ignoradas. Una Card es contrato; sólo el dynamic es mutable.
142            }
143        }
144        // Actualizar índice global. Hacemos esto fuera del scope `inc` para
145        // evitar el doble-borrow de self.
146        for cap in &added {
147            self.register_dynamic_cap(ente_id, cap.clone());
148        }
149        for cap in &removed {
150            self.unregister_dynamic_cap(ente_id, cap);
151            // Revocar grants emitidos contra esta cap por este Ente.
152            let revoked: Vec<u64> = self.grants.iter()
153                .filter(|(_, g)| g.provider == ente_id && &g.cap == cap)
154                .map(|(t, _)| *t)
155                .collect();
156            for t in revoked {
157                self.grants.remove(&t);
158            }
159        }
160        info!(
161            %ente_id,
162            added_count = added.len(),
163            removed_count = removed.len(),
164            "capabilities actualizadas en runtime"
165        );
166    }
167
168    /// Enruta un Invoke al proveedor real de la capacidad. Aloca un seq
169    /// server-side, registra el reply oneshot en `pending_invokes`, y empuja
170    /// el request por la conexión del proveedor.
171    async fn forward_invoke(
172        &mut self,
173        from: Option<Ulid>,
174        cap: Capability,
175        blob: Vec<u8>,
176        reply: oneshot::Sender<BusResponse>,
177    ) {
178        let provider_id = match self.pick_invokable_provider(&cap) {
179            Some(id) => id,
180            None => {
181                let _ = reply.send(BusResponse::Error(format!("sin proveedor invokable para {cap:?}")));
182                return;
183            }
184        };
185        let outbound = match self.bus_connections.get(&provider_id) {
186            Some(o) => o.clone(),
187            None => {
188                let _ = reply.send(BusResponse::Error("proveedor no conectado al bus".into()));
189                return;
190            }
191        };
192        let seq = self.alloc_invoke_seq();
193        self.pending_invokes.insert(seq, reply);
194        debug!(?from, ?cap, ?provider_id, seq, blob_len = blob.len(), "forwardeando Invoke");
195
196        let msg = BusMessage {
197            from: None,
198            seq,
199            payload: BusPayload::Request(BusRequest::Invoke { cap, blob }),
200        };
201        if outbound.send(msg).await.is_err() {
202            if let Some(orig) = self.pending_invokes.remove(&seq) {
203                let _ = orig.send(BusResponse::Error("conn del proveedor cerrada".into()));
204            }
205        }
206    }
207
208    fn pick_invokable_provider(&self, cap: &Capability) -> Option<Ulid> {
209        // Sólo proveedores con conexión al bus pueden recibir forwards.
210        // El propio Ente #0 está en `providers` para varias caps pero no
211        // debe recibir forwards — se filtra implícitamente porque la Semilla
212        // no tiene conexión al bus.
213        self.providers.get(cap)?
214            .iter()
215            .find(|id| self.bus_connections.contains_key(id))
216            .copied()
217    }
218
219    pub(in crate::graph) fn alloc_invoke_seq(&mut self) -> u64 {
220        self.next_invoke_seq = self.next_invoke_seq.wrapping_add(1);
221        SERVER_SEQ_FLAG | self.next_invoke_seq
222    }
223
224    pub async fn on_bus_response(&mut self, seq: u64, response: BusResponse) {
225        if let Some(orig) = self.pending_invokes.remove(&seq) {
226            let _ = orig.send(response);
227        } else {
228            warn!(seq, "Response sin pending invoke");
229        }
230    }
231
232    pub async fn on_bus_conn_closed(&mut self, ente_id: Option<Ulid>) {
233        if let Some(id) = ente_id {
234            self.bus_connections.remove(&id);
235            // No revocamos providers — la capacidad sigue declarada en su
236            // Card. Sólo perdimos el canal de invocación.
237            debug!(%id, "bus connection cerrada");
238        }
239    }
240}