Line data Source code
1 : use crate::{
2 : cancellation::CancellationHandlerMain,
3 : config::ProxyConfig,
4 : context::RequestMonitoring,
5 : error::{io_error, ReportableError},
6 : metrics::Metrics,
7 : proxy::{handle_client, ClientMode},
8 : rate_limiter::EndpointRateLimiter,
9 : };
10 : use bytes::{Buf, BufMut, Bytes, BytesMut};
11 : use framed_websockets::{Frame, OpCode, WebSocketServer};
12 : use futures::{Sink, Stream};
13 : use hyper1::upgrade::OnUpgrade;
14 : use hyper_util::rt::TokioIo;
15 : use pin_project_lite::pin_project;
16 :
17 : use std::{
18 : pin::Pin,
19 : sync::Arc,
20 : task::{ready, Context, Poll},
21 : };
22 : use tokio::io::{self, AsyncBufRead, AsyncRead, AsyncWrite, ReadBuf};
23 : use tracing::warn;
24 :
25 : pin_project! {
26 : /// This is a wrapper around a [`WebSocketStream`] that
27 : /// implements [`AsyncRead`] and [`AsyncWrite`].
28 : pub struct WebSocketRw<S> {
29 : #[pin]
30 : stream: WebSocketServer<S>,
31 : recv: Bytes,
32 : send: BytesMut,
33 : }
34 : }
35 :
36 : impl<S> WebSocketRw<S> {
37 2 : pub fn new(stream: WebSocketServer<S>) -> Self {
38 2 : Self {
39 2 : stream,
40 2 : recv: Bytes::new(),
41 2 : send: BytesMut::new(),
42 2 : }
43 2 : }
44 : }
45 :
46 : impl<S: AsyncRead + AsyncWrite + Unpin> AsyncWrite for WebSocketRw<S> {
47 2 : fn poll_write(
48 2 : self: Pin<&mut Self>,
49 2 : cx: &mut Context<'_>,
50 2 : buf: &[u8],
51 2 : ) -> Poll<io::Result<usize>> {
52 2 : let this = self.project();
53 2 : let mut stream = this.stream;
54 2 : this.send.put(buf);
55 :
56 2 : ready!(stream.as_mut().poll_ready(cx).map_err(io_error))?;
57 2 : match stream.as_mut().start_send(Frame::binary(this.send.split())) {
58 2 : Ok(()) => Poll::Ready(Ok(buf.len())),
59 0 : Err(e) => Poll::Ready(Err(io_error(e))),
60 : }
61 2 : }
62 :
63 2 : fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
64 2 : let stream = self.project().stream;
65 2 : stream.poll_flush(cx).map_err(io_error)
66 2 : }
67 :
68 0 : fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
69 0 : let stream = self.project().stream;
70 0 : stream.poll_close(cx).map_err(io_error)
71 0 : }
72 : }
73 :
74 : impl<S: AsyncRead + AsyncWrite + Unpin> AsyncRead for WebSocketRw<S> {
75 6 : fn poll_read(
76 6 : mut self: Pin<&mut Self>,
77 6 : cx: &mut Context<'_>,
78 6 : buf: &mut ReadBuf<'_>,
79 6 : ) -> Poll<io::Result<()>> {
80 6 : let bytes = ready!(self.as_mut().poll_fill_buf(cx))?;
81 4 : let len = std::cmp::min(bytes.len(), buf.remaining());
82 4 : buf.put_slice(&bytes[..len]);
83 4 : self.consume(len);
84 4 : Poll::Ready(Ok(()))
85 6 : }
86 : }
87 :
88 : impl<S: AsyncRead + AsyncWrite + Unpin> AsyncBufRead for WebSocketRw<S> {
89 6 : fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
90 6 : // Please refer to poll_fill_buf's documentation.
91 6 : const EOF: Poll<io::Result<&[u8]>> = Poll::Ready(Ok(&[]));
92 6 :
93 6 : let mut this = self.project();
94 8 : loop {
95 8 : if !this.recv.chunk().is_empty() {
96 2 : let chunk = (*this.recv).chunk();
97 2 : return Poll::Ready(Ok(chunk));
98 6 : }
99 :
100 6 : let res = ready!(this.stream.as_mut().poll_next(cx));
101 4 : match res.transpose().map_err(io_error)? {
102 4 : Some(message) => match message.opcode {
103 0 : OpCode::Ping => {}
104 0 : OpCode::Pong => {}
105 : OpCode::Text => {
106 : // We expect to see only binary messages.
107 0 : let error = "unexpected text message in the websocket";
108 0 : warn!(length = message.payload.len(), error);
109 0 : return Poll::Ready(Err(io_error(error)));
110 : }
111 : OpCode::Binary | OpCode::Continuation => {
112 2 : debug_assert!(this.recv.is_empty());
113 2 : *this.recv = message.payload.freeze();
114 : }
115 2 : OpCode::Close => return EOF,
116 : },
117 0 : None => return EOF,
118 : }
119 : }
120 6 : }
121 :
122 4 : fn consume(self: Pin<&mut Self>, amount: usize) {
123 4 : self.project().recv.advance(amount);
124 4 : }
125 : }
126 :
127 0 : pub async fn serve_websocket(
128 0 : config: &'static ProxyConfig,
129 0 : mut ctx: RequestMonitoring,
130 0 : websocket: OnUpgrade,
131 0 : cancellation_handler: Arc<CancellationHandlerMain>,
132 0 : endpoint_rate_limiter: Arc<EndpointRateLimiter>,
133 0 : hostname: Option<String>,
134 0 : ) -> anyhow::Result<()> {
135 0 : let websocket = websocket.await?;
136 0 : let websocket = WebSocketServer::after_handshake(TokioIo::new(websocket));
137 0 :
138 0 : let conn_gauge = Metrics::get()
139 0 : .proxy
140 0 : .client_connections
141 0 : .guard(crate::metrics::Protocol::Ws);
142 :
143 0 : let res = handle_client(
144 0 : config,
145 0 : &mut ctx,
146 0 : cancellation_handler,
147 0 : WebSocketRw::new(websocket),
148 0 : ClientMode::Websockets { hostname },
149 0 : endpoint_rate_limiter,
150 0 : conn_gauge,
151 0 : )
152 0 : .await;
153 :
154 0 : match res {
155 0 : Err(e) => {
156 0 : // todo: log and push to ctx the error kind
157 0 : ctx.set_error_kind(e.get_error_kind());
158 0 : Err(e.into())
159 : }
160 : Ok(None) => {
161 0 : ctx.set_success();
162 0 : Ok(())
163 : }
164 0 : Ok(Some(p)) => {
165 0 : ctx.set_success();
166 0 : ctx.log_connect();
167 0 : p.proxy_pass().await
168 : }
169 : }
170 0 : }
171 :
172 : #[cfg(test)]
173 : mod tests {
174 : use std::pin::pin;
175 :
176 : use framed_websockets::WebSocketServer;
177 : use futures::{SinkExt, StreamExt};
178 : use tokio::{
179 : io::{duplex, AsyncReadExt, AsyncWriteExt},
180 : task::JoinSet,
181 : };
182 : use tokio_tungstenite::{
183 : tungstenite::{protocol::Role, Message},
184 : WebSocketStream,
185 : };
186 :
187 : use super::WebSocketRw;
188 :
189 : #[tokio::test]
190 2 : async fn websocket_stream_wrapper_happy_path() {
191 2 : let (stream1, stream2) = duplex(1024);
192 2 :
193 2 : let mut js = JoinSet::new();
194 2 :
195 2 : js.spawn(async move {
196 2 : let mut client = WebSocketStream::from_raw_socket(stream1, Role::Client, None).await;
197 2 :
198 2 : client
199 2 : .send(Message::Binary(b"hello world".to_vec()))
200 2 : .await
201 2 : .unwrap();
202 2 :
203 2 : let message = client.next().await.unwrap().unwrap();
204 2 : assert_eq!(message, Message::Binary(b"websockets are cool".to_vec()));
205 2 :
206 2 : client.close(None).await.unwrap();
207 2 : });
208 2 :
209 2 : js.spawn(async move {
210 2 : let mut rw = pin!(WebSocketRw::new(WebSocketServer::after_handshake(stream2)));
211 2 :
212 2 : let mut buf = vec![0; 1024];
213 2 : let n = rw.read(&mut buf).await.unwrap();
214 2 : assert_eq!(&buf[..n], b"hello world");
215 2 :
216 2 : rw.write_all(b"websockets are cool").await.unwrap();
217 2 : rw.flush().await.unwrap();
218 2 :
219 2 : let n = rw.read_to_end(&mut buf).await.unwrap();
220 2 : assert_eq!(n, 0);
221 2 : });
222 2 :
223 2 : js.join_next().await.unwrap().unwrap();
224 2 : js.join_next().await.unwrap().unwrap();
225 2 : }
226 : }
|
