refactor(monorepo): reorganización lógica + renames + SDDs + split CHANGELOG
1//! `brahman-net` — capa P2P compartida de la red Brahman.
2//!
3//! Provee un nodo libp2p genérico que cualquier protocolo de la
4//! familia (handshake brahman remoto, sync minga, futuros) puede
5//! reusar. La idea: una sola malla, múltiples sub-protocolos
6//! multiplexados por `StreamProtocol`.
7//!
8//! ## Stack
9//!
10//! - **TCP + Noise + Yamux**: transporte autenticado y multiplexado.
11//! - **`stream::Behaviour`**: streams bidireccionales por
12//! `StreamProtocol`. Cada protocolo (`/brahman/handshake/1.0.0`,
13//! `/minga/sync/1.0.0`, …) se registra independientemente vía el
14//! `stream::Control` que `BrahmanNet` expone.
15//! - **`kad::Behaviour<MemoryStore>`**: Kademlia DHT en modo Server
16//! para discovery (peers cercanos + content providers).
17//! - **`identify::Behaviour`**: cada peer anuncia sus listen-addrs
18//! reales; las inyectamos automáticamente al routing table de Kad.
19//!
20//! ## Modelo
21//!
22//! El swarm corre en una task tokio dedicada. La interfaz pública son:
23//! 1. **Comandos** (canal mpsc): `dial`, `listen`, `add_dht_peer`,
24//! `find_closest_peers`, `start_providing`, `find_providers`.
25//! 2. **`stream::Control`** (acceso directo): para abrir/aceptar
26//! streams de un protocolo concreto. Cada protocolo se ocupa de
27//! su propia lógica sobre el stream resultante.
28//!
29//! La separación entre comandos y control permite que la lógica de
30//! red (DHT, dial, listen) y la lógica de protocolos (handshake/sync)
31//! evolucionen independientes — el protocolo no necesita conocer al
32//! swarm, sólo pide streams.
33//!
34//! ## Identidad
35//!
36//! Por defecto se genera una keypair Ed25519 efímera. Para identidad
37//! persistente (la misma `peer_id` across reboots), pasar la keypair
38//! con [`BrahmanNet::with_keypair`]. Esa misma keypair puede ser la
39//! base para firmas de Cards (cuando se implemente trust remoto).
40
41#![forbid(unsafe_code)]
42#![warn(rust_2018_idioms)]
43
44use std::collections::HashMap;
45use std::sync::Arc;
46use std::time::Duration;
47
48use futures::StreamExt;
49use libp2p::{
50 identify, identity, kad, noise,
51 swarm::{NetworkBehaviour, SwarmEvent},
52 tcp, yamux, Swarm, SwarmBuilder,
53};
54use libp2p_allow_block_list::{self as allow_block_list, BlockedPeers};
55use libp2p_stream as stream;
56use tokio::sync::{mpsc, oneshot, Mutex};
57
58pub use libp2p::{
59 identity::{Keypair, PublicKey},
60 multiaddr::Protocol,
61 Multiaddr, PeerId, Stream, StreamProtocol,
62};
63pub use libp2p_stream::OpenStreamError;
64
65const IDENTIFY_PROTOCOL: &str = "/brahman-net/0.1.0";
66const IDLE_CONNECTION_TIMEOUT: Duration = Duration::from_secs(60);
67
68#[derive(NetworkBehaviour)]
69struct BrahmanBehaviour {
70 /// Block-list a nivel de swarm: peers en este behaviour son
71 /// rechazados ANTES del handshake Noise. Más eficiente que
72 /// rechazar al nivel del handshake brahman (ahorra round-trip
73 /// TCP+Noise por intento denegado). Sincronizado con la
74 /// `PeerPolicy.deny` vía `block_peer`/`unblock_peer` exposed
75 /// en `BrahmanNet`.
76 block_list: allow_block_list::Behaviour<BlockedPeers>,
77 stream: stream::Behaviour,
78 kad: kad::Behaviour<kad::store::MemoryStore>,
79 identify: identify::Behaviour,
80}
81
82#[derive(Debug, thiserror::Error)]
83pub enum NodeError {
84 #[error("transport build failed: {0}")]
85 Build(String),
86}
87
88#[derive(Debug)]
89enum Command {
90 Dial(Multiaddr),
91 Listen(Multiaddr),
92 AddDhtPeer(PeerId, Multiaddr),
93 FindClosestPeers(PeerId, oneshot::Sender<Vec<DiscoveredPeer>>),
94 StartProviding(Vec<u8>),
95 StopProviding(Vec<u8>),
96 GetProviders(Vec<u8>, oneshot::Sender<Vec<PeerId>>),
97 BlockPeer(PeerId),
98 UnblockPeer(PeerId),
99}
100
101/// Peer descubierto vía DHT: identidad + direcciones conocidas.
102#[derive(Debug, Clone)]
103pub struct DiscoveredPeer {
104 pub peer_id: PeerId,
105 pub addrs: Vec<Multiaddr>,
106}
107
108/// Nodo Brahman en la malla P2P. Maneja el swarm libp2p y expone
109/// API uniforme para listen/dial/DHT/streams.
110pub struct BrahmanNet {
111 /// Identidad libp2p de este nodo. Estable mientras viva la
112 /// keypair (efímera por default; persistente si pasaste una
113 /// vía [`with_keypair`]).
114 pub peer_id: PeerId,
115 /// Keypair compartida (Arc para compartir con consumers que
116 /// necesitan firmar mensajes con la misma identidad — p. ej.
117 /// `brahman_handshake::network::connect_libp2p` que firma el
118 /// Hello). NO se expone públicamente; usar [`Self::keypair`].
119 keypair: Arc<Keypair>,
120 cmd_tx: mpsc::UnboundedSender<Command>,
121 listen_rx: Mutex<mpsc::UnboundedReceiver<Multiaddr>>,
122 /// Control para abrir y aceptar streams. Cada protocolo
123 /// (handshake brahman, sync minga, etc.) llama
124 /// `control.accept(StreamProtocol::new("/foo/1.0.0"))` para
125 /// recibir streams entrantes, o `control.open_stream(peer, proto)`
126 /// para abrirlos. Multiplexado y demultiplexado lo hace libp2p.
127 pub control: stream::Control,
128}
129
130impl BrahmanNet {
131 /// Crea un nodo con keypair Ed25519 generada al vuelo (peer_id
132 /// efímero — cambia en cada arranque).
133 pub fn new() -> Result<Self, NodeError> {
134 Self::with_keypair(identity::Keypair::generate_ed25519())
135 }
136
137 /// Crea un nodo con una keypair libp2p específica. Usá esto para
138 /// `peer_id` estable (por ejemplo si tu identidad se persiste a
139 /// disco, o si la derivás de la identidad criptográfica del
140 /// módulo).
141 ///
142 /// Sólo Ed25519 se soporta — la `keypair` se duplica internamente
143 /// vía clone del `ed25519::Keypair` para que tanto el swarm
144 /// (Noise auth) como el caller (firma de Cards) compartan la
145 /// misma identidad sin la fricción de que `identity::Keypair` no
146 /// implemente `Clone`.
147 pub fn with_keypair(keypair: identity::Keypair) -> Result<Self, NodeError> {
148 let ed_kp = keypair
149 .try_into_ed25519()
150 .map_err(|_| NodeError::Build("brahman-net sólo soporta keypairs Ed25519".into()))?;
151 let kp_for_swarm = identity::Keypair::from(ed_kp.clone());
152 let kp_for_storage = Arc::new(identity::Keypair::from(ed_kp));
153 let peer_id = kp_for_swarm.public().to_peer_id();
154
155 let mut swarm: Swarm<BrahmanBehaviour> = SwarmBuilder::with_existing_identity(kp_for_swarm)
156 .with_tokio()
157 .with_tcp(
158 tcp::Config::default(),
159 noise::Config::new,
160 yamux::Config::default,
161 )
162 .map_err(|e| NodeError::Build(format!("{e}")))?
163 .with_behaviour(|key| {
164 let local = key.public().to_peer_id();
165 let mut kad =
166 kad::Behaviour::new(local, kad::store::MemoryStore::new(local));
167 // Modo Server: respondemos a queries del DHT. Auto
168 // requiere detectar reachability; para entornos
169 // controlados (localhost, redes privadas) Server es
170 // lo correcto.
171 kad.set_mode(Some(kad::Mode::Server));
172 let identify = identify::Behaviour::new(
173 identify::Config::new(IDENTIFY_PROTOCOL.to_string(), key.public())
174 .with_agent_version(format!("brahman-net/{}", env!("CARGO_PKG_VERSION"))),
175 );
176 BrahmanBehaviour {
177 block_list: allow_block_list::Behaviour::default(),
178 stream: stream::Behaviour::new(),
179 kad,
180 identify,
181 }
182 })
183 .map_err(|e| NodeError::Build(format!("{e}")))?
184 .with_swarm_config(|c| c.with_idle_connection_timeout(IDLE_CONNECTION_TIMEOUT))
185 .build();
186
187 let control = swarm.behaviour().stream.new_control();
188
189 let (cmd_tx, mut cmd_rx) = mpsc::unbounded_channel::<Command>();
190 let (listen_tx, listen_rx) = mpsc::unbounded_channel::<Multiaddr>();
191
192 tokio::spawn(async move {
193 let mut pending_finds: HashMap<
194 kad::QueryId,
195 oneshot::Sender<Vec<DiscoveredPeer>>,
196 > = HashMap::new();
197 let mut pending_providers: HashMap<
198 kad::QueryId,
199 (Vec<PeerId>, oneshot::Sender<Vec<PeerId>>),
200 > = HashMap::new();
201
202 loop {
203 tokio::select! {
204 Some(cmd) = cmd_rx.recv() => {
205 match cmd {
206 Command::Dial(addr) => {
207 let _ = swarm.dial(addr);
208 }
209 Command::Listen(addr) => {
210 let _ = swarm.listen_on(addr);
211 }
212 Command::AddDhtPeer(peer, addr) => {
213 swarm.behaviour_mut().kad.add_address(&peer, addr);
214 }
215 Command::FindClosestPeers(target, tx) => {
216 let qid = swarm.behaviour_mut().kad.get_closest_peers(target);
217 pending_finds.insert(qid, tx);
218 }
219 Command::StartProviding(key) => {
220 // Best-effort: si falla (sin peers cercanos para
221 // replicar), seguirá viviendo en el local store
222 // y se servirá vía get_providers de quien tenga
223 // conexión con nosotros.
224 let _ = swarm.behaviour_mut().kad.start_providing(key.into());
225 }
226 Command::StopProviding(key) => {
227 // Quitamos el record local del provider store.
228 // Los peers cercanos eventualmente expiran su
229 // copia replicada por TTL natural (~24h en
230 // libp2p kad default); para retiro inmediato
231 // habría que enviar un republish con sentinel,
232 // pero kad no expone esa primitiva. Aceptable
233 // para el caso "el provider local desapareció":
234 // queries que pasen por nosotros dejan de
235 // listarnos al instante.
236 swarm.behaviour_mut().kad.stop_providing(&key.into());
237 }
238 Command::GetProviders(key, tx) => {
239 let qid = swarm.behaviour_mut().kad.get_providers(key.into());
240 pending_providers.insert(qid, (Vec::new(), tx));
241 }
242 Command::BlockPeer(peer) => {
243 swarm.behaviour_mut().block_list.block_peer(peer);
244 }
245 Command::UnblockPeer(peer) => {
246 swarm.behaviour_mut().block_list.unblock_peer(peer);
247 }
248 }
249 }
250 event = swarm.select_next_some() => {
251 match event {
252 SwarmEvent::NewListenAddr { address, .. } => {
253 let _ = listen_tx.send(address);
254 }
255 // Identify nos dice las listen-addrs reales del
256 // peer. Las inyectamos a Kad para poblar el
257 // routing table sin necesidad de add_dht_peer
258 // manual — la propagación pasa a ser automática.
259 SwarmEvent::Behaviour(BrahmanBehaviourEvent::Identify(
260 identify::Event::Received { peer_id, info, .. }
261 )) => {
262 for addr in info.listen_addrs {
263 swarm.behaviour_mut().kad.add_address(&peer_id, addr);
264 }
265 }
266 SwarmEvent::Behaviour(BrahmanBehaviourEvent::Kad(
267 kad::Event::OutboundQueryProgressed { id, result, step, .. }
268 )) => {
269 match result {
270 kad::QueryResult::GetClosestPeers(Ok(ok)) if step.last => {
271 if let Some(tx) = pending_finds.remove(&id) {
272 let infos = ok.peers.into_iter()
273 .map(|p| DiscoveredPeer {
274 peer_id: p.peer_id,
275 addrs: p.addrs,
276 })
277 .collect();
278 let _ = tx.send(infos);
279 }
280 }
281 kad::QueryResult::GetClosestPeers(Err(_)) if step.last => {
282 if let Some(tx) = pending_finds.remove(&id) {
283 let _ = tx.send(Vec::new());
284 }
285 }
286 kad::QueryResult::GetProviders(Ok(ok)) => {
287 if let Some((collected, _)) =
288 pending_providers.get_mut(&id)
289 {
290 if let kad::GetProvidersOk::FoundProviders {
291 providers, ..
292 } = ok
293 {
294 for p in providers {
295 if !collected.contains(&p) {
296 collected.push(p);
297 }
298 }
299 }
300 }
301 if step.last {
302 if let Some((providers, tx)) =
303 pending_providers.remove(&id)
304 {
305 let _ = tx.send(providers);
306 }
307 }
308 }
309 kad::QueryResult::GetProviders(Err(_)) if step.last => {
310 if let Some((providers, tx)) =
311 pending_providers.remove(&id)
312 {
313 let _ = tx.send(providers);
314 }
315 }
316 _ => {}
317 }
318 }
319 _ => {}
320 }
321 }
322 }
323 }
324 });
325
326 Ok(Self {
327 peer_id,
328 keypair: kp_for_storage,
329 cmd_tx,
330 listen_rx: Mutex::new(listen_rx),
331 control,
332 })
333 }
334
335 /// Acceso a la keypair de identidad del nodo. Usar para firmar
336 /// payloads que viajan asociados al `peer_id` (handshake brahman
337 /// firmado, futuros sub-protocolos con autenticación). El `Arc`
338 /// permite compartir sin copia — la keypair libp2p no es `Clone`.
339 pub fn keypair(&self) -> Arc<Keypair> {
340 self.keypair.clone()
341 }
342
343 /// Bloquea conexiones desde/hacia `peer` a nivel del swarm.
344 /// Conexiones existentes se cierran y nuevos intentos son
345 /// rechazados ANTES del Noise handshake — más eficiente que
346 /// rechazar al nivel del handshake brahman (ahorra round-trip
347 /// TCP+Noise por intento). Idempotente.
348 pub fn block_peer(&self, peer: PeerId) {
349 let _ = self.cmd_tx.send(Command::BlockPeer(peer));
350 }
351
352 /// Quita a `peer` de la block-list del swarm. Conexiones futuras
353 /// son aceptadas con normalidad. Idempotente.
354 pub fn unblock_peer(&self, peer: PeerId) {
355 let _ = self.cmd_tx.send(Command::UnblockPeer(peer));
356 }
357
358 /// Empieza a escuchar en `addr`. Bloquea hasta que el listener
359 /// publique su dirección real (Multiaddr resuelta — útil cuando
360 /// pediste `/ip4/0.0.0.0/tcp/0` y querés saber qué puerto te tocó).
361 pub async fn listen(&self, addr: Multiaddr) -> Multiaddr {
362 self.cmd_tx
363 .send(Command::Listen(addr))
364 .expect("swarm task alive");
365 let mut rx = self.listen_rx.lock().await;
366 rx.recv().await.expect("listen address arrives")
367 }
368
369 /// Inicia conexión con un peer en `addr`. No-op si ya hay
370 /// conexión. Best-effort — fallos se loggean al swarm pero no se
371 /// propagan al caller (consistente con libp2p).
372 pub fn dial(&self, addr: Multiaddr) {
373 let _ = self.cmd_tx.send(Command::Dial(addr));
374 }
375
376 /// Añade un peer al routing table de Kademlia. Punto de entrada
377 /// para bootstrap: tras esto, el nodo puede dirigir queries DHT
378 /// a través de este peer.
379 pub fn add_dht_peer(&self, peer: PeerId, addr: Multiaddr) {
380 let _ = self.cmd_tx.send(Command::AddDhtPeer(peer, addr));
381 }
382
383 /// Consulta el DHT por los peers más cercanos al `target` PeerId.
384 /// Devuelve la lista resuelta (vacía si la query falla o si no
385 /// hay peers conocidos). Bloquea hasta que la query completa.
386 pub async fn find_closest_peers(&self, target: PeerId) -> Vec<DiscoveredPeer> {
387 let (tx, rx) = oneshot::channel();
388 let _ = self
389 .cmd_tx
390 .send(Command::FindClosestPeers(target, tx));
391 rx.await.unwrap_or_default()
392 }
393
394 /// Anuncia en el DHT que este peer tiene el contenido identificado
395 /// por `key`. Otros peers pueden luego descubrirlo vía
396 /// [`find_providers`](Self::find_providers). Best-effort: si la
397 /// replicación falla inicialmente, el record vive en el store
398 /// local hasta que llegue conexión.
399 pub fn start_providing(&self, key: &[u8]) {
400 let _ = self.cmd_tx.send(Command::StartProviding(key.to_vec()));
401 }
402
403 /// Retira el anuncio previo de [`start_providing`] para `key`.
404 /// El record local se borra al instante (queries que lleguen a
405 /// nosotros dejan de listarnos). Los records replicados en peers
406 /// remotos viven hasta su TTL — kad no expone primitiva para
407 /// retracción inmediata cross-peer. Aceptable: simétrico al
408 /// caso "el provider apareció" (también propagación eventual).
409 pub fn stop_providing(&self, key: &[u8]) {
410 let _ = self.cmd_tx.send(Command::StopProviding(key.to_vec()));
411 }
412
413 /// Consulta el DHT por peers que han anunciado proveer `key`.
414 /// Devuelve la lista de `PeerId`s que se reportan como providers.
415 /// Lista vacía si nadie anuncia.
416 pub async fn find_providers(&self, key: &[u8]) -> Vec<PeerId> {
417 let (tx, rx) = oneshot::channel();
418 let _ = self
419 .cmd_tx
420 .send(Command::GetProviders(key.to_vec(), tx));
421 rx.await.unwrap_or_default()
422 }
423}