tawasuyu/brahman

1//! Backend libp2p del handshake brahman: el mismo protocolo (Hello /
2//! HelloAck / Ping / Pong / MatchEvent / Farewell, frames postcard
3//! length-prefixed) ahora también viaja sobre streams libp2p de la
4//! malla `brahman-net`.
5//!
6//! El servidor expone el bucle [`run_libp2p_accept_loop`] que acepta
7//! streams del protocolo `BRAHMAN_HANDSHAKE_PROTOCOL` y los delega al
8//! mismo `Server` que ya escucha por Unix socket — la `Session` es
9//! genérica sobre el transporte, así que ambas vías comparten broker,
10//! tablas de sesiones, push de MatchEvents, todo.
11//!
12//! El cliente se conecta vía [`connect_libp2p`]: abre un stream
13//! libp2p hacia un `PeerId` ya conocido y arranca el handshake como
14//! cualquier `Client`.
15//!
16//! Identidad: cada nodo libp2p tiene su `PeerId` (ed25519 derivado).
17//! La identidad del Ente (Card.id ULID + futura firma) viaja en el
18//! Hello, como en el path Unix. Trust remoto (verificación de firma
19//! antes de aceptar el Hello) es Fase 3.
20//!
21//! Ejemplo (servidor — Arje):
22//! ```ignore
23//! let server = Arc::new(Server::bind("/run/brahman-init.sock", config)?);
24//! let net = Arc::new(BrahmanNet::new()?);
25//! net.listen("/ip4/0.0.0.0/tcp/4101".parse()?).await;
26//!
27//! tokio::spawn(brahman_handshake::network::run_libp2p_accept_loop(
28//!     server.clone(),
29//!     net.clone(),
30//! ));
31//! // Server::run sigue escuchando Unix en paralelo.
32//! ```
33//!
34//! Ejemplo (cliente — sidecar de un Ente remoto):
35//! ```ignore
36//! let net = BrahmanNet::new()?;
37//! net.dial(remote_multiaddr);
38//! let mut client = brahman_handshake::network::connect_libp2p(
39//!     &net, peer_id, my_card, None,
40//! ).await?;
41//! client.ping().await?;
42//! ```
43
44use std::sync::Arc;
45
46use brahman_card::{Card, TypeRef, WitInterface};
47use brahman_net::{BrahmanNet, Keypair, OpenStreamError, PeerId, Stream, StreamProtocol};
48
49use crate::identity::SessionCert;
50use futures::StreamExt;
51use tokio_util::compat::{Compat, FuturesAsyncReadCompatExt};
52use tracing::{debug, warn};
53
54use crate::client::{Client, ClientError};
55use crate::server::Server;
56
57/// Sub-protocolo del handshake brahman sobre streams libp2p.
58pub const BRAHMAN_HANDSHAKE_PROTOCOL: StreamProtocol =
59    StreamProtocol::new("/brahman/handshake/1.0.0");
60
61/// Tipo del stream que ve la lógica del handshake una vez convertido
62/// del mundo `futures::AsyncRead/Write` (libp2p) al mundo
63/// `tokio::io::AsyncRead/Write` (resto del crate).
64pub type LibP2pHandshakeStream = Compat<Stream>;
65
66/// Errores específicos del backend libp2p.
67#[derive(Debug, thiserror::Error)]
68pub enum NetworkError {
69    #[error("abrir stream libp2p: {0}")]
70    OpenStream(#[from] OpenStreamError),
71
72    #[error("handshake: {0}")]
73    Handshake(#[from] ClientError),
74
75    #[error("aceptar stream libp2p: {0}")]
76    AcceptStream(String),
77}
78
79/// Loop de aceptación de streams libp2p del protocolo handshake.
80/// Cada stream entrante se construye como `Session` reutilizando las
81/// tablas compartidas del `Server`, así que conviven con sesiones
82/// Unix indistinguibles.
83///
84/// Vive hasta que el control libp2p se cierre o el caller drop el
85/// future. Errores por sesión se loggean (no tumban el loop).
86pub async fn run_libp2p_accept_loop(
87    server: Arc<Server>,
88    net: Arc<BrahmanNet>,
89) -> Result<(), NetworkError> {
90    let mut control = net.control.clone();
91    let mut incoming = control
92        .accept(BRAHMAN_HANDSHAKE_PROTOCOL)
93        .map_err(|e| NetworkError::AcceptStream(e.to_string()))?;
94
95    while let Some((peer, stream)) = incoming.next().await {
96        let server = server.clone();
97        // .compat() debe pasar al spawn ADENTRO; si lo hacemos afuera
98        // y capturamos `Compat<Stream>` en la closure, el future
99        // resultante hereda traits que dyn AsyncReadWrite no satisface
100        // (compatibility con thread-safety de tokio::spawn).
101        tokio::spawn(handle_libp2p_session(server, stream, peer));
102    }
103
104    Ok(())
105}
106
107async fn handle_libp2p_session(
108    server: Arc<Server>,
109    stream: Stream,
110    peer: PeerId,
111) {
112    // session_from_libp2p_stream propaga el peer_id al `do_handshake`,
113    // que exige firma del Hello cuya public key derive a este peer.
114    let session = server.session_from_libp2p_stream(stream.compat(), peer);
115    if let Err(e) = session.handle().await {
116        warn!(
117            target: "brahman_handshake::network",
118            peer = %peer,
119            error = %e,
120            "sesión libp2p terminó con error"
121        );
122    }
123}
124
125/// Conecta como cliente a un Ente remoto vía libp2p y completa el
126/// handshake **firmado** con `keypair`. Requiere que `net` ya tenga
127/// conexión (o pueda dial-ar) al `peer`; típicamente el caller hace
128/// `net.dial(multiaddr)` antes.
129///
130/// La `keypair` debe ser la misma que la del nodo libp2p (la que
131/// pasaste a [`brahman_net::BrahmanNet::with_keypair`]). Si no coincide
132/// con el `peer_id` autenticado por Noise, el server rechaza el Hello
133/// con `Unauthorized`.
134///
135/// Devuelve un `Client` típico — los métodos `ping`, `await_event`,
136/// `farewell` funcionan idéntico al path Unix. El stream subyacente
137/// es libp2p convertido vía `tokio_util::compat`.
138pub async fn connect_libp2p(
139    net: &BrahmanNet,
140    peer: PeerId,
141    card: Card,
142    wit: Option<WitInterface>,
143    keypair: &Keypair,
144) -> Result<Client<LibP2pHandshakeStream>, NetworkError> {
145    let mut control = net.control.clone();
146    let stream = control
147        .open_stream(peer, BRAHMAN_HANDSHAKE_PROTOCOL)
148        .await?;
149    let client = Client::connect_with_stream_signed(stream.compat(), card, wit, keypair).await?;
150    Ok(client)
151}
152
153/// Igual que `connect_libp2p` pero adjunta un `SessionCert` al Hello.
154/// El server, al verificar el cert, evalúa la política de admisión
155/// contra el `master_peer_id` derivado — no contra el `peer_id`
156/// libp2p. Esto permite **rotar** la session keypair sin perder
157/// reconocimiento en allowlists remotas.
158///
159/// El `keypair` debe ser la session libp2p (la que firma la conexión
160/// Noise); el `cert` debe haber sido emitido por una identity master
161/// para esa misma session pubkey (ver
162/// [`crate::identity::Identity::issue_session_cert`]).
163pub async fn connect_libp2p_with_cert(
164    net: &BrahmanNet,
165    peer: PeerId,
166    card: Card,
167    wit: Option<WitInterface>,
168    session_keypair: &Keypair,
169    cert: SessionCert,
170) -> Result<Client<LibP2pHandshakeStream>, NetworkError> {
171    let mut control = net.control.clone();
172    let stream = control
173        .open_stream(peer, BRAHMAN_HANDSHAKE_PROTOCOL)
174        .await?;
175    let client = Client::connect_with_stream_signed_with_cert(
176        stream.compat(),
177        card,
178        wit,
179        session_keypair,
180        cert,
181    )
182    .await?;
183    Ok(client)
184}
185
186// =====================================================================
187// Discovery remoto via DHT — Fase 2
188// =====================================================================
189//
190// Cuando un Ente registra una Card con outputs en el Init local, el
191// Init anuncia al DHT (`net.start_providing(key)`) bajo una key
192// derivada de `(flow_name, TypeRef)`. Cualquier nodo conectado al
193// mismo DHT puede consultar `find_remote_providers(flow_name, type)`
194// y obtener la lista de `PeerId`s que dijeron proveer ese flow.
195//
196// La key es **estable y libre de colisiones** entre versiones del
197// monorepo: usa blake3 sobre un canon textual `brahman-flow|{name}|{type_canon}`.
198// Cambiar la canonicalización rompe el discovery cross-version, así
199// que cualquier modificación requiere bump de versión documentado.
200
201/// Prefijo de namespace para todas las keys DHT del subprotocolo
202/// brahman. Discrimina contra otros usos del mismo DHT (sync minga,
203/// futuros) — protege contra colisiones accidentales.
204const FLOW_KEY_PREFIX: &str = "brahman-flow|v1|";
205
206/// Canonicaliza un `TypeRef` a string estable. Cambios aquí rompen
207/// la compatibilidad de discovery cross-version; bump documentado
208/// en `FLOW_KEY_PREFIX` al modificar.
209fn canonicalize_type(t: &TypeRef) -> String {
210    match t {
211        TypeRef::Primitive { name } => format!("prim:{}", name),
212        TypeRef::Wit {
213            package,
214            interface,
215            name,
216        } => format!(
217            "wit:{}#{}#{}",
218            package,
219            interface.as_deref().unwrap_or(""),
220            name
221        ),
222    }
223}
224
225/// Deriva la key del DHT para un `(flow_name, type_ref)` específico.
226/// blake3-32B determinístico — la misma tupla en cualquier máquina
227/// produce la misma key.
228pub fn flow_dht_key(flow_name: &str, type_ref: &TypeRef) -> [u8; 32] {
229    let canon = format!(
230        "{}{}|{}",
231        FLOW_KEY_PREFIX,
232        flow_name,
233        canonicalize_type(type_ref)
234    );
235    *blake3::hash(canon.as_bytes()).as_bytes()
236}
237
238/// Anuncia al DHT que este nodo provee cada output flow declarado
239/// en `card`. Llamarlo tras `register_session` propaga la
240/// disponibilidad a todos los peers que comparten DHT con éste.
241///
242/// Idempotente: re-anunciar la misma key actualiza el TTL del record
243/// en el DHT. Best-effort: si `start_providing` falla por falta de
244/// peers cercanos (DHT vacío), el record vive en el store local
245/// hasta que llegue una conexión.
246pub fn announce_outputs(net: &BrahmanNet, card: &Card) {
247    for flow in &card.flow.output {
248        let key = flow_dht_key(&flow.name, &flow.ty);
249        debug!(
250            target: "brahman_handshake::network",
251            flow = %flow.name,
252            "announce_output → DHT"
253        );
254        net.start_providing(&key);
255    }
256}
257
258/// Retira los anuncios DHT previos de [`announce_outputs`] para esta
259/// `card`. Llamado desde `cleanup` cuando una sesión cierra (Farewell,
260/// EOF, error). El record local se borra al instante; copias
261/// replicadas en peers remotos expiran por TTL natural de kad.
262pub fn withdraw_outputs(net: &BrahmanNet, card: &Card) {
263    for flow in &card.flow.output {
264        let key = flow_dht_key(&flow.name, &flow.ty);
265        debug!(
266            target: "brahman_handshake::network",
267            flow = %flow.name,
268            "withdraw_output → DHT (stop_providing)"
269        );
270        net.stop_providing(&key);
271    }
272}
273
274/// Consulta el DHT por peers que han anunciado proveer el flow
275/// `(flow_name, type_ref)`. Devuelve la lista resuelta de `PeerId`s.
276/// Lista vacía si nadie anuncia, si la query timeout-ea, o si el
277/// DHT no ha encontrado providers.
278///
279/// Para cada `PeerId` devuelto, el caller puede luego dial-ar al
280/// peer (a sus addrs conocidas vía Identify) y abrir un sub-handshake
281/// remoto con [`connect_libp2p`].
282pub async fn find_remote_providers(
283    net: &BrahmanNet,
284    flow_name: &str,
285    type_ref: &TypeRef,
286) -> Vec<PeerId> {
287    let key = flow_dht_key(flow_name, type_ref);
288    net.find_providers(&key).await
289}