fips_core/transport/
websocket.rs

1//! Binary WebSocket physical transport.
2//!
3//! One binary WebSocket message carries one FIPS physical record. A tiny
4//! nonce-bound key-hint exchange precedes Noise IK when a client has only a
5//! seed URL; the hint is untrusted routing metadata and never bypasses FIPS
6//! identity authentication or ACLs.
7
8use super::tcp::stream::validate_stream_record;
9use super::{
10    ConnectionState, DiscoveredPeer, PacketBuffer, PacketTx, ReceivedPacket, Transport,
11    TransportAddr, TransportError, TransportId, TransportState, TransportType,
12};
13use crate::Identity;
14use crate::config::WebSocketConfig;
15use crate::discovery::local_udp::LocalKeyHint;
16use futures::{SinkExt, StreamExt};
17use rand::RngExt;
18use secp256k1::XOnlyPublicKey;
19use serde::Serialize;
20use std::collections::{HashMap, VecDeque};
21use std::net::SocketAddr;
22use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
23use std::sync::{Arc, Mutex as StdMutex};
24use std::time::{Duration, SystemTime, UNIX_EPOCH};
25use tokio::net::{TcpListener, TcpStream};
26use tokio::sync::{Mutex, Semaphore, mpsc};
27use tokio::task::JoinHandle;
28use tokio_tungstenite::tungstenite::handshake::server::{ErrorResponse, Request, Response};
29use tokio_tungstenite::tungstenite::protocol::WebSocketConfig as TungsteniteConfig;
30use tokio_tungstenite::tungstenite::{Bytes, Message};
31use tokio_tungstenite::{WebSocketStream, accept_hdr_async_with_config, connect_async_with_config};
32use tracing::{debug, info};
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35enum Direction {
36    Inbound,
37    Outbound,
38}
39
40struct Connection {
41    generation: u64,
42    tx: mpsc::Sender<Vec<u8>>,
43}
44
45type ConnectionPool = Arc<Mutex<HashMap<TransportAddr, Connection>>>;
46type ConnectionStates = Arc<StdMutex<HashMap<TransportAddr, ConnectionState>>>;
47type DiscoveryQueue = Arc<StdMutex<VecDeque<DiscoveredPeer>>>;
48
49#[derive(Debug, Default)]
50struct WebSocketStats {
51    connections_opened: AtomicU64,
52    connections_closed: AtomicU64,
53    connections_rejected: AtomicU64,
54    reconnect_attempts: AtomicU64,
55    frames_sent: AtomicU64,
56    frames_received: AtomicU64,
57    bytes_sent: AtomicU64,
58    bytes_received: AtomicU64,
59    invalid_frames: AtomicU64,
60    send_queue_full: AtomicU64,
61}
62
63#[derive(Debug, Serialize)]
64pub struct WebSocketStatsSnapshot {
65    pub connections_opened: u64,
66    pub connections_closed: u64,
67    pub connections_rejected: u64,
68    pub reconnect_attempts: u64,
69    pub frames_sent: u64,
70    pub frames_received: u64,
71    pub bytes_sent: u64,
72    pub bytes_received: u64,
73    pub invalid_frames: u64,
74    pub send_queue_full: u64,
75}
76
77impl WebSocketStats {
78    fn snapshot(&self) -> WebSocketStatsSnapshot {
79        let load = |value: &AtomicU64| value.load(Ordering::Relaxed);
80        WebSocketStatsSnapshot {
81            connections_opened: load(&self.connections_opened),
82            connections_closed: load(&self.connections_closed),
83            connections_rejected: load(&self.connections_rejected),
84            reconnect_attempts: load(&self.reconnect_attempts),
85            frames_sent: load(&self.frames_sent),
86            frames_received: load(&self.frames_received),
87            bytes_sent: load(&self.bytes_sent),
88            bytes_received: load(&self.bytes_received),
89            invalid_frames: load(&self.invalid_frames),
90            send_queue_full: load(&self.send_queue_full),
91        }
92    }
93}
94
95#[derive(Clone)]
96struct Runtime {
97    transport_id: TransportId,
98    config: WebSocketConfig,
99    local_pubkey: [u8; 32],
100    packet_tx: PacketTx,
101    pool: ConnectionPool,
102    states: ConnectionStates,
103    discoveries: DiscoveryQueue,
104    running: Arc<AtomicBool>,
105    total_slots: Arc<Semaphore>,
106    inbound_slots: Arc<Semaphore>,
107    generation: Arc<AtomicU64>,
108    stats: Arc<WebSocketStats>,
109}
110
111impl Runtime {
112    fn websocket_config(&self) -> TungsteniteConfig {
113        let mut config = TungsteniteConfig::default();
114        config.max_message_size = Some(self.config.max_frame_bytes());
115        config.max_frame_size = Some(self.config.max_frame_bytes());
116        config.max_write_buffer_size = self.config.max_frame_bytes().saturating_mul(2);
117        config.write_buffer_size = 0;
118        config
119    }
120
121    fn next_generation(&self) -> u64 {
122        self.generation.fetch_add(1, Ordering::Relaxed)
123    }
124
125    fn set_state(&self, addr: &TransportAddr, state: ConnectionState) {
126        self.states
127            .lock()
128            .unwrap_or_else(|error| error.into_inner())
129            .insert(addr.clone(), state);
130    }
131
132    fn clear_state_if(&self, addr: &TransportAddr, generation: u64) {
133        let connected_generation = self
134            .pool
135            .try_lock()
136            .ok()
137            .and_then(|pool| pool.get(addr).map(|connection| connection.generation));
138        if connected_generation != Some(generation) {
139            self.states
140                .lock()
141                .unwrap_or_else(|error| error.into_inner())
142                .remove(addr);
143        }
144    }
145}
146
147/// Generic WebSocket physical transport.
148pub struct WebSocketTransport {
149    transport_id: TransportId,
150    name: Option<String>,
151    config: WebSocketConfig,
152    state: TransportState,
153    local_addr: Option<SocketAddr>,
154    runtime: Runtime,
155    tasks: Arc<StdMutex<Vec<JoinHandle<()>>>>,
156}
157
158impl WebSocketTransport {
159    pub fn new(
160        transport_id: TransportId,
161        name: Option<String>,
162        config: WebSocketConfig,
163        packet_tx: PacketTx,
164        identity: &Identity,
165    ) -> Self {
166        let max_connections = config.max_connections();
167        let max_inbound = config.max_inbound_connections();
168        let runtime = Runtime {
169            transport_id,
170            config: config.clone(),
171            local_pubkey: identity.pubkey().serialize(),
172            packet_tx,
173            pool: Arc::new(Mutex::new(HashMap::new())),
174            states: Arc::new(StdMutex::new(HashMap::new())),
175            discoveries: Arc::new(StdMutex::new(VecDeque::new())),
176            running: Arc::new(AtomicBool::new(false)),
177            total_slots: Arc::new(Semaphore::new(max_connections)),
178            inbound_slots: Arc::new(Semaphore::new(max_inbound)),
179            generation: Arc::new(AtomicU64::new(1)),
180            stats: Arc::new(WebSocketStats::default()),
181        };
182        Self {
183            transport_id,
184            name,
185            config,
186            state: TransportState::Configured,
187            local_addr: None,
188            runtime,
189            tasks: Arc::new(StdMutex::new(Vec::new())),
190        }
191    }
192
193    pub fn name(&self) -> Option<&str> {
194        self.name.as_deref()
195    }
196
197    pub fn local_addr(&self) -> Option<SocketAddr> {
198        self.local_addr
199    }
200
201    pub fn public_url(&self) -> Option<&str> {
202        self.config.public_url.as_deref()
203    }
204
205    pub fn stats(&self) -> WebSocketStatsSnapshot {
206        self.runtime.stats.snapshot()
207    }
208
209    fn push_task(&self, task: JoinHandle<()>) {
210        self.tasks
211            .lock()
212            .unwrap_or_else(|error| error.into_inner())
213            .push(task);
214    }
215
216    pub async fn start_async(&mut self) -> Result<(), TransportError> {
217        if !self.state.can_start() {
218            return Err(TransportError::AlreadyStarted);
219        }
220        self.config
221            .validate()
222            .map_err(TransportError::StartFailed)?;
223        self.state = TransportState::Starting;
224        self.runtime.running.store(true, Ordering::Release);
225
226        if let Some(bind_addr) = self.config.bind_addr.as_deref() {
227            let bind_addr = bind_addr
228                .parse::<SocketAddr>()
229                .map_err(|error| TransportError::StartFailed(error.to_string()))?;
230            let listener = TcpListener::bind(bind_addr)
231                .await
232                .map_err(|error| TransportError::bind_failed(bind_addr, error))?;
233            self.local_addr = Some(
234                listener
235                    .local_addr()
236                    .map_err(|error| TransportError::StartFailed(error.to_string()))?,
237            );
238            self.push_task(tokio::spawn(run_accept_loop(
239                self.runtime.clone(),
240                listener,
241            )));
242        }
243
244        for seed_url in self.config.seed_urls.clone() {
245            let addr = TransportAddr::from_string(&seed_url);
246            self.runtime.set_state(&addr, ConnectionState::Connecting);
247            self.push_task(tokio::spawn(run_seed_dialer(self.runtime.clone(), addr)));
248        }
249
250        self.state = TransportState::Up;
251        info!(
252            transport_id = %self.transport_id,
253            local_addr = ?self.local_addr,
254            seeds = self.config.seed_urls.len(),
255            "WebSocket transport started"
256        );
257        Ok(())
258    }
259
260    pub async fn stop_async(&mut self) -> Result<(), TransportError> {
261        if !self.state.is_operational() {
262            return Err(TransportError::NotStarted);
263        }
264        self.runtime.running.store(false, Ordering::Release);
265        let tasks = {
266            let mut tasks = self.tasks.lock().unwrap_or_else(|error| error.into_inner());
267            std::mem::take(&mut *tasks)
268        };
269        for task in &tasks {
270            task.abort();
271        }
272        for task in tasks {
273            let _ = task.await;
274        }
275        self.runtime.pool.lock().await.clear();
276        self.runtime
277            .states
278            .lock()
279            .unwrap_or_else(|error| error.into_inner())
280            .clear();
281        self.local_addr = None;
282        self.state = TransportState::Down;
283        Ok(())
284    }
285
286    pub async fn send_async(
287        &self,
288        addr: &TransportAddr,
289        data: &[u8],
290    ) -> Result<usize, TransportError> {
291        if !self.state.is_operational() {
292            return Err(TransportError::NotStarted);
293        }
294        if data.len() > self.config.max_frame_bytes() {
295            return Err(TransportError::MtuExceeded {
296                packet_size: data.len(),
297                mtu: self.config.max_frame_bytes().min(u16::MAX as usize) as u16,
298            });
299        }
300        validate_stream_record(data).map_err(|error| {
301            TransportError::SendFailed(format!("invalid FIPS physical record: {error}"))
302        })?;
303        let tx = self
304            .runtime
305            .pool
306            .lock()
307            .await
308            .get(addr)
309            .map(|connection| connection.tx.clone())
310            .ok_or(TransportError::NotStarted)?;
311        tx.try_send(data.to_vec()).map_err(|error| match error {
312            mpsc::error::TrySendError::Full(_) => {
313                self.runtime
314                    .stats
315                    .send_queue_full
316                    .fetch_add(1, Ordering::Relaxed);
317                TransportError::SendFailed("WebSocket send queue full".into())
318            }
319            mpsc::error::TrySendError::Closed(_) => {
320                TransportError::SendFailed("WebSocket connection closed".into())
321            }
322        })?;
323        Ok(data.len())
324    }
325
326    pub async fn connect_async(&self, addr: &TransportAddr) -> Result<(), TransportError> {
327        if !self.state.is_operational() {
328            return Err(TransportError::NotStarted);
329        }
330        let raw = addr
331            .as_str()
332            .ok_or_else(|| TransportError::InvalidAddress(addr.to_string()))?;
333        let candidate = WebSocketConfig {
334            seed_urls: vec![raw.to_owned()],
335            ..self.config.clone()
336        };
337        candidate
338            .validate()
339            .map_err(TransportError::InvalidAddress)?;
340        if self.connection_state_sync(addr) == ConnectionState::Connected {
341            return Ok(());
342        }
343        {
344            let mut states = self
345                .runtime
346                .states
347                .lock()
348                .unwrap_or_else(|error| error.into_inner());
349            if matches!(states.get(addr), Some(ConnectionState::Connecting)) {
350                return Ok(());
351            }
352            states.insert(addr.clone(), ConnectionState::Connecting);
353        }
354        self.push_task(tokio::spawn(run_one_shot_dial(
355            self.runtime.clone(),
356            addr.clone(),
357        )));
358        Ok(())
359    }
360
361    pub fn connection_state_sync(&self, addr: &TransportAddr) -> ConnectionState {
362        if let Ok(pool) = self.runtime.pool.try_lock()
363            && pool.contains_key(addr)
364        {
365            return ConnectionState::Connected;
366        }
367        self.runtime
368            .states
369            .lock()
370            .unwrap_or_else(|error| error.into_inner())
371            .get(addr)
372            .cloned()
373            .unwrap_or(ConnectionState::None)
374    }
375
376    pub async fn close_connection_async(&self, addr: &TransportAddr) {
377        self.runtime.pool.lock().await.remove(addr);
378        self.runtime
379            .states
380            .lock()
381            .unwrap_or_else(|error| error.into_inner())
382            .remove(addr);
383    }
384}
385
386impl Transport for WebSocketTransport {
387    fn transport_id(&self) -> TransportId {
388        self.transport_id
389    }
390
391    fn transport_type(&self) -> &TransportType {
392        &TransportType::WEBSOCKET
393    }
394
395    fn state(&self) -> TransportState {
396        self.state
397    }
398
399    fn mtu(&self) -> u16 {
400        self.config.mtu()
401    }
402
403    fn start(&mut self) -> Result<(), TransportError> {
404        Err(TransportError::NotSupported(
405            "use start_async() for WebSocket transport".into(),
406        ))
407    }
408
409    fn stop(&mut self) -> Result<(), TransportError> {
410        Err(TransportError::NotSupported(
411            "use stop_async() for WebSocket transport".into(),
412        ))
413    }
414
415    fn send(&self, _addr: &TransportAddr, _data: &[u8]) -> Result<(), TransportError> {
416        Err(TransportError::NotSupported(
417            "use send_async() for WebSocket transport".into(),
418        ))
419    }
420
421    fn discover(&self) -> Result<Vec<DiscoveredPeer>, TransportError> {
422        Ok(self
423            .runtime
424            .discoveries
425            .lock()
426            .unwrap_or_else(|error| error.into_inner())
427            .drain(..)
428            .collect())
429    }
430
431    fn auto_connect(&self) -> bool {
432        true
433    }
434
435    fn accept_connections(&self) -> bool {
436        self.config.accept_connections()
437    }
438}
439
440async fn run_accept_loop(runtime: Runtime, listener: TcpListener) {
441    while runtime.running.load(Ordering::Acquire) {
442        let Ok((stream, peer_addr)) = listener.accept().await else {
443            continue;
444        };
445        let Ok(inbound_permit) = runtime.inbound_slots.clone().try_acquire_owned() else {
446            runtime
447                .stats
448                .connections_rejected
449                .fetch_add(1, Ordering::Relaxed);
450            continue;
451        };
452        let Ok(total_permit) = runtime.total_slots.clone().try_acquire_owned() else {
453            runtime
454                .stats
455                .connections_rejected
456                .fetch_add(1, Ordering::Relaxed);
457            continue;
458        };
459        let accepted_runtime = runtime.clone();
460        tokio::spawn(async move {
461            let _inbound_permit = inbound_permit;
462            let _total_permit = total_permit;
463            if let Err(error) = accept_connection(accepted_runtime, stream, peer_addr).await {
464                debug!(%peer_addr, %error, "WebSocket inbound connection ended");
465            }
466        });
467    }
468}
469
470#[allow(clippy::result_large_err)]
471async fn accept_connection(
472    runtime: Runtime,
473    stream: TcpStream,
474    peer_addr: SocketAddr,
475) -> Result<(), TransportError> {
476    let path = runtime.config.path().to_owned();
477    let callback = move |request: &Request, response: Response| {
478        if request.uri().path() == path {
479            Ok(response)
480        } else {
481            let mut error = ErrorResponse::new(Some("not found".into()));
482            *error.status_mut() = tokio_tungstenite::tungstenite::http::StatusCode::NOT_FOUND;
483            Err(error)
484        }
485    };
486    let websocket =
487        accept_hdr_async_with_config(stream, callback, Some(runtime.websocket_config()))
488            .await
489            .map_err(|_| TransportError::ConnectionRefused)?;
490    let generation = runtime.next_generation();
491    let addr = TransportAddr::from_string(&format!("ws-peer://{peer_addr}/{generation}"));
492    run_connection(
493        runtime,
494        addr,
495        websocket,
496        generation,
497        Direction::Inbound,
498        false,
499    )
500    .await
501}
502
503async fn run_seed_dialer(runtime: Runtime, addr: TransportAddr) {
504    let mut delay_ms = runtime.config.reconnect_initial_ms();
505    while runtime.running.load(Ordering::Acquire) {
506        runtime.set_state(&addr, ConnectionState::Connecting);
507        runtime
508            .stats
509            .reconnect_attempts
510            .fetch_add(1, Ordering::Relaxed);
511        let result = dial_and_run(runtime.clone(), addr.clone()).await;
512        if !runtime.running.load(Ordering::Acquire) {
513            break;
514        }
515        match result {
516            Ok(()) => {
517                delay_ms = runtime.config.reconnect_initial_ms();
518            }
519            Err(error) => {
520                runtime.set_state(&addr, ConnectionState::Failed(error.to_string()));
521                debug!(remote_addr = %addr, %error, "WebSocket seed connection failed");
522                delay_ms = delay_ms
523                    .saturating_mul(2)
524                    .min(runtime.config.reconnect_max_ms());
525            }
526        }
527        tokio::time::sleep(Duration::from_millis(delay_ms)).await;
528    }
529}
530
531async fn run_one_shot_dial(runtime: Runtime, addr: TransportAddr) {
532    let result = dial_and_run(runtime.clone(), addr.clone()).await;
533    if let Err(error) = result {
534        runtime.set_state(&addr, ConnectionState::Failed(error.to_string()));
535    }
536}
537
538async fn dial_and_run(runtime: Runtime, addr: TransportAddr) -> Result<(), TransportError> {
539    let _slot = runtime
540        .total_slots
541        .clone()
542        .try_acquire_owned()
543        .map_err(|_| TransportError::ConnectionRefused)?;
544    let url = addr
545        .as_str()
546        .ok_or_else(|| TransportError::InvalidAddress(addr.to_string()))?;
547    let connect = connect_async_with_config(url, Some(runtime.websocket_config()), false);
548    let (websocket, _) = tokio::time::timeout(
549        Duration::from_millis(runtime.config.connect_timeout_ms()),
550        connect,
551    )
552    .await
553    .map_err(|_| TransportError::Timeout)?
554    .map_err(|error| TransportError::StartFailed(error.to_string()))?;
555    let generation = runtime.next_generation();
556    run_connection(
557        runtime,
558        addr,
559        websocket,
560        generation,
561        Direction::Outbound,
562        true,
563    )
564    .await
565}
566
567async fn run_connection<S>(
568    runtime: Runtime,
569    addr: TransportAddr,
570    websocket: WebSocketStream<S>,
571    generation: u64,
572    direction: Direction,
573    request_key_hint: bool,
574) -> Result<(), TransportError>
575where
576    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
577{
578    let (mut sink, mut stream) = websocket.split();
579    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(runtime.config.max_send_queue());
580    {
581        let mut pool = runtime.pool.lock().await;
582        if pool.contains_key(&addr) {
583            return Err(TransportError::AlreadyStarted);
584        }
585        pool.insert(addr.clone(), Connection { generation, tx });
586    }
587    runtime.set_state(&addr, ConnectionState::Connected);
588    runtime
589        .stats
590        .connections_opened
591        .fetch_add(1, Ordering::Relaxed);
592
593    let mut pending_nonce = request_key_hint.then(|| rand::rng().random::<u64>());
594    if let Some(nonce) = pending_nonce {
595        sink.send(Message::Binary(
596            LocalKeyHint::Request { nonce }.encode().into(),
597        ))
598        .await
599        .map_err(|error| TransportError::SendFailed(error.to_string()))?;
600    }
601
602    let started = tokio::time::Instant::now();
603    let mut last_received = started;
604    let ping_secs = runtime.config.ping_interval_secs();
605    let idle_secs = runtime.config.idle_timeout_secs();
606    let mut ping = tokio::time::interval(if ping_secs == 0 {
607        Duration::from_secs(24 * 60 * 60)
608    } else {
609        Duration::from_secs(ping_secs)
610    });
611    ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
612    let mut check = tokio::time::interval(Duration::from_secs(1));
613    check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
614
615    enum Event {
616        Outbound(Option<Vec<u8>>),
617        Inbound(Option<Result<Message, tokio_tungstenite::tungstenite::Error>>),
618        Ping,
619        Check,
620    }
621
622    let outcome = loop {
623        let event = tokio::select! {
624            outbound = rx.recv() => Event::Outbound(outbound),
625            inbound = stream.next() => Event::Inbound(inbound),
626            _ = ping.tick(), if ping_secs > 0 => Event::Ping,
627            _ = check.tick() => Event::Check,
628        };
629        match event {
630            Event::Outbound(Some(data)) => {
631                let len = data.len();
632                if let Err(error) = sink.send(Message::Binary(data.into())).await {
633                    break Err(TransportError::SendFailed(error.to_string()));
634                }
635                runtime.stats.frames_sent.fetch_add(1, Ordering::Relaxed);
636                runtime
637                    .stats
638                    .bytes_sent
639                    .fetch_add(len as u64, Ordering::Relaxed);
640            }
641            Event::Outbound(None) => break Ok(()),
642            Event::Inbound(Some(Ok(Message::Binary(data)))) => {
643                last_received = tokio::time::Instant::now();
644                if let Some(hint) = LocalKeyHint::decode(&data) {
645                    match hint {
646                        LocalKeyHint::Request { nonce } => {
647                            let reply = LocalKeyHint::Response {
648                                nonce,
649                                pubkey: runtime.local_pubkey,
650                            };
651                            if let Err(error) =
652                                sink.send(Message::Binary(reply.encode().into())).await
653                            {
654                                break Err(TransportError::SendFailed(error.to_string()));
655                            }
656                        }
657                        LocalKeyHint::Response { nonce, pubkey }
658                            if pending_nonce == Some(nonce) =>
659                        {
660                            pending_nonce = None;
661                            if pubkey != runtime.local_pubkey
662                                && let Ok(pubkey) = XOnlyPublicKey::from_slice(&pubkey)
663                            {
664                                runtime
665                                    .discoveries
666                                    .lock()
667                                    .unwrap_or_else(|error| error.into_inner())
668                                    .push_back(DiscoveredPeer::with_hint(
669                                        runtime.transport_id,
670                                        addr.clone(),
671                                        pubkey,
672                                    ));
673                            }
674                        }
675                        LocalKeyHint::Response { .. } => {}
676                    }
677                    continue;
678                }
679                if data.len() > runtime.config.max_frame_bytes()
680                    || validate_stream_record(&data).is_err()
681                {
682                    runtime.stats.invalid_frames.fetch_add(1, Ordering::Relaxed);
683                    break Err(TransportError::RecvFailed(
684                        "invalid WebSocket FIPS physical record".into(),
685                    ));
686                }
687                let len = data.len();
688                let packet = ReceivedPacket::with_timestamp(
689                    runtime.transport_id,
690                    addr.clone(),
691                    PacketBuffer::new(data.to_vec()),
692                    now_ms(),
693                );
694                if runtime.packet_tx.send(packet).is_err() {
695                    break Err(TransportError::RecvFailed(
696                        "node packet channel closed".into(),
697                    ));
698                }
699                runtime
700                    .stats
701                    .frames_received
702                    .fetch_add(1, Ordering::Relaxed);
703                runtime
704                    .stats
705                    .bytes_received
706                    .fetch_add(len as u64, Ordering::Relaxed);
707            }
708            Event::Inbound(Some(Ok(Message::Ping(payload)))) => {
709                last_received = tokio::time::Instant::now();
710                if let Err(error) = sink.send(Message::Pong(payload)).await {
711                    break Err(TransportError::SendFailed(error.to_string()));
712                }
713            }
714            Event::Inbound(Some(Ok(Message::Pong(_)))) => {
715                last_received = tokio::time::Instant::now();
716            }
717            Event::Inbound(Some(Ok(Message::Close(_))) | None) => break Ok(()),
718            Event::Inbound(Some(Ok(Message::Text(_) | Message::Frame(_)))) => {
719                runtime.stats.invalid_frames.fetch_add(1, Ordering::Relaxed);
720                break Err(TransportError::RecvFailed(
721                    "WebSocket transport accepts binary messages only".into(),
722                ));
723            }
724            Event::Inbound(Some(Err(error))) => {
725                break Err(TransportError::RecvFailed(error.to_string()));
726            }
727            Event::Ping => {
728                if let Err(error) = sink.send(Message::Ping(Bytes::new())).await {
729                    break Err(TransportError::SendFailed(error.to_string()));
730                }
731            }
732            Event::Check => {
733                if pending_nonce.is_some()
734                    && started.elapsed()
735                        >= Duration::from_millis(runtime.config.key_hint_timeout_ms())
736                {
737                    break Err(TransportError::Timeout);
738                }
739                if idle_secs > 0 && last_received.elapsed() >= Duration::from_secs(idle_secs) {
740                    break Err(TransportError::Timeout);
741                }
742            }
743        }
744    };
745
746    {
747        let mut pool = runtime.pool.lock().await;
748        if pool
749            .get(&addr)
750            .is_some_and(|connection| connection.generation == generation)
751        {
752            pool.remove(&addr);
753        }
754    }
755    runtime.clear_state_if(&addr, generation);
756    runtime
757        .stats
758        .connections_closed
759        .fetch_add(1, Ordering::Relaxed);
760    debug!(
761        transport_id = %runtime.transport_id,
762        remote_addr = %addr,
763        ?direction,
764        "WebSocket physical connection closed"
765    );
766    outcome
767}
768
769fn now_ms() -> u64 {
770    SystemTime::now()
771        .duration_since(UNIX_EPOCH)
772        .unwrap_or_default()
773        .as_millis() as u64
774}
775
776#[cfg(test)]
777mod tests;
fips_core/transport/websocket.rs

fips_core/transport/
websocket.rs
