⎇ 53dbdf0f1deea10d6fc567ea2b43e1e75d6d60cd
crates/modules/semantic_dht/minga-p2p/src/session.rs
Historia
Quién tocó qué
crudo
chore: monorepo inicial con arje + minga + yahweh absorbidos
1//! Máquina de estados de sincronización recursiva sobre la estructura
2//! del MST, con verificación criptográfica de cada nodo entregado.
3//!
4//! La sesión es **pura**: no hace IO, no toca la red, no usa async. El
5//! transporte la alimenta vía `handle(msg)` y consume sus salidas como
6//! `Vec<Message>`.
7//!
8//! ## Algoritmo
9//!
10//! 1. Cada peer construye al inicio un `own_probes: HashMap<ContentHash,
11//! NodeProbe>` que indexa cada nodo interno de su MST por su hash
12//! Merkle de subárbol. Es la tabla con la que respondemos
13//! `ProbeReq`s en `O(1)`.
14//!
15//! 2. Cada peer envía `Hello` con el hash de su raíz. Si el peer
16//! contrario reconoce ese hash en su propio `own_probes` (o coincide
17//! con su propia raíz, o es la raíz vacía), no hay nada estructural
18//! que descubrir — la rama está ya alineada.
19//!
20//! 3. Si el hash no se reconoce, el peer emite un `ProbeReq` para
21//! pedirle al otro la estructura de ese subárbol. Cuando llega el
22//! `ProbeRes`, el peer:
23//! - Para cada **clave** del probe que no tiene en su MST, programa
24//! un `Fetch` (la clave entrará al MST cuando llegue su `Deliver`).
25//! - Para cada **child_hash** del probe que no aparece en
26//! `own_probes`, recurre con un nuevo `ProbeReq`. Si el child_hash
27//! ya está en `own_probes`, la rama se poda — toda esa subestructura
28//! es idéntica a la nuestra.
29//!
30//! 4. Cuando un peer recibe un `Deliver`, verifica que el hash
31//! anunciado coincida con el `hash_stored` real del nodo. Si no,
32//! descarta. Si sí, inserta en el `MemStore` y, si el hash venía de
33//! la raíz del MST del peer (no de un descendiente), también lo
34//! inserta en su MST.
35//!
36//! 5. Cada `StoredNode` recibido contiene los hashes de sus hijos. Si
37//! el receptor no los tiene, los pide vía `Fetch` (sync transitivo).
38//!
39//! 6. Un peer envía `Done` cuando: emitió y recibió `Hello`, no tiene
40//! probes pendientes, ni fetches pendientes (raíz o hijo). La sesión
41//! cierra cuando ambos `Done`s han cruzado.
42
43use minga_core::{
44 cas::ContentHash, empty_subtree_hash, hash_stored, AttestationStore, Did, Keypair, MemStore,
45 Mst, NodeProbe, NodeStore,
46};
47use rand::rngs::OsRng;
48use rand::RngCore;
49use std::collections::{HashMap, HashSet};
50
51use crate::message::Message;
52
53/// Construye el payload firmado del `Hello` con orden fijo:
54/// `verifier_nonce(32) || peer_did(32) || root_subtree_hash(32) = 96 bytes`.
55/// El `verifier_nonce` es el nonce que emitió el peer que verificará
56/// la firma; al firmar sobre él se vincula la firma a esta sesión.
57/// Cualquier cambio al formato es incompatible al protocolo.
58pub(crate) fn hello_payload(
59 verifier_nonce: &[u8; 32],
60 did: &Did,
61 root: &ContentHash,
62) -> [u8; 96] {
63 let mut p = [0u8; 96];
64 p[..32].copy_from_slice(verifier_nonce);
65 p[32..64].copy_from_slice(&did.0);
66 p[64..].copy_from_slice(&root.0);
67 p
68}
69
70pub struct SyncSession {
71 mst: Mst,
72 store: MemStore,
73 attestations: AttestationStore,
74
75 /// Llave del peer local: firma el `Hello` y queda asociada al
76 /// `Did` que el peer remoto verá.
77 keypair: Keypair,
78
79 /// Identidad del peer remoto, capturada tras verificar la firma
80 /// de su `Hello`.
81 peer_did: Option<Did>,
82
83 own_probes: HashMap<ContentHash, NodeProbe>,
84 own_root_subtree_hash: ContentHash,
85
86 awaited_probes: HashSet<ContentHash>,
87 seen_probes: HashSet<ContentHash>,
88 awaiting_root: HashSet<ContentHash>,
89 awaiting_child: HashSet<ContentHash>,
90
91 rejected_hellos: usize,
92 rejected_delivers: usize,
93 /// Contador de atestaciones rechazadas: firma rota, llegada antes
94 /// de autenticar al peer, o cualquier otra inconsistencia que el
95 /// `AttestationStore` rechace.
96 rejected_attests: usize,
97
98 /// Nonce aleatorio que **nosotros** emitimos en `Challenge`. La
99 /// firma del `Hello` del peer debe ser sobre este nonce.
100 self_nonce: [u8; 32],
101 /// Nonce que el peer publicó en su `Challenge` — sobre este
102 /// nonce firmamos nosotros nuestro `Hello`.
103 peer_nonce: Option<[u8; 32]>,
104
105 sent_challenge: bool,
106 received_challenge: bool,
107 sent_hello: bool,
108 received_hello: bool,
109 sent_attestations: bool,
110 sent_done: bool,
111 received_done: bool,
112}
113
114impl SyncSession {
115 pub fn new(
116 mst: Mst,
117 store: MemStore,
118 attestations: AttestationStore,
119 keypair: Keypair,
120 ) -> Self {
121 let own_probes = mst.build_probe_index();
122 let own_root_subtree_hash = mst.root_hash();
123 let mut self_nonce = [0u8; 32];
124 OsRng.fill_bytes(&mut self_nonce);
125 Self {
126 mst,
127 store,
128 attestations,
129 keypair,
130 peer_did: None,
131 own_probes,
132 own_root_subtree_hash,
133 awaited_probes: HashSet::new(),
134 seen_probes: HashSet::new(),
135 awaiting_root: HashSet::new(),
136 awaiting_child: HashSet::new(),
137 rejected_hellos: 0,
138 rejected_delivers: 0,
139 rejected_attests: 0,
140 self_nonce,
141 peer_nonce: None,
142 sent_challenge: false,
143 received_challenge: false,
144 sent_hello: false,
145 received_hello: false,
146 sent_attestations: false,
147 sent_done: false,
148 received_done: false,
149 }
150 }
151
152 /// Conveniencia para sesiones sin atestaciones previas. Equivalente
153 /// a `new(mst, store, AttestationStore::new(), keypair)`.
154 pub fn without_attestations(mst: Mst, store: MemStore, keypair: Keypair) -> Self {
155 Self::new(mst, store, AttestationStore::new(), keypair)
156 }
157
158 /// Mensaje inicial: `Challenge` con un nonce aleatorio. El `Hello`
159 /// y las atestaciones llegarán como respuesta al `Challenge` del
160 /// otro peer (cuando lo recibamos, ya tendremos su nonce sobre el
161 /// que firmar nuestra identidad).
162 pub fn start(&mut self) -> Vec<Message> {
163 if self.sent_challenge {
164 return Vec::new();
165 }
166 self.sent_challenge = true;
167 let mut out = vec![Message::Challenge {
168 nonce: self.self_nonce,
169 }];
170 out.extend(self.maybe_done());
171 out
172 }
173
174 pub fn handle(&mut self, msg: Message) -> Vec<Message> {
175 let mut out = Vec::new();
176 match msg {
177 Message::Challenge { nonce } => {
178 if self.received_challenge {
179 // Challenge duplicado: ignoramos. Un peer
180 // legítimo no debería enviar dos.
181 return out;
182 }
183 self.received_challenge = true;
184 self.peer_nonce = Some(nonce);
185
186 // Ahora podemos firmar nuestro Hello sobre el nonce
187 // del peer — lo que ata la firma a esta sesión.
188 let payload =
189 hello_payload(&nonce, &self.keypair.did(), &self.own_root_subtree_hash);
190 let signature = self.keypair.sign(&payload);
191 self.sent_hello = true;
192 out.push(Message::Hello {
193 peer_did: self.keypair.did(),
194 root_subtree_hash: self.own_root_subtree_hash,
195 signature,
196 });
197
198 // Empuje de atestaciones: el peer ya nos verificará
199 // como remitente cuando reciba nuestro Hello.
200 let atts: Vec<_> = self.attestations.all().cloned().collect();
201 if !atts.is_empty() {
202 out.push(Message::AttestPush { attestations: atts });
203 }
204 self.sent_attestations = true;
205 }
206
207 Message::Hello {
208 peer_did,
209 root_subtree_hash,
210 signature,
211 } => {
212 // ── Autenticación del peer + anti-replay ─────────
213 // La firma debe ser sobre nuestro `self_nonce` (que
214 // emitimos en nuestro Challenge), atándola a esta
215 // sesión. Un Hello capturado de otra sesión tendría
216 // un nonce distinto y la verificación fallaría.
217 let payload = hello_payload(&self.self_nonce, &peer_did, &root_subtree_hash);
218 if !peer_did.verify(&payload, &signature) {
219 self.rejected_hellos += 1;
220 return out;
221 }
222 self.peer_did = Some(peer_did);
223 self.received_hello = true;
224 if self.should_probe(&root_subtree_hash) {
225 self.awaited_probes.insert(root_subtree_hash);
226 out.push(Message::ProbeReq {
227 subtree_hash: root_subtree_hash,
228 });
229 }
230 }
231
232 Message::ProbeReq { subtree_hash } => {
233 let probe = self.own_probes.get(&subtree_hash).cloned();
234 // Si el subárbol pedido era vacío (o desconocido para
235 // nosotros), respondemos con `None` — el peer lo
236 // tratará como un punto sin descendientes que descubrir.
237 out.push(Message::ProbeRes {
238 subtree_hash,
239 probe,
240 });
241 }
242
243 Message::ProbeRes {
244 subtree_hash,
245 probe,
246 } => {
247 self.awaited_probes.remove(&subtree_hash);
248 self.seen_probes.insert(subtree_hash);
249 if let Some(probe) = probe {
250 out.extend(self.process_probe(&probe));
251 }
252 }
253
254 Message::Fetch { hash } => {
255 if let Some(stored) = self.store.get(&hash).cloned() {
256 out.push(Message::Deliver { hash, stored });
257 }
258 // Si no lo tenemos, callamos. El peer no debería estar
259 // pidiéndonos algo que no le hayamos anunciado.
260 }
261
262 Message::Deliver { hash, stored } => {
263 // ── Verificación criptográfica ────────────────────
264 // Recomputamos el hash del nodo entregado a partir de
265 // sus componentes. Si no coincide con el anunciado,
266 // alguien (peer malicioso o ruido en transporte) está
267 // intentando colar contenido distinto bajo un hash que
268 // no le corresponde. Descartamos silenciosamente y
269 // contamos para diagnóstico.
270 if hash_stored(&stored) != hash {
271 self.rejected_delivers += 1;
272 // No tocamos awaiting_*: la solicitud sigue
273 // pendiente y el peer (legítimo o no) puede
274 // reintentarla.
275 return out;
276 }
277
278 let was_root = self.awaiting_root.remove(&hash);
279 self.awaiting_child.remove(&hash);
280
281 // Antes de mover `stored`, descubrimos qué hijos
282 // faltan y los pedimos.
283 let mut new_fetches = Vec::new();
284 for ch in &stored.children {
285 if !self.store.contains(ch)
286 && !self.awaiting_root.contains(ch)
287 && !self.awaiting_child.contains(ch)
288 {
289 self.awaiting_child.insert(*ch);
290 new_fetches.push(*ch);
291 }
292 }
293
294 self.store.put_chunked(hash, stored);
295 if was_root {
296 self.mst.insert(hash);
297 }
298
299 for h in new_fetches {
300 out.push(Message::Fetch { hash: h });
301 }
302 }
303
304 Message::AttestPush { attestations } => {
305 // Antes de procesar atestaciones del peer, exigimos
306 // haber autenticado su identidad. Un push antes del
307 // `Hello` es protocolo malformado o ataque — todas las
308 // atestaciones se cuentan como rechazadas.
309 if !self.received_hello {
310 self.rejected_attests += attestations.len();
311 return out;
312 }
313 for att in attestations {
314 // `AttestationStore::add` re-verifica cada firma.
315 // Una sola atestación corrupta no contamina las
316 // demás del lote.
317 if self.attestations.add(att).is_err() {
318 self.rejected_attests += 1;
319 }
320 }
321 }
322
323 Message::Done => {
324 self.received_done = true;
325 }
326 }
327 out.extend(self.maybe_done());
328 out
329 }
330
331 fn process_probe(&mut self, probe: &NodeProbe) -> Vec<Message> {
332 let mut out = Vec::new();
333
334 // Cada clave del probe que no tenemos pasa a `awaiting_root` y
335 // generamos un Fetch. Si ya está en el store (sin estar aún en
336 // el MST), simplemente la promovemos al MST sin pedirla.
337 for k in &probe.keys {
338 if self.mst.contains(k) {
339 continue;
340 }
341 if self.store.contains(k) {
342 self.mst.insert(*k);
343 continue;
344 }
345 if self.awaiting_root.contains(k) {
346 continue;
347 }
348 self.awaiting_root.insert(*k);
349 out.push(Message::Fetch { hash: *k });
350 }
351
352 // Para cada subárbol hijo, decidimos si recurrir o podar:
353 // - el vacío se reconoce por hash sin red,
354 // - los que ya tenemos en `own_probes` (igualdad de hash =
355 // subestructura idéntica) se podan,
356 // - los ya vistos o solicitados no se duplican,
357 // - el resto dispara un `ProbeReq` recursivo.
358 for ch in &probe.child_hashes {
359 if self.should_probe(ch) {
360 self.awaited_probes.insert(*ch);
361 out.push(Message::ProbeReq { subtree_hash: *ch });
362 }
363 }
364
365 out
366 }
367
368 /// Decide si vale la pena solicitar un probe sobre `h`. Cuatro
369 /// razones para NO pedirlo:
370 /// - es el subárbol vacío (lo conocemos por convención),
371 /// - coincide con nuestra propia raíz (igualdad estructural),
372 /// - aparece en `own_probes` (ya tenemos un subárbol idéntico),
373 /// - ya lo solicitamos o ya lo recibimos.
374 fn should_probe(&self, h: &ContentHash) -> bool {
375 if *h == empty_subtree_hash() {
376 return false;
377 }
378 if *h == self.own_root_subtree_hash {
379 return false;
380 }
381 if self.own_probes.contains_key(h) {
382 return false;
383 }
384 if self.awaited_probes.contains(h) || self.seen_probes.contains(h) {
385 return false;
386 }
387 true
388 }
389
390 fn maybe_done(&mut self) -> Vec<Message> {
391 if self.sent_done {
392 return Vec::new();
393 }
394 if !self.sent_challenge || !self.received_challenge {
395 return Vec::new();
396 }
397 if !self.sent_hello || !self.received_hello {
398 return Vec::new();
399 }
400 if !self.sent_attestations {
401 return Vec::new();
402 }
403 if !self.awaited_probes.is_empty() {
404 return Vec::new();
405 }
406 if !self.awaiting_root.is_empty() || !self.awaiting_child.is_empty() {
407 return Vec::new();
408 }
409 self.sent_done = true;
410 vec![Message::Done]
411 }
412
413 pub fn is_done(&self) -> bool {
414 self.sent_done && self.received_done
415 }
416
417 pub fn rejected_delivers(&self) -> usize {
418 self.rejected_delivers
419 }
420
421 pub fn rejected_hellos(&self) -> usize {
422 self.rejected_hellos
423 }
424
425 pub fn rejected_attests(&self) -> usize {
426 self.rejected_attests
427 }
428
429 pub fn attestations(&self) -> &AttestationStore {
430 &self.attestations
431 }
432
433 /// Identidad del peer remoto, capturada tras verificar su `Hello`.
434 /// `None` si todavía no llegó un `Hello` válido.
435 pub fn peer_did(&self) -> Option<Did> {
436 self.peer_did
437 }
438
439 pub fn local_did(&self) -> Did {
440 self.keypair.did()
441 }
442
443 /// Nonce aleatorio que esta sesión emitió en su `Challenge`.
444 /// Expuesto principalmente para tests y debugging — el nonce
445 /// viaja en claro por el wire y no es secreto.
446 pub fn self_nonce(&self) -> [u8; 32] {
447 self.self_nonce
448 }
449
450 pub fn mst(&self) -> &Mst {
451 &self.mst
452 }
453
454 pub fn store(&self) -> &MemStore {
455 &self.store
456 }
457
458 pub fn into_parts(self) -> (Mst, MemStore, AttestationStore) {
459 (self.mst, self.store, self.attestations)
460 }
461}