sova-ws 0.1.1

WebSocket plugin for Sova
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100

//! WebSocket plugin tests.

use http::Method;
use sova_core::{App,  Plugin, Request};
use sova_core::extend::Bind;
use sova_ws::{origin_allowed, Message, Ws, WsRouteExt};
use tokio_tungstenite::tungstenite::client::IntoClientRequest;

#[test]
fn origin_allowlist_unit_cases() {
    let mut headers = http::HeaderMap::new();
    headers.insert("origin", "https://app.test".parse().unwrap());
    assert!(origin_allowed(&headers, &["https://app.test".to_string()]));
    assert!(!origin_allowed(&headers, &["https://other.test".to_string()]));
}

#[tokio::test]
async fn origin_rejected_before_upgrade() {
    let mut app = App::new();
    Ws::new()
        .origins(["https://allowed.test"])
        .install(&mut app);
    app.ws("/ws", |_| async {});

    let req = Request::builder()
        .method(Method::GET)
        .path("/ws")
        .header("origin", "https://evil.test")
        .header("upgrade", "websocket")
        .header("connection", "Upgrade")
        .header("sec-websocket-key", "dGhlIHNhbXBsZSBub25jZQ==")
        .header("sec-websocket-version", "13")
        .build();

    let res = app.handle(req).await;
    assert_eq!(res.status_code().as_u16(), 403);
}

#[tokio::test]
async fn missing_upgrade_returns_426() {
    let mut app = App::new();
    Ws::new().install(&mut app);
    app.ws("/ws", |_| async {});

    let req = Request::builder()
        .method(Method::GET)
        .path("/ws")
        .header("upgrade", "websocket")
        .header("connection", "Upgrade")
        .header("sec-websocket-key", "dGhlIHNhbXBsZSBub25jZQ==")
        .header("sec-websocket-version", "13")
        .build();

    let res = app.handle(req).await;
    assert_eq!(res.status_code().as_u16(), 426);
}

#[tokio::test]
async fn websocket_upgrade_smoke() {
    use futures_util::{SinkExt, StreamExt};
    use tokio_tungstenite::connect_async;

    let mut app = App::new();
    Ws::new().install(&mut app);
    app.ws("/ws", |mut ws| async move {
        let _room = ws.join("chat");
        while let Some(Ok(msg)) = ws.recv().await {
            if let Message::Text(text) = msg {
                ws.hub()
                    .broadcast("chat", Message::Text(format!("echo:{text}").into()))
                    .await;
            }
        }
    });

    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    listener.set_nonblocking(true).unwrap();
    let addr = listener.local_addr().unwrap();
    let server = tokio::spawn(async move { app.bind(Bind::Listener(listener)).serve().await });

    tokio::time::sleep(std::time::Duration::from_millis(20)).await;

    let url = format!("ws://{addr}/ws");
    let req = url.into_client_request().unwrap();
    let (mut client, _) = connect_async(req).await.expect("websocket handshake");

    client
        .send(Message::Text("hi".into()))
        .await
        .expect("send");
    let got = tokio::time::timeout(std::time::Duration::from_secs(2), client.next())
        .await
        .expect("timeout")
        .expect("stream")
        .expect("frame");
    assert_eq!(got, Message::Text("echo:hi".into()));

    drop(client);
    server.abort();
}