tawasuyu/brahman

1//! Cliente de handshake. Conecta a un Unix socket y mantiene la sesión.
2
3use std::collections::VecDeque;
4use std::path::Path;
5use std::time::Duration;
6
7use brahman_card::{Card, WitInterface, CARD_SCHEMA_VERSION};
8use brahman_net::Keypair;
9use thiserror::Error;
10use tokio::io::{AsyncRead, AsyncWrite};
11use tokio::net::UnixStream;
12
13use crate::codec::{read_frame, write_frame};
14use crate::identity::SessionCert;
15use crate::messages::{Farewell, Frame, HandshakeError, Hello, HelloAck, MatchEvent, Ping, SessionId};
16use crate::signature::{sign_hello, SignatureError};
17
18/// Errores del cliente.
19#[derive(Debug, Error)]
20pub enum ClientError {
21    #[error("E/S: {0}")]
22    Io(#[from] std::io::Error),
23
24    /// El servidor respondió con un error explícito.
25    #[error("servidor: {0}")]
26    Server(#[source] HandshakeError),
27
28    /// El servidor envió un frame que no esperábamos en este punto del protocolo.
29    #[error("frame inesperado: {got}")]
30    UnexpectedFrame { got: &'static str },
31
32    /// La Card que el cliente intentó enviar no pasa su propia validación.
33    #[error("card inválida pre-envío: {0}")]
34    InvalidCard(String),
35
36    /// Firma del Hello falló al construirse (rara — sólo puede pasar
37    /// si la keypair pasada está en un estado inválido).
38    #[error("firma del Hello falló: {0}")]
39    Signature(#[from] SignatureError),
40}
41
42/// Cliente conectado y autenticado. Tras `connect` ya completó el handshake
43/// y tiene su `SessionId`. Los `MatchEvent` recibidos durante operaciones
44/// request/response se buferean en `pending_events` y se obtienen vía
45/// [`Client::take_event`] o [`Client::await_event`].
46///
47/// Genérico sobre el transport (`AsyncRead + AsyncWrite + Unpin + Send`):
48/// funciona indistintamente sobre `UnixStream` (path local) o sobre un
49/// stream libp2p wrapped con `tokio_util::compat` (path remoto, vía
50/// `brahman_handshake::network`).
51#[derive(Debug)]
52pub struct Client<S = UnixStream> {
53    stream: S,
54    session: SessionId,
55    server_info: HelloAck,
56    pending_events: VecDeque<MatchEvent>,
57}
58
59impl Client<UnixStream> {
60    /// Conecta como módulo agnóstico (sin WIT) sobre Unix socket.
61    /// Equivalente a `connect_with(path, card, None)`.
62    pub async fn connect(path: impl AsRef<Path>, card: Card) -> Result<Self, ClientError> {
63        Self::connect_with(path, card, None).await
64    }
65
66    /// Conecta al socket Unix enviando Hello con la Card dada y
67    /// opcionalmente una `WitInterface` ya extraída. Si `wit` es `Some`,
68    /// el server registra el módulo como "consciente".
69    pub async fn connect_with(
70        path: impl AsRef<Path>,
71        card: Card,
72        wit: Option<WitInterface>,
73    ) -> Result<Self, ClientError> {
74        let stream = UnixStream::connect(path).await?;
75        Self::connect_with_stream(stream, card, wit).await
76    }
77}
78
79impl<S> Client<S>
80where
81    S: AsyncRead + AsyncWrite + Unpin + Send,
82{
83    /// Constructor genérico sobre un stream ya abierto, **sin firma**.
84    /// Apto para path Unix (donde SO_PEERCRED del kernel ya autentica)
85    /// o tests in-memory. Para libp2p remoto usá
86    /// [`connect_with_stream_signed`](Self::connect_with_stream_signed) —
87    /// el server libp2p rechaza Hello sin firma.
88    pub async fn connect_with_stream(
89        stream: S,
90        card: Card,
91        wit: Option<WitInterface>,
92    ) -> Result<Self, ClientError> {
93        Self::connect_inner(stream, card, wit, None, None).await
94    }
95
96    /// Igual que `connect_with_stream` pero firma el Hello con
97    /// `keypair`. Usar para conexiones libp2p donde el server exige
98    /// firma. La public key derivada de `keypair` debe coincidir con
99    /// el `peer_id` libp2p autenticado por Noise — típicamente la
100    /// keypair pasada a [`brahman_net::BrahmanNet::with_keypair`].
101    pub async fn connect_with_stream_signed(
102        stream: S,
103        card: Card,
104        wit: Option<WitInterface>,
105        keypair: &Keypair,
106    ) -> Result<Self, ClientError> {
107        Self::connect_inner(stream, card, wit, Some(keypair), None).await
108    }
109
110    /// Igual que `connect_with_stream_signed` pero además adjunta un
111    /// `SessionCert` que vincula la session keypair a una identity
112    /// master estable. El server, al recibir el cert, evalúa la
113    /// política de admisión contra el `master_peer_id` (no contra
114    /// el session peer_id) — permitiendo rotar la session sin perder
115    /// la identidad reconocida en allowlists remotas.
116    pub async fn connect_with_stream_signed_with_cert(
117        stream: S,
118        card: Card,
119        wit: Option<WitInterface>,
120        session_keypair: &Keypair,
121        identity_cert: SessionCert,
122    ) -> Result<Self, ClientError> {
123        Self::connect_inner(stream, card, wit, Some(session_keypair), Some(identity_cert)).await
124    }
125
126    async fn connect_inner(
127        mut stream: S,
128        card: Card,
129        wit: Option<WitInterface>,
130        keypair: Option<&Keypair>,
131        identity_cert: Option<SessionCert>,
132    ) -> Result<Self, ClientError> {
133        card.validate()
134            .map_err(|e| ClientError::InvalidCard(e.to_string()))?;
135
136        let wire_card = brahman_card::WireCard::from(card);
137        let signature = match keypair {
138            Some(kp) => Some(sign_hello(kp, &wire_card, &wit)?),
139            None => None,
140        };
141
142        let hello = Hello {
143            schema_version: CARD_SCHEMA_VERSION,
144            protocol_version: brahman_card::PROTOCOL_VERSION.to_string(),
145            card: wire_card,
146            wit,
147            signature,
148            identity_cert,
149        };
150        write_frame(&mut stream, &Frame::Hello(hello)).await?;
151
152        let frame = read_frame(&mut stream).await?;
153        let ack = match frame {
154            Frame::HelloAck(a) => a,
155            Frame::Error(e) => return Err(ClientError::Server(e)),
156            Frame::Hello(_) => return Err(ClientError::UnexpectedFrame { got: "Hello" }),
157            Frame::Ping(_) => return Err(ClientError::UnexpectedFrame { got: "Ping" }),
158            Frame::Pong(_) => return Err(ClientError::UnexpectedFrame { got: "Pong" }),
159            Frame::Farewell(_) => return Err(ClientError::UnexpectedFrame { got: "Farewell" }),
160            Frame::MatchEvent(_) => {
161                return Err(ClientError::UnexpectedFrame {
162                    got: "MatchEvent (pre-handshake)",
163                });
164            }
165            Frame::ListSessions(_) => {
166                return Err(ClientError::UnexpectedFrame {
167                    got: "ListSessions (pre-handshake)",
168                });
169            }
170            Frame::SessionList(_) => {
171                return Err(ClientError::UnexpectedFrame {
172                    got: "SessionList (pre-handshake)",
173                });
174            }
175            Frame::ListMatches(_) => {
176                return Err(ClientError::UnexpectedFrame {
177                    got: "ListMatches (pre-handshake)",
178                });
179            }
180            Frame::MatchList(_) => {
181                return Err(ClientError::UnexpectedFrame {
182                    got: "MatchList (pre-handshake)",
183                });
184            }
185        };
186        Ok(Self {
187            stream,
188            session: ack.session,
189            server_info: ack,
190            pending_events: VecDeque::new(),
191        })
192    }
193
194    /// `SessionId` asignado por el servidor.
195    pub fn session(&self) -> SessionId {
196        self.session
197    }
198
199    /// Información del servidor recibida en el handshake.
200    pub fn server_info(&self) -> &HelloAck {
201        &self.server_info
202    }
203
204    /// Envía un Ping y devuelve el timestamp del servidor. Los frames
205    /// `MatchEvent` que lleguen mezclados se buferean en `pending_events`.
206    pub async fn ping(&mut self) -> Result<u64, ClientError> {
207        write_frame(
208            &mut self.stream,
209            &Frame::Ping(Ping {
210                session: self.session,
211            }),
212        )
213        .await?;
214        loop {
215            match read_frame(&mut self.stream).await? {
216                Frame::Pong(p) => return Ok(p.timestamp_ms),
217                Frame::MatchEvent(ev) => self.pending_events.push_back(ev),
218                Frame::Error(e) => return Err(ClientError::Server(e)),
219                _ => return Err(ClientError::UnexpectedFrame { got: "non-pong" }),
220            }
221        }
222    }
223
224    /// Saca un evento pendiente del buffer, sin bloquear ni leer del wire.
225    pub fn take_event(&mut self) -> Option<MatchEvent> {
226        self.pending_events.pop_front()
227    }
228
229    /// Espera un `MatchEvent` con timeout. Drena primero el buffer; si
230    /// está vacío, lee del wire hasta el timeout. Otros frames recibidos
231    /// (Pong huérfano, Error) cortan la espera con error.
232    pub async fn await_event(
233        &mut self,
234        timeout: Duration,
235    ) -> Result<Option<MatchEvent>, ClientError> {
236        if let Some(ev) = self.pending_events.pop_front() {
237            return Ok(Some(ev));
238        }
239        match tokio::time::timeout(timeout, read_frame(&mut self.stream)).await {
240            Err(_) => Ok(None),
241            Ok(Err(e)) => Err(ClientError::Io(e)),
242            Ok(Ok(Frame::MatchEvent(ev))) => Ok(Some(ev)),
243            Ok(Ok(Frame::Error(e))) => Err(ClientError::Server(e)),
244            Ok(Ok(_)) => Err(ClientError::UnexpectedFrame {
245                got: "non-event en await_event",
246            }),
247        }
248    }
249
250    /// Pide al servidor el listado de sesiones activas. Pensado para
251    /// observadores (broker-explorer, CLIs de diagnóstico). Como
252    /// `ping`, los `MatchEvent` que lleguen intercalados se bufean
253    /// en `pending_events` y no rompen la respuesta.
254    pub async fn list_sessions(&mut self) -> Result<crate::messages::SessionList, ClientError> {
255        write_frame(
256            &mut self.stream,
257            &Frame::ListSessions(crate::messages::ListSessions {
258                session: self.session,
259            }),
260        )
261        .await?;
262        loop {
263            match read_frame(&mut self.stream).await? {
264                Frame::SessionList(list) => return Ok(list),
265                Frame::MatchEvent(ev) => self.pending_events.push_back(ev),
266                Frame::Error(e) => return Err(ClientError::Server(e)),
267                _ => {
268                    return Err(ClientError::UnexpectedFrame {
269                        got: "non-session-list",
270                    });
271                }
272            }
273        }
274    }
275
276    /// Pide al servidor el listado de matches actuales del broker
277    /// (consumer↔producer pares con tipo y estrategia). Mismo patrón
278    /// de drenado de `MatchEvent`s intermedios.
279    pub async fn list_matches(&mut self) -> Result<crate::messages::MatchList, ClientError> {
280        write_frame(
281            &mut self.stream,
282            &Frame::ListMatches(crate::messages::ListMatches {
283                session: self.session,
284            }),
285        )
286        .await?;
287        loop {
288            match read_frame(&mut self.stream).await? {
289                Frame::MatchList(list) => return Ok(list),
290                Frame::MatchEvent(ev) => self.pending_events.push_back(ev),
291                Frame::Error(e) => return Err(ClientError::Server(e)),
292                _ => {
293                    return Err(ClientError::UnexpectedFrame {
294                        got: "non-match-list",
295                    });
296                }
297            }
298        }
299    }
300
301    /// Cierre cooperativo. Consume el cliente.
302    pub async fn farewell(mut self) -> Result<(), ClientError> {
303        write_frame(
304            &mut self.stream,
305            &Frame::Farewell(Farewell {
306                session: self.session,
307            }),
308        )
309        .await?;
310        Ok(())
311    }
312}