⎇ 99cd685dc1175e5d36b04e8c53057f29501ec50b
crates/core/brahman-handshake/src/server.rs
Historia
Quién tocó qué
crudo
feat(brahman-handshake): ListSessions endpoint + cliente + UI broker-explorer
1//! Servidor de handshake. Listener Unix socket → sesiones por conexión.
2
3use std::collections::HashMap;
4use std::path::{Path, PathBuf};
5use std::sync::Arc;
6use std::time::{SystemTime, UNIX_EPOCH};
7
8use brahman_broker::{Broker, Endpoint};
9use brahman_card::{Card, ResolvedCard, WitInterface, CARD_SCHEMA_VERSION};
10use brahman_net::{BrahmanNet, PeerId};
11use tokio::io::{split, AsyncRead, AsyncWrite, WriteHalf};
12use tokio::net::{UnixListener, UnixStream};
13use tokio::sync::{mpsc, Mutex};
14use tracing::{debug, warn};
15use ulid::Ulid;
16
17use crate::codec::{read_frame, write_frame};
18use crate::messages::{
19 Farewell, Frame, HandshakeError, Hello, HelloAck, MatchEvent, MatchEventKind, Ping, Pong,
20 SessionId,
21};
22
23/// Tabla de sesiones vivas indexada por `SessionId`.
24pub type SessionRegistry = Arc<Mutex<HashMap<SessionId, ResolvedCard>>>;
25
26/// Broker compartido (opcional) que el servidor mantiene en sincronía con
27/// el ciclo de vida de las sesiones.
28pub type SharedBroker = Arc<Mutex<Broker>>;
29
30/// Tabla de canales push por sesión: el server inyecta frames hacia el
31/// cliente (p. ej. `MatchEvent`) sin requerir que el cliente haga request.
32type SessionTxTable = Arc<Mutex<HashMap<SessionId, mpsc::Sender<Frame>>>>;
33
34/// Por sesión, último match conocido por nombre de input. Se usa para
35/// emitir diffs (Available/Lost) en lugar del estado completo.
36type LastMatches = Arc<Mutex<HashMap<SessionId, HashMap<String, Endpoint>>>>;
37
38/// Capacidad del canal push por sesión. Si se llena (cliente lento), los
39/// eventos extra se descartan — el cliente puede re-consultar el estado.
40const PUSH_CHANNEL_CAPACITY: usize = 32;
41
42/// Configuración del servidor.
43#[derive(Clone, Default)]
44pub struct ServerConfig {
45 /// `true` si el Init está atado al servidor (se reporta en `HelloAck`).
46 pub init_attached: bool,
47 /// Broker compartido. Si está presente, el servidor llama
48 /// `register` tras un Hello aceptado y `unregister` al cerrar la
49 /// sesión (Farewell o EOF). Si es `None`, el broker no se usa.
50 pub broker: Option<SharedBroker>,
51 /// Capa P2P compartida. Si está presente, cada Card registrada
52 /// con outputs se anuncia automáticamente al DHT vía
53 /// [`brahman_handshake::network::announce_outputs`], permitiendo
54 /// que un consumer remoto los descubra con
55 /// [`brahman_handshake::network::find_remote_providers`]. Si es
56 /// `None`, el server queda "ciego al DHT" — sólo matchea sesiones
57 /// locales (lo cual es correcto cuando no hay conectividad o no
58 /// se desea exponer al exterior).
59 pub net: Option<Arc<BrahmanNet>>,
60 /// Política de admisión de peers libp2p (allow + deny + hot
61 /// reload opcional). Si está presente, el trust gate del path
62 /// libp2p evalúa cada `peer_id` (ya autenticado por Noise)
63 /// contra esta política. `None` → modo totalmente abierto
64 /// (cualquier peer Ed25519-válido pasa). El path Unix la ignora.
65 pub policy: Option<crate::peer_policy::PeerPolicy>,
66}
67
68// Manual Debug porque BrahmanNet no implementa Debug (libp2p Swarm
69// no es Debug). Sólo loggeamos los campos relevantes para tracing.
70impl std::fmt::Debug for ServerConfig {
71 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72 f.debug_struct("ServerConfig")
73 .field("init_attached", &self.init_attached)
74 .field("broker", &self.broker.as_ref().map(|_| "<broker>"))
75 .field("net", &self.net.as_ref().map(|_| "<net>"))
76 .field("policy", &self.policy.as_ref().map(|p| p.sizes()))
77 .finish()
78 }
79}
80
81/// Servidor de handshake escuchando en un Unix socket.
82pub struct Server {
83 listener: UnixListener,
84 socket_path: PathBuf,
85 sessions: SessionRegistry,
86 push_table: SessionTxTable,
87 last_matches: LastMatches,
88 config: ServerConfig,
89}
90
91impl Server {
92 /// Crea el listener en `path`. Si el archivo existe, lo elimina (asume
93 /// que es un socket stale de una sesión previa).
94 pub fn bind(path: impl Into<PathBuf>, config: ServerConfig) -> std::io::Result<Self> {
95 let socket_path = path.into();
96 if socket_path.exists() {
97 std::fs::remove_file(&socket_path)?;
98 }
99 if let Some(parent) = socket_path.parent() {
100 if !parent.as_os_str().is_empty() {
101 std::fs::create_dir_all(parent)?;
102 }
103 }
104 let listener = UnixListener::bind(&socket_path)?;
105 Ok(Self {
106 listener,
107 socket_path,
108 sessions: Arc::new(Mutex::new(HashMap::new())),
109 push_table: Arc::new(Mutex::new(HashMap::new())),
110 last_matches: Arc::new(Mutex::new(HashMap::new())),
111 config,
112 })
113 }
114
115 /// Devuelve la ruta del socket (útil para clientes en el mismo proceso).
116 pub fn socket_path(&self) -> &Path {
117 &self.socket_path
118 }
119
120 /// Vista compartida del registro de sesiones — útil para el Init/Admin
121 /// para inspeccionar quién está conectado.
122 pub fn sessions(&self) -> SessionRegistry {
123 self.sessions.clone()
124 }
125
126 /// Acepta UNA conexión Unix, devuelve la `Session` lista para `handle()`.
127 /// No corre el handler — eso es responsabilidad del llamante.
128 /// Path Unix → `expected_peer = None` (firma del Hello opcional;
129 /// SO_PEERCRED del kernel ya autentica al cliente).
130 pub async fn accept_one(&self) -> std::io::Result<Session<UnixStream>> {
131 let (stream, _addr) = self.listener.accept().await?;
132 Ok(self.session_from_stream(stream))
133 }
134
135 /// Construye una `Session` a partir de un stream arbitrario que
136 /// implemente `AsyncRead + AsyncWrite + Unpin + Send`. Path
137 /// agnóstico al transport (Unix, in-memory, etc.) — `expected_peer`
138 /// queda en `None`, así que la firma del Hello es opcional.
139 pub fn session_from_stream<S>(&self, stream: S) -> Session<S>
140 where
141 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
142 {
143 self.session_from_stream_inner(stream, None)
144 }
145
146 /// Variante para conexiones libp2p: el `peer_id` viene autenticado
147 /// por Noise. La sesión exige firma del Hello cuya public key
148 /// derive a este `peer_id` exacto. Ver
149 /// [`super::network::run_libp2p_accept_loop`].
150 pub fn session_from_libp2p_stream<S>(
151 &self,
152 stream: S,
153 peer: PeerId,
154 ) -> Session<S>
155 where
156 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
157 {
158 self.session_from_stream_inner(stream, Some(peer))
159 }
160
161 fn session_from_stream_inner<S>(
162 &self,
163 stream: S,
164 expected_peer: Option<PeerId>,
165 ) -> Session<S>
166 where
167 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
168 {
169 Session {
170 stream,
171 sessions: self.sessions.clone(),
172 push_table: self.push_table.clone(),
173 last_matches: self.last_matches.clone(),
174 config: self.config.clone(),
175 expected_peer,
176 }
177 }
178
179 /// Loop de aceptación: cada conexión se despacha en una task separada.
180 /// Vive hasta que el listener falle o el caller drop el future.
181 pub async fn run(self) -> std::io::Result<()> {
182 loop {
183 let session = self.accept_one().await?;
184 tokio::spawn(async move {
185 if let Err(e) = session.handle().await {
186 warn!(error = %e, "session terminó con error");
187 }
188 });
189 }
190 }
191}
192
193impl Drop for Server {
194 fn drop(&mut self) {
195 // Limpieza best-effort del socket. Si falla, log y seguir.
196 if let Err(e) = std::fs::remove_file(&self.socket_path) {
197 if e.kind() != std::io::ErrorKind::NotFound {
198 warn!(path = %self.socket_path.display(), error = %e, "no se pudo limpiar socket");
199 }
200 }
201 }
202}
203
204/// Conexión individual aceptada por el servidor. Genérica sobre el
205/// transport — funciona indistinguiblemente sobre `UnixStream` (modo
206/// local), libp2p stream wrapped con `tokio_util::compat`, in-memory
207/// duplex (tests), etc.
208pub struct Session<S> {
209 stream: S,
210 sessions: SessionRegistry,
211 push_table: SessionTxTable,
212 last_matches: LastMatches,
213 config: ServerConfig,
214 /// Si está set, el path es libp2p y `do_handshake` exige firma
215 /// del Hello cuya public key derive a este `peer_id`. Si es
216 /// `None`, el path es Unix/in-memory y la firma es opcional
217 /// (pero si está, se verifica anyway por defensa en profundidad).
218 expected_peer: Option<PeerId>,
219}
220
221impl<S> Session<S>
222where
223 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
224{
225 /// Procesa la conexión hasta `Farewell` o EOF.
226 ///
227 /// Estructura: handshake (sobre el stream entero) → split en
228 /// halves (read + write) → reader loop principal + writer task
229 /// que drena el push channel. Garantiza cleanup (sessions + broker
230 /// + canales) sin importar la rama de salida.
231 ///
232 /// El split es necesario para soportar streams `!Sync` como
233 /// `libp2p::Stream`: `tokio::select!` sobre `&mut self.stream`
234 /// requeriría `S: Sync`. Con `tokio::io::split` cada mitad va a
235 /// su propia task, eliminando el requirement y permitiendo que
236 /// la misma `Session` sirva indistinta sobre Unix socket o
237 /// stream libp2p remoto.
238 pub async fn handle(self) -> std::io::Result<()> {
239 let Self {
240 mut stream,
241 sessions,
242 push_table,
243 last_matches,
244 config,
245 expected_peer,
246 } = self;
247
248 let session_id = match do_handshake(&mut stream, &config, &sessions, expected_peer).await?
249 {
250 Some(id) => id,
251 None => return Ok(()), // Hello rechazado, no se registró nada
252 };
253
254 let result = run_post_handshake(
255 stream,
256 session_id,
257 sessions.clone(),
258 push_table.clone(),
259 last_matches.clone(),
260 config.clone(),
261 )
262 .await;
263
264 cleanup(
265 session_id,
266 &sessions,
267 &push_table,
268 &last_matches,
269 &config,
270 )
271 .await;
272
273 result
274 }
275
276}
277
278// ============================================================================
279// Free functions (post-refactor): la lógica del post-handshake corre sobre
280// halves del stream; no necesita más `&mut Session<S>` y por eso vive afuera.
281// ============================================================================
282
283async fn run_post_handshake<S>(
284 stream: S,
285 session_id: SessionId,
286 sessions: SessionRegistry,
287 push_table: SessionTxTable,
288 last_matches: LastMatches,
289 config: ServerConfig,
290) -> std::io::Result<()>
291where
292 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
293{
294 // Canal por donde el server inyecta frames push (MatchEvent, etc.).
295 let (tx, mut rx) = mpsc::channel::<Frame>(PUSH_CHANNEL_CAPACITY);
296 push_table.lock().await.insert(session_id, tx);
297
298 // Tras registrar el canal, recomputar matches y emitir diffs.
299 broadcast_match_diffs(&push_table, &last_matches, &config).await;
300
301 // Split: reader en el hilo principal, writer compartido bajo Mutex
302 // entre la writer task (push channel) y el handler de inbound
303 // (que escribe Pong/Error). Mutex serializa writes; es OK porque
304 // la frecuencia de writes por sesión es baja.
305 let (mut reader, writer) = split(stream);
306 let writer = Arc::new(Mutex::new(writer));
307
308 // Writer task: drena el push channel.
309 let writer_for_push = writer.clone();
310 let writer_task = tokio::spawn(async move {
311 while let Some(frame) = rx.recv().await {
312 let mut w = writer_for_push.lock().await;
313 if write_frame(&mut *w, &frame).await.is_err() {
314 break;
315 }
316 }
317 });
318
319 // Reader loop principal.
320 let result: std::io::Result<()> = loop {
321 match read_frame(&mut reader).await {
322 Ok(frame) => {
323 match handle_inbound_frame(session_id, frame, &writer, &sessions).await {
324 Ok(true) => continue,
325 Ok(false) => break Ok(()),
326 Err(e) => break Err(e),
327 }
328 }
329 Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
330 debug!(session = %session_id, "cliente cerró sin Farewell");
331 break Ok(());
332 }
333 Err(e) => break Err(e),
334 }
335 };
336
337 // Cerrar writer: drop nuestro Arc y abortar la task. La task
338 // saldrá igual cuando rx se cierre por drop del último Sender,
339 // pero abortarla es más rápido que esperar a que próximo recv()
340 // observe el cierre.
341 drop(writer);
342 writer_task.abort();
343 let _ = writer_task.await;
344
345 result
346}
347
348async fn handle_inbound_frame<S>(
349 session_id: SessionId,
350 frame: Frame,
351 writer: &Arc<Mutex<WriteHalf<S>>>,
352 sessions: &SessionRegistry,
353) -> std::io::Result<bool>
354where
355 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
356{
357 match frame {
358 Frame::Ping(Ping { session }) if session == session_id => {
359 let pong = Pong {
360 timestamp_ms: now_ms(),
361 };
362 let mut w = writer.lock().await;
363 write_frame(&mut *w, &Frame::Pong(pong)).await?;
364 Ok(true)
365 }
366 Frame::Ping(_) => {
367 let mut w = writer.lock().await;
368 write_frame(
369 &mut *w,
370 &Frame::Error(HandshakeError::Unauthorized(
371 "session-id no coincide".into(),
372 )),
373 )
374 .await?;
375 Ok(true)
376 }
377 Frame::Farewell(Farewell { session }) if session == session_id => Ok(false),
378 Frame::Farewell(_) => {
379 let mut w = writer.lock().await;
380 write_frame(
381 &mut *w,
382 &Frame::Error(HandshakeError::Unauthorized(
383 "session-id no coincide".into(),
384 )),
385 )
386 .await?;
387 Ok(true)
388 }
389 Frame::ListSessions(crate::messages::ListSessions { session })
390 if session == session_id =>
391 {
392 let list = build_session_list(sessions).await;
393 let mut w = writer.lock().await;
394 write_frame(&mut *w, &Frame::SessionList(list)).await?;
395 Ok(true)
396 }
397 Frame::ListSessions(_) => {
398 let mut w = writer.lock().await;
399 write_frame(
400 &mut *w,
401 &Frame::Error(HandshakeError::Unauthorized(
402 "session-id no coincide".into(),
403 )),
404 )
405 .await?;
406 Ok(true)
407 }
408 _ => {
409 let mut w = writer.lock().await;
410 write_frame(
411 &mut *w,
412 &Frame::Error(HandshakeError::Rejected(
413 "frame inesperado tras handshake".into(),
414 )),
415 )
416 .await?;
417 Ok(true)
418 }
419 }
420}
421
422/// Snapshot read-only de la `SessionRegistry` proyectado a la forma
423/// de wire para el frame `SessionList`. Suelta el lock antes de
424/// retornar para que el writer del frame no contenga el mutex.
425async fn build_session_list(sessions: &SessionRegistry) -> crate::messages::SessionList {
426 let table = sessions.lock().await;
427 let entries = table
428 .iter()
429 .map(|(id, resolved)| crate::messages::SessionEntry {
430 session: *id,
431 label: resolved.card.label.clone(),
432 schema_version: resolved.card.schema_version,
433 outputs: resolved
434 .card
435 .flow
436 .output
437 .iter()
438 .map(|f| f.name.clone())
439 .collect(),
440 inputs: resolved
441 .card
442 .flow
443 .input
444 .iter()
445 .map(|f| f.name.clone())
446 .collect(),
447 conscious: resolved.wit.is_some(),
448 })
449 .collect();
450 crate::messages::SessionList { entries }
451}
452
453/// Limpieza atómica de las vistas registradas + (si net activo) retiro
454/// de anuncios DHT de los outputs de la Card. Se ejecuta tanto si la
455/// sesión cierra por Farewell, EOF, o error. Tras desregistrar, emite
456/// diffs a las sesiones que perdieron el match contra ésta.
457async fn cleanup(
458 session_id: SessionId,
459 sessions: &SessionRegistry,
460 push_table: &SessionTxTable,
461 last_matches: &LastMatches,
462 config: &ServerConfig,
463) {
464 // Tomamos la Card ANTES de borrarla — si net está configurado
465 // necesitamos sus outputs para llamar withdraw_outputs. `remove`
466 // devuelve el valor extraído.
467 let removed_card = sessions.lock().await.remove(&session_id);
468 push_table.lock().await.remove(&session_id);
469 last_matches.lock().await.remove(&session_id);
470 if let Some(broker) = &config.broker {
471 broker.lock().await.unregister(session_id);
472 }
473 if let (Some(net), Some(resolved)) = (&config.net, removed_card) {
474 crate::network::withdraw_outputs(net, &resolved.card);
475 }
476 broadcast_match_diffs(push_table, last_matches, config).await;
477}
478
479/// Recomputa los matches para todas las sesiones registradas y empuja
480/// `MatchEvent::Available` / `MatchEvent::Lost` por las que cambiaron
481/// respecto al último estado conocido.
482async fn broadcast_match_diffs(
483 push_table: &SessionTxTable,
484 last_matches: &LastMatches,
485 config: &ServerConfig,
486) {
487 let broker = match &config.broker {
488 Some(b) => b,
489 None => return,
490 };
491
492 let b = broker.lock().await;
493 let push_table = push_table.lock().await;
494 let mut last = last_matches.lock().await;
495
496 debug!(
497 target: "brahman_handshake::broadcast",
498 cards = b.len(),
499 push_subscribers = push_table.len(),
500 "broadcast_match_diffs"
501 );
502
503 let cards: Vec<_> = b.cards().cloned().collect();
504
505 for cons in &cards {
506 let cons_session = cons.session;
507 let tx = match push_table.get(&cons_session) {
508 Some(tx) => tx,
509 None => continue,
510 };
511 let cons_last = last.entry(cons_session).or_default();
512
513 for input in &cons.inputs {
514 let new_match = b.find_producer_for(cons_session, &input.name);
515 let new_endpoint = new_match.as_ref().map(|m| m.producer.clone());
516 let old_endpoint = cons_last.get(&input.name).cloned();
517
518 if old_endpoint == new_endpoint {
519 continue;
520 }
521
522 if let Some(m) = &new_match {
523 let producer_service_socket = b
524 .cards()
525 .find(|c| c.session == m.producer.session)
526 .and_then(|c| c.service_socket.clone());
527 let event = MatchEvent {
528 kind: MatchEventKind::Available,
529 consumer_flow: input.name.clone(),
530 producer_session: m.producer.session,
531 producer_label: m.producer_label.clone(),
532 producer_flow: m.producer.flow_name.clone(),
533 ty: m.ty.clone(),
534 via: m.via,
535 pinned: m.pinned,
536 producer_service_socket,
537 };
538 let send_res = tx.try_send(Frame::MatchEvent(event));
539 debug!(
540 target: "brahman_handshake::broadcast",
541 consumer = %cons_session,
542 flow = %input.name,
543 producer = %m.producer_label,
544 result = ?send_res.as_ref().map(|_| "ok").unwrap_or_else(|e| match e {
545 tokio::sync::mpsc::error::TrySendError::Full(_) => "full",
546 tokio::sync::mpsc::error::TrySendError::Closed(_) => "closed",
547 }),
548 "Available pushed"
549 );
550 } else {
551 let event = MatchEvent {
552 kind: MatchEventKind::Lost,
553 consumer_flow: input.name.clone(),
554 producer_session: Ulid::nil(),
555 producer_label: String::new(),
556 producer_flow: String::new(),
557 ty: input.ty.clone(),
558 via: brahman_broker::MatchStrategy::Exact,
559 pinned: false,
560 producer_service_socket: None,
561 };
562 let _ = tx.try_send(Frame::MatchEvent(event));
563 }
564
565 if let Some(ep) = new_endpoint {
566 cons_last.insert(input.name.clone(), ep);
567 } else {
568 cons_last.remove(&input.name);
569 }
570 }
571 }
572}
573
574/// Lee el Hello, valida (incluyendo firma cuando aplica), registra la
575/// sesión y emite HelloAck.
576async fn do_handshake<S>(
577 stream: &mut S,
578 config: &ServerConfig,
579 sessions: &SessionRegistry,
580 expected_peer: Option<PeerId>,
581) -> std::io::Result<Option<SessionId>>
582where
583 S: AsyncRead + AsyncWrite + Unpin,
584{
585 let frame = read_frame(stream).await?;
586 let hello = match frame {
587 Frame::Hello(h) => h,
588 _ => {
589 write_frame(
590 stream,
591 &Frame::Error(HandshakeError::Rejected(
592 "primer frame debe ser Hello".into(),
593 )),
594 )
595 .await?;
596 return Ok(None);
597 }
598 };
599
600 if let Some(err) = validate_hello(&hello) {
601 write_frame(stream, &Frame::Error(err)).await?;
602 return Ok(None);
603 }
604
605 // Identity cert (multi-key identity, opcional): si el cliente
606 // adjuntó cert, la "identidad lógica" del peer es el master
607 // derivado del cert (estable across rotaciones), no el session
608 // peer_id (efímero). Sin cert, fallback al modelo de Fase 3
609 // (logical = session). Esto permite migración gradual y
610 // backwards compatibility con clientes que no usan identity.
611 let logical_peer = if let (Some(session_peer), Some(cert)) =
612 (expected_peer, &hello.identity_cert)
613 {
614 let session_pk_bytes: &[u8] = match &hello.signature {
615 Some(sig) => &sig.public_key,
616 None => {
617 write_frame(
618 stream,
619 &Frame::Error(HandshakeError::Unauthorized(
620 "Hello con identity_cert requiere también signature".into(),
621 )),
622 )
623 .await?;
624 return Ok(None);
625 }
626 };
627 match cert.verify_against_session(session_pk_bytes) {
628 Ok(master_peer) => {
629 debug!(
630 session = %session_peer,
631 master = %master_peer,
632 "identity cert válido — policy se evalúa contra master_peer"
633 );
634 Some(master_peer)
635 }
636 Err(e) => {
637 write_frame(
638 stream,
639 &Frame::Error(HandshakeError::Unauthorized(format!(
640 "identity cert inválido: {e}"
641 ))),
642 )
643 .await?;
644 debug!(peer = %session_peer, error = %e, "cert rechazado");
645 return Ok(None);
646 }
647 }
648 } else {
649 expected_peer
650 };
651
652 // Policy gate (path libp2p): si está configurada, el peer
653 // autenticado debe pasar la política (deny first, luego allow).
654 // El peer evaluado es `logical_peer`: master si hay cert,
655 // session si no. Se chequea ANTES de la firma porque es
656 // comparación O(log n) sin crypto. La política no se aplica
657 // al path Unix (autenticación por SO_PEERCRED, no por PeerId).
658 if let (Some(peer), Some(policy)) = (logical_peer, &config.policy) {
659 let decision = policy.evaluate(&peer);
660 if !decision.is_admitted() {
661 write_frame(
662 stream,
663 &Frame::Error(HandshakeError::Unauthorized(format!(
664 "peer {peer}: {}",
665 decision.reason()
666 ))),
667 )
668 .await?;
669 debug!(peer = %peer, reason = decision.reason(), "rechazado por policy");
670 return Ok(None);
671 }
672 }
673
674 // Trust gate: en path libp2p (expected_peer = Some), exigir
675 // firma cuya public key derive al peer autenticado por Noise.
676 // En path Unix (expected_peer = None), si la firma viene se
677 // verifica anyway por defensa en profundidad — no es un error
678 // que esté ahí, pero si está debe ser válida.
679 if let Some(peer) = expected_peer {
680 let sig = match &hello.signature {
681 Some(s) => s,
682 None => {
683 write_frame(
684 stream,
685 &Frame::Error(HandshakeError::Unauthorized(
686 "Hello sin firma en conexión remota libp2p".into(),
687 )),
688 )
689 .await?;
690 return Ok(None);
691 }
692 };
693 if let Err(e) = crate::signature::verify_hello(sig, &hello.card, &hello.wit, peer) {
694 write_frame(
695 stream,
696 &Frame::Error(HandshakeError::Unauthorized(format!("firma inválida: {e}"))),
697 )
698 .await?;
699 debug!(peer = %peer, error = %e, "firma rechazada");
700 return Ok(None);
701 }
702 } else if let Some(sig) = &hello.signature {
703 // Firma presente en path local: no exigida pero verificada.
704 // Si está y no valida, es un signo de Hello mal-construido y
705 // rechazamos por seguridad.
706 // Para Unix no tenemos peer_id contra el cual comparar; se
707 // verifica sólo la consistencia interna (firma sobre payload
708 // con la public_key declarada).
709 match brahman_net::PublicKey::try_decode_protobuf(&sig.public_key) {
710 Ok(pk) => {
711 let payload = match postcard::to_allocvec(&(
712 crate::signature::SIGNATURE_VERSION,
713 &hello.card,
714 &hello.wit,
715 )) {
716 Ok(b) => b,
717 Err(_) => {
718 write_frame(
719 stream,
720 &Frame::Error(HandshakeError::Internal(
721 "no pude codificar payload para verificar firma".into(),
722 )),
723 )
724 .await?;
725 return Ok(None);
726 }
727 };
728 if !pk.verify(&payload, &sig.signature) {
729 write_frame(
730 stream,
731 &Frame::Error(HandshakeError::Unauthorized(
732 "firma del Hello presente pero inválida".into(),
733 )),
734 )
735 .await?;
736 return Ok(None);
737 }
738 }
739 Err(e) => {
740 write_frame(
741 stream,
742 &Frame::Error(HandshakeError::Unauthorized(format!(
743 "public_key inválida en firma: {e}"
744 ))),
745 )
746 .await?;
747 return Ok(None);
748 }
749 }
750 }
751
752 let session_id = Ulid::new();
753 let card: Card = hello.card.into();
754 register_session(session_id, card, hello.wit, config, sessions).await;
755
756 let ack = HelloAck {
757 server_version: crate::HANDSHAKE_VERSION.to_string(),
758 protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
759 session: session_id,
760 init_attached: config.init_attached,
761 };
762 write_frame(stream, &Frame::HelloAck(ack)).await?;
763 debug!(session = %session_id, "handshake completado");
764 Ok(Some(session_id))
765}
766
767async fn register_session(
768 session_id: SessionId,
769 card: Card,
770 wit: Option<WitInterface>,
771 config: &ServerConfig,
772 sessions: &SessionRegistry,
773) {
774 if let Some(broker) = &config.broker {
775 broker
776 .lock()
777 .await
778 .register(session_id, &card, wit.clone());
779 }
780 // Si el server tiene net configurado, anunciar los outputs al
781 // DHT para que peers remotos puedan descubrirlos. Idempotente
782 // y best-effort — fallos de Kad no propagan al handshake.
783 if let Some(net) = &config.net {
784 crate::network::announce_outputs(net, &card);
785 }
786 let resolved = match wit {
787 Some(w) => ResolvedCard::from_conscious(card, w),
788 None => ResolvedCard::from_agnostic(card),
789 };
790 sessions.lock().await.insert(session_id, resolved);
791}
792
793fn validate_hello(hello: &Hello) -> Option<HandshakeError> {
794 if hello.schema_version != CARD_SCHEMA_VERSION {
795 return Some(HandshakeError::SchemaMismatch {
796 client: hello.schema_version,
797 server: CARD_SCHEMA_VERSION,
798 });
799 }
800 if hello.protocol_version != brahman_card::PROTOCOL_VERSION {
801 return Some(HandshakeError::ProtocolMismatch(format!(
802 "cliente={}, servidor={}",
803 hello.protocol_version,
804 brahman_card::PROTOCOL_VERSION
805 )));
806 }
807 let as_card: Card = Card::from(hello.card.clone());
808 if let Err(e) = as_card.validate() {
809 return Some(HandshakeError::InvalidCard(e.to_string()));
810 }
811 None
812}
813
814fn now_ms() -> u64 {
815 SystemTime::now()
816 .duration_since(UNIX_EPOCH)
817 .map(|d| d.as_millis() as u64)
818 .unwrap_or(0)
819}