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

fips_core/transport/
websocket.rs
