tawasuyu/brahman

1//! Test E2E: handshake brahman remoto sobre libp2p stream.
2//!
3//! Pipeline:
4//! 1. Server: bind Unix socket (necesario aunque no lo use el cliente);
5//!    crear `BrahmanNet` y escuchar en `/ip4/127.0.0.1/tcp/0`;
6//!    montar `run_libp2p_accept_loop`.
7//! 2. Client: crear su propio `BrahmanNet`; dial al multiaddr del
8//!    server; `connect_libp2p` con su Card; `ping`; `farewell`.
9//! 3. Verificar: el server registró la sesión; sessions.len() == 1
10//!    durante la sesión, == 0 después del farewell.
11
12use std::collections::BTreeSet;
13use std::sync::Arc;
14use std::time::Duration;
15
16use brahman_broker::{Broker, BrokerConfig};
17use brahman_card::{
18    ulid::Ulid, Card, CardKind, Lifecycle, Payload, Priority, Supervision,
19    CARD_SCHEMA_VERSION,
20};
21use brahman_handshake::identity::{Identity, DEFAULT_SESSION_TTL};
22use brahman_handshake::network::{connect_libp2p, connect_libp2p_with_cert, run_libp2p_accept_loop};
23use brahman_handshake::peer_policy::PeerPolicy;
24use brahman_handshake::server::{Server, ServerConfig};
25use brahman_net::{BrahmanNet, Keypair, Multiaddr, PeerId, Protocol};
26use tempfile::TempDir;
27use tokio::sync::Mutex;
28
29fn sample_card(label: &str) -> Card {
30    Card {
31        schema_version: CARD_SCHEMA_VERSION,
32        id: Ulid::new(),
33        label: label.into(),
34        provides: BTreeSet::new(),
35        requires: BTreeSet::new(),
36        permissions: Default::default(),
37        soma: Default::default(),
38        payload: Payload::Virtual,
39        supervision: Supervision::OneShot,
40        lifecycle: Lifecycle::default(),
41        priority: Priority::default(),
42        kind: CardKind::Ente,
43        ..Default::default()
44    }
45}
46
47#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
48async fn libp2p_handshake_roundtrip() {
49    // ---- Server side ----
50    let tmp = TempDir::new().unwrap();
51    let unix_socket = tmp.path().join("brahman-init.sock");
52
53    let broker = Arc::new(Mutex::new(Broker::new(BrokerConfig::default())));
54    let server = Arc::new(
55        Server::bind(
56            &unix_socket,
57            ServerConfig {
58                init_attached: true,
59                broker: Some(broker.clone()),
60                net: None,
61                policy: None,
62            },
63        )
64        .unwrap(),
65    );
66    let sessions = server.sessions();
67
68    let server_net = Arc::new(BrahmanNet::new().unwrap());
69    let server_peer_id = server_net.peer_id;
70
71    // Listen on a random TCP port.
72    let listen_addr: Multiaddr = "/ip4/127.0.0.1/tcp/0".parse().unwrap();
73    let actual_addr = server_net.listen(listen_addr).await;
74    // Inject the libp2p PeerId into the multiaddr so the client knows
75    // who to dial.
76    let mut full_addr = actual_addr.clone();
77    full_addr.push(Protocol::P2p(server_peer_id));
78
79    // Spawn the libp2p accept loop.
80    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
81
82    // ---- Client side ----
83    let client_net = BrahmanNet::new().unwrap();
84    client_net.dial(full_addr.clone());
85
86    // Pequeña espera para que el dial conecte. En un entorno real el
87    // caller usaría un mecanismo de barrier, pero para tests un sleep
88    // corto es suficiente y deterministic en localhost.
89    tokio::time::sleep(Duration::from_millis(200)).await;
90
91    let card = sample_card("test.remote_ente");
92    let client_kp = client_net.keypair();
93    let mut client = connect_libp2p(&client_net, server_peer_id, card, None, &client_kp)
94        .await
95        .expect("handshake remoto debería completar");
96
97    // Verificación: el server vio la sesión.
98    {
99        let s = sessions.lock().await;
100        assert_eq!(s.len(), 1, "una sesión registrada");
101        let resolved = s.values().next().unwrap();
102        assert_eq!(resolved.card.label, "test.remote_ente");
103    }
104
105    // Ping roundtrip.
106    let ts = client.ping().await.expect("ping debería responder");
107    assert!(ts > 0, "timestamp del Pong > 0");
108
109    // Farewell limpio.
110    client.farewell().await.expect("farewell debería completar");
111
112    // Tras el farewell, el cleanup remueve la sesión.
113    // Damos un tick para que el handler procese el frame.
114    tokio::time::sleep(Duration::from_millis(100)).await;
115    {
116        let s = sessions.lock().await;
117        assert_eq!(s.len(), 0, "sesión removida tras farewell");
118    }
119
120    // peer_id no usado aquí, pero validamos que la API existe.
121    let _ = PeerId::random();
122}
123
124/// Fase 3 negativo: el cliente intenta firmar el Hello con una keypair
125/// distinta a la del peer libp2p. El server (que verifica que la
126/// public key del Hello derive al peer_id autenticado por Noise) debe
127/// rechazar con `Unauthorized`.
128#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
129async fn libp2p_handshake_rejects_mismatched_signing_key() {
130    let tmp = TempDir::new().unwrap();
131    let unix_socket = tmp.path().join("brahman-init.sock");
132
133    let server = Arc::new(
134        Server::bind(
135            &unix_socket,
136            ServerConfig {
137                init_attached: true,
138                broker: None,
139                net: None,
140                policy: None,
141            },
142        )
143        .unwrap(),
144    );
145    let sessions = server.sessions();
146
147    let server_net = Arc::new(BrahmanNet::new().unwrap());
148    let server_peer = server_net.peer_id;
149    let listen_addr: Multiaddr = "/ip4/127.0.0.1/tcp/0".parse().unwrap();
150    let actual = server_net.listen(listen_addr).await;
151    let mut full = actual.clone();
152    full.push(Protocol::P2p(server_peer));
153
154    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
155
156    let client_net = BrahmanNet::new().unwrap();
157    client_net.dial(full);
158    tokio::time::sleep(Duration::from_millis(200)).await;
159
160    // Keypair fraudulenta: NO es la del client_net.
161    let evil_keypair = Keypair::generate_ed25519();
162
163    let card = sample_card("test.evil");
164    let result = connect_libp2p(&client_net, server_peer, card, None, &evil_keypair).await;
165
166    assert!(
167        result.is_err(),
168        "handshake con keypair fraudulenta debe fallar"
169    );
170
171    // Sanidad: ninguna sesión registrada.
172    let s = sessions.lock().await;
173    assert_eq!(s.len(), 0, "no debería haber sesión registrada");
174}
175
176/// Allowlist gate: A configura `allowlist = [client_authorized_peer]`.
177/// Un cliente con peer_id en la lista pasa el handshake; otro con
178/// peer_id distinto es rechazado con `Unauthorized` ANTES de la
179/// verificación de firma (la allowlist se chequea primero, es más
180/// barata).
181#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
182async fn libp2p_handshake_allowlist_admits_listed_rejects_others() {
183    // Pre-generamos las dos identidades cliente para que A pueda
184    // construir la allowlist conociendo cuál es la "permitida".
185    let allowed_kp = Keypair::generate_ed25519();
186    let allowed_peer = allowed_kp.public().to_peer_id();
187    let denied_kp = Keypair::generate_ed25519();
188    // (denied_peer no se necesita para la lista — sólo para clarity)
189    let _ = denied_kp.public().to_peer_id();
190
191    // ---- Server con allowlist activa ----
192    let tmp = TempDir::new().unwrap();
193    let unix_socket = tmp.path().join("brahman-init.sock");
194    let server = Arc::new(
195        Server::bind(
196            &unix_socket,
197            ServerConfig {
198                init_attached: true,
199                broker: None,
200                net: None,
201                policy: Some(PeerPolicy::from_sets(
202                    Some([allowed_peer].into_iter().collect()),
203                    std::collections::BTreeSet::new(),
204                )),
205            },
206        )
207        .unwrap(),
208    );
209    let sessions = server.sessions();
210
211    let server_net = Arc::new(BrahmanNet::new().unwrap());
212    let server_peer = server_net.peer_id;
213    let actual = server_net.listen("/ip4/127.0.0.1/tcp/0".parse().unwrap()).await;
214    let mut full = actual.clone();
215    full.push(Protocol::P2p(server_peer));
216
217    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
218
219    // ---- Cliente PERMITIDO ----
220    let allowed_net = BrahmanNet::with_keypair(allowed_kp.clone()).unwrap();
221    allowed_net.dial(full.clone());
222    tokio::time::sleep(Duration::from_millis(200)).await;
223
224    let card_ok = sample_card("test.allowed");
225    let mut allowed_client = connect_libp2p(&allowed_net, server_peer, card_ok, None, &allowed_kp)
226        .await
227        .expect("peer en allowlist debe pasar");
228
229    {
230        let s = sessions.lock().await;
231        assert_eq!(s.len(), 1, "sesión del peer permitido registrada");
232    }
233    allowed_client.farewell().await.ok();
234    tokio::time::sleep(Duration::from_millis(100)).await;
235
236    // ---- Cliente DENEGADO ----
237    let denied_net = BrahmanNet::with_keypair(denied_kp.clone()).unwrap();
238    denied_net.dial(full.clone());
239    tokio::time::sleep(Duration::from_millis(200)).await;
240
241    let card_no = sample_card("test.denied");
242    let result = connect_libp2p(&denied_net, server_peer, card_no, None, &denied_kp).await;
243
244    assert!(
245        result.is_err(),
246        "peer fuera de allowlist debe ser rechazado, got: {:?}",
247        result.is_ok()
248    );
249    {
250        let s = sessions.lock().await;
251        assert_eq!(s.len(), 0, "ninguna sesión adicional registrada tras intento denegado");
252    }
253}
254
255/// Denylist gate: A configura `policy` con un peer en la denylist.
256/// Modo abierto para todo lo demás (sin allowlist), pero el peer
257/// baneado es rechazado aún teniendo Ed25519 válida y peer_id que
258/// derivaría limpio del Noise handshake.
259#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
260async fn libp2p_handshake_denylist_blocks_listed_peer() {
261    let banned_kp = Keypair::generate_ed25519();
262    let banned_peer = banned_kp.public().to_peer_id();
263    let other_kp = Keypair::generate_ed25519();
264
265    let tmp = TempDir::new().unwrap();
266    let unix_socket = tmp.path().join("brahman-init.sock");
267    let server = Arc::new(
268        Server::bind(
269            &unix_socket,
270            ServerConfig {
271                init_attached: true,
272                broker: None,
273                net: None,
274                policy: Some(PeerPolicy::from_sets(
275                    None, // sin allowlist (abierto)
276                    [banned_peer].into_iter().collect(),
277                )),
278            },
279        )
280        .unwrap(),
281    );
282    let sessions = server.sessions();
283
284    let server_net = Arc::new(BrahmanNet::new().unwrap());
285    let server_peer = server_net.peer_id;
286    let actual = server_net
287        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
288        .await;
289    let mut full = actual.clone();
290    full.push(Protocol::P2p(server_peer));
291
292    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
293
294    // Cliente baneado: connect debe fallar.
295    let banned_net = BrahmanNet::with_keypair(banned_kp.clone()).unwrap();
296    banned_net.dial(full.clone());
297    tokio::time::sleep(Duration::from_millis(200)).await;
298
299    let card_x = sample_card("test.banned");
300    let result = connect_libp2p(&banned_net, server_peer, card_x, None, &banned_kp).await;
301    assert!(
302        result.is_err(),
303        "peer en denylist debe ser rechazado, got Ok"
304    );
305    {
306        let s = sessions.lock().await;
307        assert_eq!(s.len(), 0, "el peer baneado no debería tener sesión");
308    }
309
310    // Cliente no-baneado pasa.
311    let other_net = BrahmanNet::with_keypair(other_kp.clone()).unwrap();
312    other_net.dial(full.clone());
313    tokio::time::sleep(Duration::from_millis(200)).await;
314
315    let card_ok = sample_card("test.other");
316    let mut other_client = connect_libp2p(&other_net, server_peer, card_ok, None, &other_kp)
317        .await
318        .expect("peer fuera de denylist debe pasar");
319    {
320        let s = sessions.lock().await;
321        assert_eq!(s.len(), 1, "sesión del peer no-baneado registrada");
322    }
323    other_client.farewell().await.ok();
324}
325
326/// Swarm-level deny via `PeerPolicy::attach_to_net`: cuando la deny
327/// se aplica al swarm vía `block_list`, el peer baneado es rechazado
328/// en el dial — la conexión TCP/Noise nunca completa, así que el
329/// cliente nunca llega siquiera a mandar el Hello. Más eficiente que
330/// el handshake-level deny.
331#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
332async fn swarm_level_deny_blocks_before_noise() {
333    let banned_kp = Keypair::generate_ed25519();
334    let banned_peer = banned_kp.public().to_peer_id();
335
336    let tmp = TempDir::new().unwrap();
337    let unix_socket = tmp.path().join("brahman-init.sock");
338    let policy = brahman_handshake::peer_policy::PeerPolicy::from_sets(
339        None,
340        [banned_peer].into_iter().collect(),
341    );
342    let server = Arc::new(
343        Server::bind(
344            &unix_socket,
345            ServerConfig {
346                init_attached: true,
347                broker: None,
348                net: None,
349                policy: Some(policy.clone()),
350            },
351        )
352        .unwrap(),
353    );
354    let server_net = Arc::new(BrahmanNet::new().unwrap());
355    let server_peer = server_net.peer_id;
356
357    // ATTACH: la deny se proyecta al swarm. Es lo nuevo de este
358    // commit — sin esta llamada, el deny seguiría aplicando sólo
359    // al nivel de handshake brahman (lo que también funciona pero
360    // gasta un round-trip Noise).
361    policy.attach_to_net(server_net.clone());
362
363    let actual = server_net
364        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
365        .await;
366    let mut full = actual.clone();
367    full.push(Protocol::P2p(server_peer));
368
369    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
370
371    // Cliente baneado intenta dial + handshake. Con swarm-level
372    // deny, la conexión libp2p ni siquiera completa: `connect_libp2p`
373    // falla con error de open_stream (peer inalcanzable / connection
374    // refused) en lugar del Unauthorized del handshake-level path.
375    let banned_net = BrahmanNet::with_keypair(banned_kp.clone()).unwrap();
376    banned_net.dial(full.clone());
377
378    let card = sample_card("test.swarm_banned");
379    // Timeout corto: si el block falla, el handshake completaría
380    // rápido en localhost. Si funciona, debería fallar el dial casi
381    // instantáneo o colgarse hasta el timeout.
382    let result = tokio::time::timeout(
383        Duration::from_secs(3),
384        connect_libp2p(&banned_net, server_peer, card, None, &banned_kp),
385    )
386    .await;
387
388    match result {
389        Ok(Ok(_)) => panic!("peer baneado a nivel swarm NO debería completar handshake"),
390        Ok(Err(e)) => {
391            // Esperado: error de transporte/stream, no de handshake.
392            tracing::info!(error = %e, "swarm-level deny rechazó como esperado");
393        }
394        Err(_) => {
395            // También aceptable: timeout porque el dial nunca completa.
396            tracing::info!("swarm-level deny → connect timeout (también OK)");
397        }
398    }
399}
400
401/// Multi-key identity: la propiedad fundamental que cierra el
402/// proyecto. El cliente B tiene una identity master estable; el
403/// server A le permite el master_peer en allowlist. B se conecta con
404/// **session1**; pasa. B "rota": genera **session2** distinta, emite
405/// un nuevo cert con la misma identity, se conecta de nuevo. Pasa
406/// también — sin que A toque su allowlist.
407#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
408async fn identity_cert_allows_session_rotation_without_policy_change() {
409    // Master de B (estable, persistente).
410    let master_kp = Keypair::generate_ed25519();
411    let master_peer = master_kp.public().to_peer_id();
412    let identity = Identity::from_keypair(master_kp);
413
414    // A configura policy: allowlist con master_peer (NO sessions).
415    let tmp = TempDir::new().unwrap();
416    let unix_socket = tmp.path().join("brahman-init.sock");
417    let server = Arc::new(
418        Server::bind(
419            &unix_socket,
420            ServerConfig {
421                init_attached: true,
422                broker: None,
423                net: None,
424                policy: Some(PeerPolicy::from_sets(
425                    Some([master_peer].into_iter().collect()),
426                    std::collections::BTreeSet::new(),
427                )),
428            },
429        )
430        .unwrap(),
431    );
432    let sessions = server.sessions();
433
434    let server_net = Arc::new(BrahmanNet::new().unwrap());
435    let server_peer = server_net.peer_id;
436    let actual = server_net
437        .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
438        .await;
439    let mut full = actual.clone();
440    full.push(Protocol::P2p(server_peer));
441
442    tokio::spawn(run_libp2p_accept_loop(server.clone(), server_net.clone()));
443
444    // ---- Conexión 1: session1 ----
445    let session1_kp = Keypair::generate_ed25519();
446    let cert1 = identity
447        .issue_session_cert(&session1_kp, DEFAULT_SESSION_TTL)
448        .unwrap();
449    let net1 = BrahmanNet::with_keypair(session1_kp.clone()).unwrap();
450    net1.dial(full.clone());
451    tokio::time::sleep(Duration::from_millis(200)).await;
452
453    let mut client1 = connect_libp2p_with_cert(
454        &net1,
455        server_peer,
456        sample_card("test.session1"),
457        None,
458        &session1_kp,
459        cert1,
460    )
461    .await
462    .expect("session1 con cert válido del master allowlisted debe pasar");
463
464    {
465        let s = sessions.lock().await;
466        assert_eq!(s.len(), 1, "session1 registrada");
467    }
468    client1.farewell().await.ok();
469    tokio::time::sleep(Duration::from_millis(100)).await;
470
471    // ---- ROTACIÓN: session2 distinta, mismo master ----
472    let session2_kp = Keypair::generate_ed25519();
473    assert_ne!(
474        session1_kp.public().to_peer_id(),
475        session2_kp.public().to_peer_id(),
476        "test inválido si las sessions son iguales"
477    );
478    let cert2 = identity
479        .issue_session_cert(&session2_kp, DEFAULT_SESSION_TTL)
480        .unwrap();
481
482    let net2 = BrahmanNet::with_keypair(session2_kp.clone()).unwrap();
483    net2.dial(full.clone());
484    tokio::time::sleep(Duration::from_millis(200)).await;
485
486    let mut client2 = connect_libp2p_with_cert(
487        &net2,
488        server_peer,
489        sample_card("test.session2"),
490        None,
491        &session2_kp,
492        cert2,
493    )
494    .await
495    .expect(
496        "session2 (rotada) con cert del MISMO master debe pasar sin tocar allowlist",
497    );
498
499    {
500        let s = sessions.lock().await;
501        assert_eq!(s.len(), 1, "session2 registrada");
502    }
503    client2.farewell().await.ok();
504
505    // Sanity: una session sin cert (path Fase 3) cuyo session_peer_id
506    // NO está en la allowlist (porque la allowlist tiene master, no
507    // sessions) DEBE ser rechazada.
508    let session_other = Keypair::generate_ed25519();
509    let net_other = BrahmanNet::with_keypair(session_other.clone()).unwrap();
510    net_other.dial(full.clone());
511    tokio::time::sleep(Duration::from_millis(200)).await;
512
513    let result = connect_libp2p(
514        &net_other,
515        server_peer,
516        sample_card("test.no_cert"),
517        None,
518        &session_other,
519    )
520    .await;
521    assert!(
522        result.is_err(),
523        "sin cert, session_peer_id (no listado) debe ser rechazado"
524    );
525}