|
|
|
|
@ -4,18 +4,19 @@ pub mod frame; |
|
|
|
|
|
|
|
|
|
mod message; |
|
|
|
|
|
|
|
|
|
pub use self::message::Message; |
|
|
|
|
pub use self::frame::CloseFrame; |
|
|
|
|
pub use self::message::Message; |
|
|
|
|
|
|
|
|
|
use log::*; |
|
|
|
|
use std::collections::VecDeque; |
|
|
|
|
use std::io::{Read, Write, ErrorKind as IoErrorKind}; |
|
|
|
|
use std::io::{ErrorKind as IoErrorKind, Read, Write}; |
|
|
|
|
use std::mem::replace; |
|
|
|
|
|
|
|
|
|
use error::{Error, Result}; |
|
|
|
|
use self::message::{IncompleteMessage, IncompleteMessageType}; |
|
|
|
|
use self::frame::coding::{CloseCode, Control as OpCtl, Data as OpData, OpCode}; |
|
|
|
|
use self::frame::{Frame, FrameCodec}; |
|
|
|
|
use self::frame::coding::{OpCode, Data as OpData, Control as OpCtl, CloseCode}; |
|
|
|
|
use util::NonBlockingResult; |
|
|
|
|
use self::message::{IncompleteMessage, IncompleteMessageType}; |
|
|
|
|
use crate::error::{Error, Result}; |
|
|
|
|
use crate::util::NonBlockingResult; |
|
|
|
|
|
|
|
|
|
/// Indicates a Client or Server role of the websocket
|
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)] |
|
|
|
|
@ -147,7 +148,6 @@ impl<Stream: Read + Write> WebSocket<Stream> { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/// A context for managing WebSocket stream.
|
|
|
|
|
#[derive(Debug)] |
|
|
|
|
pub struct WebSocketContext { |
|
|
|
|
@ -182,11 +182,7 @@ impl WebSocketContext { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/// Create a WebSocket context that manages an post-handshake stream.
|
|
|
|
|
pub fn from_partially_read( |
|
|
|
|
part: Vec<u8>, |
|
|
|
|
role: Role, |
|
|
|
|
config: Option<WebSocketConfig>, |
|
|
|
|
) -> Self { |
|
|
|
|
pub fn from_partially_read(part: Vec<u8>, role: Role, config: Option<WebSocketConfig>) -> Self { |
|
|
|
|
WebSocketContext { |
|
|
|
|
frame: FrameCodec::from_partially_read(part), |
|
|
|
|
..WebSocketContext::new(role, config) |
|
|
|
|
@ -217,7 +213,7 @@ impl WebSocketContext { |
|
|
|
|
// Thus if read blocks, just let it return WouldBlock.
|
|
|
|
|
if let Some(message) = self.read_message_frame(stream)? { |
|
|
|
|
trace!("Received message {}", message); |
|
|
|
|
return Ok(message) |
|
|
|
|
return Ok(message); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
@ -251,20 +247,14 @@ impl WebSocketContext { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
let frame = match message { |
|
|
|
|
Message::Text(data) => { |
|
|
|
|
Frame::message(data.into(), OpCode::Data(OpData::Text), true) |
|
|
|
|
} |
|
|
|
|
Message::Binary(data) => { |
|
|
|
|
Frame::message(data, OpCode::Data(OpData::Binary), true) |
|
|
|
|
} |
|
|
|
|
Message::Text(data) => Frame::message(data.into(), OpCode::Data(OpData::Text), true), |
|
|
|
|
Message::Binary(data) => Frame::message(data, OpCode::Data(OpData::Binary), true), |
|
|
|
|
Message::Ping(data) => Frame::ping(data), |
|
|
|
|
Message::Pong(data) => { |
|
|
|
|
self.pong = Some(Frame::pong(data)); |
|
|
|
|
return self.write_pending(stream) |
|
|
|
|
} |
|
|
|
|
Message::Close(code) => { |
|
|
|
|
return self.close(stream, code) |
|
|
|
|
return self.write_pending(stream); |
|
|
|
|
} |
|
|
|
|
Message::Close(code) => return self.close(stream, code), |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
self.send_queue.push_back(frame); |
|
|
|
|
@ -342,7 +332,6 @@ impl WebSocketContext { |
|
|
|
|
Stream: Read + Write, |
|
|
|
|
{ |
|
|
|
|
if let Some(mut frame) = self.frame.read_frame(stream, self.config.max_frame_size)? { |
|
|
|
|
|
|
|
|
|
// MUST be 0 unless an extension is negotiated that defines meanings
|
|
|
|
|
// for non-zero values. If a nonzero value is received and none of
|
|
|
|
|
// the negotiated extensions defines the meaning of such a nonzero
|
|
|
|
|
@ -351,7 +340,7 @@ impl WebSocketContext { |
|
|
|
|
{ |
|
|
|
|
let hdr = frame.header(); |
|
|
|
|
if hdr.rsv1 || hdr.rsv2 || hdr.rsv3 { |
|
|
|
|
return Err(Error::Protocol("Reserved bits are non-zero".into())) |
|
|
|
|
return Err(Error::Protocol("Reserved bits are non-zero".into())); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -364,19 +353,22 @@ impl WebSocketContext { |
|
|
|
|
} else { |
|
|
|
|
// The server MUST close the connection upon receiving a
|
|
|
|
|
// frame that is not masked. (RFC 6455)
|
|
|
|
|
return Err(Error::Protocol("Received an unmasked frame from client".into())) |
|
|
|
|
return Err(Error::Protocol( |
|
|
|
|
"Received an unmasked frame from client".into(), |
|
|
|
|
)); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
Role::Client => { |
|
|
|
|
if frame.is_masked() { |
|
|
|
|
// A client MUST close a connection if it detects a masked frame. (RFC 6455)
|
|
|
|
|
return Err(Error::Protocol("Received a masked frame from server".into())) |
|
|
|
|
return Err(Error::Protocol( |
|
|
|
|
"Received a masked frame from server".into(), |
|
|
|
|
)); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
match frame.header().opcode { |
|
|
|
|
|
|
|
|
|
OpCode::Control(ctl) => { |
|
|
|
|
match ctl { |
|
|
|
|
// All control frames MUST have a payload length of 125 bytes or less
|
|
|
|
|
@ -387,12 +379,10 @@ impl WebSocketContext { |
|
|
|
|
_ if frame.payload().len() > 125 => { |
|
|
|
|
Err(Error::Protocol("Control frame too big".into())) |
|
|
|
|
} |
|
|
|
|
OpCtl::Close => { |
|
|
|
|
Ok(self.do_close(frame.into_close()?).map(Message::Close)) |
|
|
|
|
} |
|
|
|
|
OpCtl::Reserved(i) => { |
|
|
|
|
Err(Error::Protocol(format!("Unknown control frame type {}", i).into())) |
|
|
|
|
} |
|
|
|
|
OpCtl::Close => Ok(self.do_close(frame.into_close()?).map(Message::Close)), |
|
|
|
|
OpCtl::Reserved(i) => Err(Error::Protocol( |
|
|
|
|
format!("Unknown control frame type {}", i).into(), |
|
|
|
|
)), |
|
|
|
|
OpCtl::Ping | OpCtl::Pong if !self.state.is_active() => { |
|
|
|
|
// No ping processing while closing.
|
|
|
|
|
Ok(None) |
|
|
|
|
@ -402,9 +392,7 @@ impl WebSocketContext { |
|
|
|
|
self.pong = Some(Frame::pong(data.clone())); |
|
|
|
|
Ok(Some(Message::Ping(data))) |
|
|
|
|
} |
|
|
|
|
OpCtl::Pong => { |
|
|
|
|
Ok(Some(Message::Pong(frame.into_data()))) |
|
|
|
|
} |
|
|
|
|
OpCtl::Pong => Ok(Some(Message::Pong(frame.into_data()))), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -420,7 +408,9 @@ impl WebSocketContext { |
|
|
|
|
if let Some(ref mut msg) = self.incomplete { |
|
|
|
|
msg.extend(frame.into_data(), self.config.max_message_size)?; |
|
|
|
|
} else { |
|
|
|
|
return Err(Error::Protocol("Continue frame but nothing to continue".into())) |
|
|
|
|
return Err(Error::Protocol( |
|
|
|
|
"Continue frame but nothing to continue".into(), |
|
|
|
|
)); |
|
|
|
|
} |
|
|
|
|
if fin { |
|
|
|
|
Ok(Some(self.incomplete.take().unwrap().complete()?)) |
|
|
|
|
@ -428,11 +418,9 @@ impl WebSocketContext { |
|
|
|
|
Ok(None) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
c if self.incomplete.is_some() => { |
|
|
|
|
Err(Error::Protocol( |
|
|
|
|
format!("Received {} while waiting for more fragments", c).into() |
|
|
|
|
)) |
|
|
|
|
} |
|
|
|
|
c if self.incomplete.is_some() => Err(Error::Protocol( |
|
|
|
|
format!("Received {} while waiting for more fragments", c).into(), |
|
|
|
|
)), |
|
|
|
|
OpData::Text | OpData::Binary => { |
|
|
|
|
let msg = { |
|
|
|
|
let message_type = match data { |
|
|
|
|
@ -451,28 +439,27 @@ impl WebSocketContext { |
|
|
|
|
Ok(None) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
OpData::Reserved(i) => { |
|
|
|
|
Err(Error::Protocol(format!("Unknown data frame type {}", i).into())) |
|
|
|
|
} |
|
|
|
|
OpData::Reserved(i) => Err(Error::Protocol( |
|
|
|
|
format!("Unknown data frame type {}", i).into(), |
|
|
|
|
)), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
} // match opcode
|
|
|
|
|
|
|
|
|
|
} else { |
|
|
|
|
// Connection closed by peer
|
|
|
|
|
match replace(&mut self.state, WebSocketState::Terminated) { |
|
|
|
|
WebSocketState::ClosedByPeer | WebSocketState::CloseAcknowledged => { |
|
|
|
|
Err(Error::ConnectionClosed) |
|
|
|
|
} |
|
|
|
|
_ => { |
|
|
|
|
Err(Error::Protocol("Connection reset without closing handshake".into())) |
|
|
|
|
} |
|
|
|
|
_ => Err(Error::Protocol( |
|
|
|
|
"Connection reset without closing handshake".into(), |
|
|
|
|
)), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
/// Received a close frame. Tells if we need to return a close frame to the user.
|
|
|
|
|
#[allow(clippy::option_option)] |
|
|
|
|
fn do_close<'t>(&mut self, close: Option<CloseFrame<'t>>) -> Option<Option<CloseFrame<'t>>> { |
|
|
|
|
debug!("Received close frame: {:?}", close); |
|
|
|
|
match self.state { |
|
|
|
|
@ -488,7 +475,7 @@ impl WebSocketContext { |
|
|
|
|
} else { |
|
|
|
|
Frame::close(Some(CloseFrame { |
|
|
|
|
code: CloseCode::Protocol, |
|
|
|
|
reason: "Protocol violation".into() |
|
|
|
|
reason: "Protocol violation".into(), |
|
|
|
|
})) |
|
|
|
|
} |
|
|
|
|
} else { |
|
|
|
|
@ -518,8 +505,7 @@ impl WebSocketContext { |
|
|
|
|
Stream: Read + Write, |
|
|
|
|
{ |
|
|
|
|
match self.role { |
|
|
|
|
Role::Server => { |
|
|
|
|
} |
|
|
|
|
Role::Server => {} |
|
|
|
|
Role::Client => { |
|
|
|
|
// 5. If the data is being sent by the client, the frame(s) MUST be
|
|
|
|
|
// masked as defined in Section 5.3. (RFC 6455)
|
|
|
|
|
@ -535,7 +521,9 @@ impl WebSocketContext { |
|
|
|
|
match self.state { |
|
|
|
|
WebSocketState::ClosedByPeer | WebSocketState::CloseAcknowledged |
|
|
|
|
if err.kind() == IoErrorKind::ConnectionReset => |
|
|
|
|
Error::ConnectionClosed, |
|
|
|
|
{ |
|
|
|
|
Error::ConnectionClosed |
|
|
|
|
} |
|
|
|
|
_ => Error::Io(err), |
|
|
|
|
} |
|
|
|
|
}), |
|
|
|
|
@ -544,7 +532,6 @@ impl WebSocketContext { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
/// The current connection state.
|
|
|
|
|
#[derive(Debug)] |
|
|
|
|
enum WebSocketState { |
|
|
|
|
@ -580,7 +567,7 @@ impl WebSocketState { |
|
|
|
|
|
|
|
|
|
#[cfg(test)] |
|
|
|
|
mod tests { |
|
|
|
|
use super::{WebSocket, Role, Message, WebSocketConfig}; |
|
|
|
|
use super::{Message, Role, WebSocket, WebSocketConfig}; |
|
|
|
|
|
|
|
|
|
use std::io; |
|
|
|
|
use std::io::Cursor; |
|
|
|
|
@ -602,57 +589,53 @@ mod tests { |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#[test] |
|
|
|
|
fn receive_messages() { |
|
|
|
|
let incoming = Cursor::new(vec![ |
|
|
|
|
0x89, 0x02, 0x01, 0x02, |
|
|
|
|
0x8a, 0x01, 0x03, |
|
|
|
|
0x01, 0x07, |
|
|
|
|
0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x2c, 0x20, |
|
|
|
|
0x80, 0x06, |
|
|
|
|
0x57, 0x6f, 0x72, 0x6c, 0x64, 0x21, |
|
|
|
|
0x82, 0x03, |
|
|
|
|
0x01, 0x02, 0x03, |
|
|
|
|
0x89, 0x02, 0x01, 0x02, 0x8a, 0x01, 0x03, 0x01, 0x07, 0x48, 0x65, 0x6c, 0x6c, 0x6f, |
|
|
|
|
0x2c, 0x20, 0x80, 0x06, 0x57, 0x6f, 0x72, 0x6c, 0x64, 0x21, 0x82, 0x03, 0x01, 0x02, |
|
|
|
|
0x03, |
|
|
|
|
]); |
|
|
|
|
let mut socket = WebSocket::from_raw_socket(WriteMoc(incoming), Role::Client, None); |
|
|
|
|
assert_eq!(socket.read_message().unwrap(), Message::Ping(vec![1, 2])); |
|
|
|
|
assert_eq!(socket.read_message().unwrap(), Message::Pong(vec![3])); |
|
|
|
|
assert_eq!(socket.read_message().unwrap(), Message::Text("Hello, World!".into())); |
|
|
|
|
assert_eq!(socket.read_message().unwrap(), Message::Binary(vec![0x01, 0x02, 0x03])); |
|
|
|
|
assert_eq!( |
|
|
|
|
socket.read_message().unwrap(), |
|
|
|
|
Message::Text("Hello, World!".into()) |
|
|
|
|
); |
|
|
|
|
assert_eq!( |
|
|
|
|
socket.read_message().unwrap(), |
|
|
|
|
Message::Binary(vec![0x01, 0x02, 0x03]) |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
#[test] |
|
|
|
|
fn size_limiting_text_fragmented() { |
|
|
|
|
let incoming = Cursor::new(vec![ |
|
|
|
|
0x01, 0x07, |
|
|
|
|
0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x2c, 0x20, |
|
|
|
|
0x80, 0x06, |
|
|
|
|
0x57, 0x6f, 0x72, 0x6c, 0x64, 0x21, |
|
|
|
|
0x01, 0x07, 0x48, 0x65, 0x6c, 0x6c, 0x6f, 0x2c, 0x20, 0x80, 0x06, 0x57, 0x6f, 0x72, |
|
|
|
|
0x6c, 0x64, 0x21, |
|
|
|
|
]); |
|
|
|
|
let limit = WebSocketConfig { |
|
|
|
|
max_message_size: Some(10), |
|
|
|
|
..WebSocketConfig::default() |
|
|
|
|
}; |
|
|
|
|
let mut socket = WebSocket::from_raw_socket(WriteMoc(incoming), Role::Client, Some(limit)); |
|
|
|
|
assert_eq!(socket.read_message().unwrap_err().to_string(), |
|
|
|
|
assert_eq!( |
|
|
|
|
socket.read_message().unwrap_err().to_string(), |
|
|
|
|
"Space limit exceeded: Message too big: 7 + 6 > 10" |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#[test] |
|
|
|
|
fn size_limiting_binary() { |
|
|
|
|
let incoming = Cursor::new(vec![ |
|
|
|
|
0x82, 0x03, |
|
|
|
|
0x01, 0x02, 0x03, |
|
|
|
|
]); |
|
|
|
|
let incoming = Cursor::new(vec![0x82, 0x03, 0x01, 0x02, 0x03]); |
|
|
|
|
let limit = WebSocketConfig { |
|
|
|
|
max_message_size: Some(2), |
|
|
|
|
..WebSocketConfig::default() |
|
|
|
|
}; |
|
|
|
|
let mut socket = WebSocket::from_raw_socket(WriteMoc(incoming), Role::Client, Some(limit)); |
|
|
|
|
assert_eq!(socket.read_message().unwrap_err().to_string(), |
|
|
|
|
assert_eq!( |
|
|
|
|
socket.read_message().unwrap_err().to_string(), |
|
|
|
|
"Space limit exceeded: Message too big: 0 + 3 > 2" |
|
|
|
|
); |
|
|
|
|
} |
|
|
|
|
|
Daniel Abramov
5 years ago
committed by