⎇ 72758e75ce515f355f27615671c95ecaf8c2a8bc
crates/core/brahman-handshake/src/server.rs
History
Who touched what
raw
feat(brahman-handshake): ListMatches endpoint + timeline en 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 broker_for_loop = config.broker.clone();
321 let result: std::io::Result<()> = loop {
322 match read_frame(&mut reader).await {
323 Ok(frame) => {
324 match handle_inbound_frame(
325 session_id,
326 frame,
327 &writer,
328 &sessions,
329 broker_for_loop.as_ref(),
330 )
331 .await
332 {
333 Ok(true) => continue,
334 Ok(false) => break Ok(()),
335 Err(e) => break Err(e),
336 }
337 }
338 Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
339 debug!(session = %session_id, "cliente cerró sin Farewell");
340 break Ok(());
341 }
342 Err(e) => break Err(e),
343 }
344 };
345
346 // Cerrar writer: drop nuestro Arc y abortar la task. La task
347 // saldrá igual cuando rx se cierre por drop del último Sender,
348 // pero abortarla es más rápido que esperar a que próximo recv()
349 // observe el cierre.
350 drop(writer);
351 writer_task.abort();
352 let _ = writer_task.await;
353
354 result
355}
356
357async fn handle_inbound_frame<S>(
358 session_id: SessionId,
359 frame: Frame,
360 writer: &Arc<Mutex<WriteHalf<S>>>,
361 sessions: &SessionRegistry,
362 broker_for_match: Option<&SharedBroker>,
363) -> std::io::Result<bool>
364where
365 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
366{
367 match frame {
368 Frame::Ping(Ping { session }) if session == session_id => {
369 let pong = Pong {
370 timestamp_ms: now_ms(),
371 };
372 let mut w = writer.lock().await;
373 write_frame(&mut *w, &Frame::Pong(pong)).await?;
374 Ok(true)
375 }
376 Frame::Ping(_) => {
377 let mut w = writer.lock().await;
378 write_frame(
379 &mut *w,
380 &Frame::Error(HandshakeError::Unauthorized(
381 "session-id no coincide".into(),
382 )),
383 )
384 .await?;
385 Ok(true)
386 }
387 Frame::Farewell(Farewell { session }) if session == session_id => Ok(false),
388 Frame::Farewell(_) => {
389 let mut w = writer.lock().await;
390 write_frame(
391 &mut *w,
392 &Frame::Error(HandshakeError::Unauthorized(
393 "session-id no coincide".into(),
394 )),
395 )
396 .await?;
397 Ok(true)
398 }
399 Frame::ListSessions(crate::messages::ListSessions { session })
400 if session == session_id =>
401 {
402 let list = build_session_list(sessions).await;
403 let mut w = writer.lock().await;
404 write_frame(&mut *w, &Frame::SessionList(list)).await?;
405 Ok(true)
406 }
407 Frame::ListSessions(_) => {
408 let mut w = writer.lock().await;
409 write_frame(
410 &mut *w,
411 &Frame::Error(HandshakeError::Unauthorized(
412 "session-id no coincide".into(),
413 )),
414 )
415 .await?;
416 Ok(true)
417 }
418 Frame::ListMatches(crate::messages::ListMatches { session })
419 if session == session_id =>
420 {
421 let matches = match &broker_for_match {
422 Some(b) => b.lock().await.all_matches(),
423 None => Vec::new(),
424 };
425 let mut w = writer.lock().await;
426 write_frame(
427 &mut *w,
428 &Frame::MatchList(crate::messages::MatchList { matches }),
429 )
430 .await?;
431 Ok(true)
432 }
433 Frame::ListMatches(_) => {
434 let mut w = writer.lock().await;
435 write_frame(
436 &mut *w,
437 &Frame::Error(HandshakeError::Unauthorized(
438 "session-id no coincide".into(),
439 )),
440 )
441 .await?;
442 Ok(true)
443 }
444 _ => {
445 let mut w = writer.lock().await;
446 write_frame(
447 &mut *w,
448 &Frame::Error(HandshakeError::Rejected(
449 "frame inesperado tras handshake".into(),
450 )),
451 )
452 .await?;
453 Ok(true)
454 }
455 }
456}
457
458/// Snapshot read-only de la `SessionRegistry` proyectado a la forma
459/// de wire para el frame `SessionList`. Suelta el lock antes de
460/// retornar para que el writer del frame no contenga el mutex.
461async fn build_session_list(sessions: &SessionRegistry) -> crate::messages::SessionList {
462 let table = sessions.lock().await;
463 let entries = table
464 .iter()
465 .map(|(id, resolved)| crate::messages::SessionEntry {
466 session: *id,
467 label: resolved.card.label.clone(),
468 schema_version: resolved.card.schema_version,
469 outputs: resolved
470 .card
471 .flow
472 .output
473 .iter()
474 .map(|f| f.name.clone())
475 .collect(),
476 inputs: resolved
477 .card
478 .flow
479 .input
480 .iter()
481 .map(|f| f.name.clone())
482 .collect(),
483 conscious: resolved.wit.is_some(),
484 })
485 .collect();
486 crate::messages::SessionList { entries }
487}
488
489/// Limpieza atómica de las vistas registradas + (si net activo) retiro
490/// de anuncios DHT de los outputs de la Card. Se ejecuta tanto si la
491/// sesión cierra por Farewell, EOF, o error. Tras desregistrar, emite
492/// diffs a las sesiones que perdieron el match contra ésta.
493async fn cleanup(
494 session_id: SessionId,
495 sessions: &SessionRegistry,
496 push_table: &SessionTxTable,
497 last_matches: &LastMatches,
498 config: &ServerConfig,
499) {
500 // Tomamos la Card ANTES de borrarla — si net está configurado
501 // necesitamos sus outputs para llamar withdraw_outputs. `remove`
502 // devuelve el valor extraído.
503 let removed_card = sessions.lock().await.remove(&session_id);
504 push_table.lock().await.remove(&session_id);
505 last_matches.lock().await.remove(&session_id);
506 if let Some(broker) = &config.broker {
507 broker.lock().await.unregister(session_id);
508 }
509 if let (Some(net), Some(resolved)) = (&config.net, removed_card) {
510 crate::network::withdraw_outputs(net, &resolved.card);
511 }
512 broadcast_match_diffs(push_table, last_matches, config).await;
513}
514
515/// Recomputa los matches para todas las sesiones registradas y empuja
516/// `MatchEvent::Available` / `MatchEvent::Lost` por las que cambiaron
517/// respecto al último estado conocido.
518async fn broadcast_match_diffs(
519 push_table: &SessionTxTable,
520 last_matches: &LastMatches,
521 config: &ServerConfig,
522) {
523 let broker = match &config.broker {
524 Some(b) => b,
525 None => return,
526 };
527
528 let b = broker.lock().await;
529 let push_table = push_table.lock().await;
530 let mut last = last_matches.lock().await;
531
532 debug!(
533 target: "brahman_handshake::broadcast",
534 cards = b.len(),
535 push_subscribers = push_table.len(),
536 "broadcast_match_diffs"
537 );
538
539 let cards: Vec<_> = b.cards().cloned().collect();
540
541 for cons in &cards {
542 let cons_session = cons.session;
543 let tx = match push_table.get(&cons_session) {
544 Some(tx) => tx,
545 None => continue,
546 };
547 let cons_last = last.entry(cons_session).or_default();
548
549 for input in &cons.inputs {
550 let new_match = b.find_producer_for(cons_session, &input.name);
551 let new_endpoint = new_match.as_ref().map(|m| m.producer.clone());
552 let old_endpoint = cons_last.get(&input.name).cloned();
553
554 if old_endpoint == new_endpoint {
555 continue;
556 }
557
558 if let Some(m) = &new_match {
559 let producer_service_socket = b
560 .cards()
561 .find(|c| c.session == m.producer.session)
562 .and_then(|c| c.service_socket.clone());
563 let event = MatchEvent {
564 kind: MatchEventKind::Available,
565 consumer_flow: input.name.clone(),
566 producer_session: m.producer.session,
567 producer_label: m.producer_label.clone(),
568 producer_flow: m.producer.flow_name.clone(),
569 ty: m.ty.clone(),
570 via: m.via,
571 pinned: m.pinned,
572 producer_service_socket,
573 };
574 let send_res = tx.try_send(Frame::MatchEvent(event));
575 debug!(
576 target: "brahman_handshake::broadcast",
577 consumer = %cons_session,
578 flow = %input.name,
579 producer = %m.producer_label,
580 result = ?send_res.as_ref().map(|_| "ok").unwrap_or_else(|e| match e {
581 tokio::sync::mpsc::error::TrySendError::Full(_) => "full",
582 tokio::sync::mpsc::error::TrySendError::Closed(_) => "closed",
583 }),
584 "Available pushed"
585 );
586 } else {
587 let event = MatchEvent {
588 kind: MatchEventKind::Lost,
589 consumer_flow: input.name.clone(),
590 producer_session: Ulid::nil(),
591 producer_label: String::new(),
592 producer_flow: String::new(),
593 ty: input.ty.clone(),
594 via: brahman_broker::MatchStrategy::Exact,
595 pinned: false,
596 producer_service_socket: None,
597 };
598 let _ = tx.try_send(Frame::MatchEvent(event));
599 }
600
601 if let Some(ep) = new_endpoint {
602 cons_last.insert(input.name.clone(), ep);
603 } else {
604 cons_last.remove(&input.name);
605 }
606 }
607 }
608}
609
610/// Lee el Hello, valida (incluyendo firma cuando aplica), registra la
611/// sesión y emite HelloAck.
612async fn do_handshake<S>(
613 stream: &mut S,
614 config: &ServerConfig,
615 sessions: &SessionRegistry,
616 expected_peer: Option<PeerId>,
617) -> std::io::Result<Option<SessionId>>
618where
619 S: AsyncRead + AsyncWrite + Unpin,
620{
621 let frame = read_frame(stream).await?;
622 let hello = match frame {
623 Frame::Hello(h) => h,
624 _ => {
625 write_frame(
626 stream,
627 &Frame::Error(HandshakeError::Rejected(
628 "primer frame debe ser Hello".into(),
629 )),
630 )
631 .await?;
632 return Ok(None);
633 }
634 };
635
636 if let Some(err) = validate_hello(&hello) {
637 write_frame(stream, &Frame::Error(err)).await?;
638 return Ok(None);
639 }
640
641 // Identity cert (multi-key identity, opcional): si el cliente
642 // adjuntó cert, la "identidad lógica" del peer es el master
643 // derivado del cert (estable across rotaciones), no el session
644 // peer_id (efímero). Sin cert, fallback al modelo de Fase 3
645 // (logical = session). Esto permite migración gradual y
646 // backwards compatibility con clientes que no usan identity.
647 let logical_peer = if let (Some(session_peer), Some(cert)) =
648 (expected_peer, &hello.identity_cert)
649 {
650 let session_pk_bytes: &[u8] = match &hello.signature {
651 Some(sig) => &sig.public_key,
652 None => {
653 write_frame(
654 stream,
655 &Frame::Error(HandshakeError::Unauthorized(
656 "Hello con identity_cert requiere también signature".into(),
657 )),
658 )
659 .await?;
660 return Ok(None);
661 }
662 };
663 match cert.verify_against_session(session_pk_bytes) {
664 Ok(master_peer) => {
665 debug!(
666 session = %session_peer,
667 master = %master_peer,
668 "identity cert válido — policy se evalúa contra master_peer"
669 );
670 Some(master_peer)
671 }
672 Err(e) => {
673 write_frame(
674 stream,
675 &Frame::Error(HandshakeError::Unauthorized(format!(
676 "identity cert inválido: {e}"
677 ))),
678 )
679 .await?;
680 debug!(peer = %session_peer, error = %e, "cert rechazado");
681 return Ok(None);
682 }
683 }
684 } else {
685 expected_peer
686 };
687
688 // Policy gate (path libp2p): si está configurada, el peer
689 // autenticado debe pasar la política (deny first, luego allow).
690 // El peer evaluado es `logical_peer`: master si hay cert,
691 // session si no. Se chequea ANTES de la firma porque es
692 // comparación O(log n) sin crypto. La política no se aplica
693 // al path Unix (autenticación por SO_PEERCRED, no por PeerId).
694 if let (Some(peer), Some(policy)) = (logical_peer, &config.policy) {
695 let decision = policy.evaluate(&peer);
696 if !decision.is_admitted() {
697 write_frame(
698 stream,
699 &Frame::Error(HandshakeError::Unauthorized(format!(
700 "peer {peer}: {}",
701 decision.reason()
702 ))),
703 )
704 .await?;
705 debug!(peer = %peer, reason = decision.reason(), "rechazado por policy");
706 return Ok(None);
707 }
708 }
709
710 // Trust gate: en path libp2p (expected_peer = Some), exigir
711 // firma cuya public key derive al peer autenticado por Noise.
712 // En path Unix (expected_peer = None), si la firma viene se
713 // verifica anyway por defensa en profundidad — no es un error
714 // que esté ahí, pero si está debe ser válida.
715 if let Some(peer) = expected_peer {
716 let sig = match &hello.signature {
717 Some(s) => s,
718 None => {
719 write_frame(
720 stream,
721 &Frame::Error(HandshakeError::Unauthorized(
722 "Hello sin firma en conexión remota libp2p".into(),
723 )),
724 )
725 .await?;
726 return Ok(None);
727 }
728 };
729 if let Err(e) = crate::signature::verify_hello(sig, &hello.card, &hello.wit, peer) {
730 write_frame(
731 stream,
732 &Frame::Error(HandshakeError::Unauthorized(format!("firma inválida: {e}"))),
733 )
734 .await?;
735 debug!(peer = %peer, error = %e, "firma rechazada");
736 return Ok(None);
737 }
738 } else if let Some(sig) = &hello.signature {
739 // Firma presente en path local: no exigida pero verificada.
740 // Si está y no valida, es un signo de Hello mal-construido y
741 // rechazamos por seguridad.
742 // Para Unix no tenemos peer_id contra el cual comparar; se
743 // verifica sólo la consistencia interna (firma sobre payload
744 // con la public_key declarada).
745 match brahman_net::PublicKey::try_decode_protobuf(&sig.public_key) {
746 Ok(pk) => {
747 let payload = match postcard::to_allocvec(&(
748 crate::signature::SIGNATURE_VERSION,
749 &hello.card,
750 &hello.wit,
751 )) {
752 Ok(b) => b,
753 Err(_) => {
754 write_frame(
755 stream,
756 &Frame::Error(HandshakeError::Internal(
757 "no pude codificar payload para verificar firma".into(),
758 )),
759 )
760 .await?;
761 return Ok(None);
762 }
763 };
764 if !pk.verify(&payload, &sig.signature) {
765 write_frame(
766 stream,
767 &Frame::Error(HandshakeError::Unauthorized(
768 "firma del Hello presente pero inválida".into(),
769 )),
770 )
771 .await?;
772 return Ok(None);
773 }
774 }
775 Err(e) => {
776 write_frame(
777 stream,
778 &Frame::Error(HandshakeError::Unauthorized(format!(
779 "public_key inválida en firma: {e}"
780 ))),
781 )
782 .await?;
783 return Ok(None);
784 }
785 }
786 }
787
788 let session_id = Ulid::new();
789 let card: Card = hello.card.into();
790 register_session(session_id, card, hello.wit, config, sessions).await;
791
792 let ack = HelloAck {
793 server_version: crate::HANDSHAKE_VERSION.to_string(),
794 protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
795 session: session_id,
796 init_attached: config.init_attached,
797 };
798 write_frame(stream, &Frame::HelloAck(ack)).await?;
799 debug!(session = %session_id, "handshake completado");
800 Ok(Some(session_id))
801}
802
803async fn register_session(
804 session_id: SessionId,
805 card: Card,
806 wit: Option<WitInterface>,
807 config: &ServerConfig,
808 sessions: &SessionRegistry,
809) {
810 if let Some(broker) = &config.broker {
811 broker
812 .lock()
813 .await
814 .register(session_id, &card, wit.clone());
815 }
816 // Si el server tiene net configurado, anunciar los outputs al
817 // DHT para que peers remotos puedan descubrirlos. Idempotente
818 // y best-effort — fallos de Kad no propagan al handshake.
819 if let Some(net) = &config.net {
820 crate::network::announce_outputs(net, &card);
821 }
822 let resolved = match wit {
823 Some(w) => ResolvedCard::from_conscious(card, w),
824 None => ResolvedCard::from_agnostic(card),
825 };
826 sessions.lock().await.insert(session_id, resolved);
827}
828
829fn validate_hello(hello: &Hello) -> Option<HandshakeError> {
830 if hello.schema_version != CARD_SCHEMA_VERSION {
831 return Some(HandshakeError::SchemaMismatch {
832 client: hello.schema_version,
833 server: CARD_SCHEMA_VERSION,
834 });
835 }
836 if hello.protocol_version != brahman_card::PROTOCOL_VERSION {
837 return Some(HandshakeError::ProtocolMismatch(format!(
838 "cliente={}, servidor={}",
839 hello.protocol_version,
840 brahman_card::PROTOCOL_VERSION
841 )));
842 }
843 let as_card: Card = Card::from(hello.card.clone());
844 if let Err(e) = as_card.validate() {
845 return Some(HandshakeError::InvalidCard(e.to_string()));
846 }
847 None
848}
849
850fn now_ms() -> u64 {
851 SystemTime::now()
852 .duration_since(UNIX_EPOCH)
853 .map(|d| d.as_millis() as u64)
854 .unwrap_or(0)
855}