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        };
176        Ok(Self {
177            stream,
178            session: ack.session,
179            server_info: ack,
180            pending_events: VecDeque::new(),
181        })
182    }
183
184    /// `SessionId` asignado por el servidor.
185    pub fn session(&self) -> SessionId {
186        self.session
187    }
188
189    /// Información del servidor recibida en el handshake.
190    pub fn server_info(&self) -> &HelloAck {
191        &self.server_info
192    }
193
194    /// Envía un Ping y devuelve el timestamp del servidor. Los frames
195    /// `MatchEvent` que lleguen mezclados se buferean en `pending_events`.
196    pub async fn ping(&mut self) -> Result<u64, ClientError> {
197        write_frame(
198            &mut self.stream,
199            &Frame::Ping(Ping {
200                session: self.session,
201            }),
202        )
203        .await?;
204        loop {
205            match read_frame(&mut self.stream).await? {
206                Frame::Pong(p) => return Ok(p.timestamp_ms),
207                Frame::MatchEvent(ev) => self.pending_events.push_back(ev),
208                Frame::Error(e) => return Err(ClientError::Server(e)),
209                _ => return Err(ClientError::UnexpectedFrame { got: "non-pong" }),
210            }
211        }
212    }
213
214    /// Saca un evento pendiente del buffer, sin bloquear ni leer del wire.
215    pub fn take_event(&mut self) -> Option<MatchEvent> {
216        self.pending_events.pop_front()
217    }
218
219    /// Espera un `MatchEvent` con timeout. Drena primero el buffer; si
220    /// está vacío, lee del wire hasta el timeout. Otros frames recibidos
221    /// (Pong huérfano, Error) cortan la espera con error.
222    pub async fn await_event(
223        &mut self,
224        timeout: Duration,
225    ) -> Result<Option<MatchEvent>, ClientError> {
226        if let Some(ev) = self.pending_events.pop_front() {
227            return Ok(Some(ev));
228        }
229        match tokio::time::timeout(timeout, read_frame(&mut self.stream)).await {
230            Err(_) => Ok(None),
231            Ok(Err(e)) => Err(ClientError::Io(e)),
232            Ok(Ok(Frame::MatchEvent(ev))) => Ok(Some(ev)),
233            Ok(Ok(Frame::Error(e))) => Err(ClientError::Server(e)),
234            Ok(Ok(_)) => Err(ClientError::UnexpectedFrame {
235                got: "non-event en await_event",
236            }),
237        }
238    }
239
240    /// Pide al servidor el listado de sesiones activas. Pensado para
241    /// observadores (broker-explorer, CLIs de diagnóstico). Como
242    /// `ping`, los `MatchEvent` que lleguen intercalados se bufean
243    /// en `pending_events` y no rompen la respuesta.
244    pub async fn list_sessions(&mut self) -> Result<crate::messages::SessionList, ClientError> {
245        write_frame(
246            &mut self.stream,
247            &Frame::ListSessions(crate::messages::ListSessions {
248                session: self.session,
249            }),
250        )
251        .await?;
252        loop {
253            match read_frame(&mut self.stream).await? {
254                Frame::SessionList(list) => return Ok(list),
255                Frame::MatchEvent(ev) => self.pending_events.push_back(ev),
256                Frame::Error(e) => return Err(ClientError::Server(e)),
257                _ => {
258                    return Err(ClientError::UnexpectedFrame {
259                        got: "non-session-list",
260                    });
261                }
262            }
263        }
264    }
265
266    /// Cierre cooperativo. Consume el cliente.
267    pub async fn farewell(mut self) -> Result<(), ClientError> {
268        write_frame(
269            &mut self.stream,
270            &Frame::Farewell(Farewell {
271                session: self.session,
272            }),
273        )
274        .await?;
275        Ok(())
276    }
277}