Unverified Commit 9498fc8a authored by Dominik Charousset's avatar Dominik Charousset Committed by GitHub

Merge pull request #1470

Separate WebSocket control frames
parents f3ccebc3 c5921b51
......@@ -132,7 +132,8 @@ private:
// nop
}
bool handle(uint8_t opcode, byte_span payload);
// Returns `frame_size` on success and -1 on error.
ptrdiff_t handle(uint8_t opcode, byte_span payload, size_t frame_size);
void ship_pong(byte_span payload);
......
......@@ -84,9 +84,6 @@ ptrdiff_t rfc6455::decode_header(const_byte_span data, header& hdr) {
bool masked = (byte2 & 0x80) != 0;
auto len_field = byte2 & 0x7F;
size_t header_length;
// control frames can only have payload up to 125 bytes
if (rfc6455::is_control_frame(hdr.opcode) && len_field > 125)
return -1;
if (len_field < 126) {
header_length = 2 + (masked ? 4 : 0);
hdr.payload_len = len_field;
......
......@@ -38,11 +38,29 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span) {
// Wait for more input.
return 0;
}
// Make sure the entire frame (including header) fits into max_frame_size.
if (hdr.payload_len >= (max_frame_size - static_cast<size_t>(hdr_bytes))) {
CAF_LOG_DEBUG("WebSocket frame too large");
abort_and_shutdown(sec::protocol_error, "WebSocket frame too large");
return -1;
if (detail::rfc6455::is_control_frame(hdr.opcode)) {
// control frames can only have payload up to 125 bytes
if (hdr.payload_len > 125) {
CAF_LOG_DEBUG("WebSocket control frame payload exceeds allowed size");
abort_and_shutdown(sec::protocol_error, "WebSocket control frame payload "
"exceeds allowed size");
return -1;
}
} else {
// Make sure the entire frame (including header) fits into max_frame_size.
if (hdr.payload_len >= max_frame_size - static_cast<size_t>(hdr_bytes)) {
CAF_LOG_DEBUG("WebSocket frame too large");
abort_and_shutdown(sec::protocol_error, "WebSocket frame too large");
return -1;
}
// Reject any payload that exceeds max_frame_size. This covers assembled
// payloads as well by including payload_buf_.
if (payload_buf_.size() + hdr.payload_len > max_frame_size) {
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
abort_and_shutdown(sec::protocol_error, "fragmented WebSocket payload "
"exceeds maximum size");
return -1;
}
}
// Wait for more data if necessary.
size_t frame_size = hdr_bytes + hdr.payload_len;
......@@ -56,65 +74,52 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span) {
if (hdr.mask_key != 0) {
detail::rfc6455::mask_data(hdr.mask_key, payload);
}
// Handle control frames first, since these may not me fragmented,
// and can arrive between regular message fragments.
if (detail::rfc6455::is_control_frame(hdr.opcode)
&& hdr.opcode != detail::rfc6455::continuation_frame) {
if (!hdr.fin) {
abort_and_shutdown(sec::protocol_error,
"received a fragmented WebSocket control message");
return -1;
}
return handle(hdr.opcode, payload, frame_size);
}
if (hdr.fin) {
if (opcode_ == nil_code) {
// Call upper layer.
if (hdr.opcode == detail::rfc6455::connection_close) {
abort_and_shutdown(sec::connection_closed);
return -1;
} else if (!handle(hdr.opcode, payload)) {
return -1;
}
} else if (hdr.opcode != detail::rfc6455::continuation_frame) {
return handle(hdr.opcode, payload, frame_size);
}
if (hdr.opcode != detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("expected a WebSocket continuation_frame");
abort_and_shutdown(sec::protocol_error,
"expected a WebSocket continuation_frame");
return -1;
} else if (payload_buf_.size() + payload_len > max_frame_size) {
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
abort_and_shutdown(sec::protocol_error, "fragmented WebSocket payload "
"exceeds maximum size");
return -1;
} else {
if (hdr.opcode == detail::rfc6455::connection_close) {
abort_and_shutdown(sec::connection_closed);
return -1;
} else {
// End of fragmented input.
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
if (!handle(opcode_, payload_buf_)) {
return -1;
}
opcode_ = nil_code;
payload_buf_.clear();
}
}
} else {
// The first frame must not be a continuation frame. Any frame that is not
// the first frame must be a continuation frame.
if (opcode_ == nil_code) {
if (hdr.opcode == detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("received WebSocket continuation "
"frame without prior opcode");
abort_and_shutdown(sec::protocol_error,
"received WebSocket continuation "
"frame without prior opcode");
return -1;
}
opcode_ = hdr.opcode;
} else if (hdr.opcode != detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("expected a continuation frame");
abort_and_shutdown(sec::protocol_error, "expected a continuation frame");
return -1;
} else if (payload_buf_.size() + payload_len > max_frame_size) {
// Reject assembled payloads that exceed max_frame_size.
CAF_LOG_DEBUG("fragmented WebSocket payload exceeds maximum size");
abort_and_shutdown(sec::protocol_error, "fragmented WebSocket payload "
"exceeds maximum size");
// End of fragmented input.
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
auto result = handle(opcode_, payload_buf_, frame_size);
opcode_ = nil_code;
payload_buf_.clear();
return result;
}
// The first frame must not be a continuation frame. Any frame that is not
// the first frame must be a continuation frame.
if (opcode_ == nil_code) {
if (hdr.opcode == detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("received WebSocket continuation "
"frame without prior opcode");
abort_and_shutdown(sec::protocol_error, "received WebSocket continuation "
"frame without prior opcode");
return -1;
}
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
opcode_ = hdr.opcode;
} else if (hdr.opcode != detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("expected a continuation frame");
abort_and_shutdown(sec::protocol_error, "expected a continuation frame");
return -1;
}
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
return static_cast<ptrdiff_t>(frame_size);
}
......@@ -186,27 +191,32 @@ bool framing::end_text_message() {
// -- implementation details ---------------------------------------------------
bool framing::handle(uint8_t opcode, byte_span payload) {
// Code rfc6455::connection_close must be treated separately.
CAF_ASSERT(opcode != detail::rfc6455::connection_close);
ptrdiff_t framing::handle(uint8_t opcode, byte_span payload,
size_t frame_size) {
// opcodes are checked for validity when decoding the header
switch (opcode) {
case detail::rfc6455::connection_close:
abort_and_shutdown(sec::connection_closed);
return -1;
case detail::rfc6455::text_frame: {
std::string_view text{reinterpret_cast<const char*>(payload.data()),
payload.size()};
return up_->consume_text(text) >= 0;
if (up_->consume_text(text) < 0)
return -1;
break;
}
case detail::rfc6455::binary_frame:
return up_->consume_binary(payload) >= 0;
if (up_->consume_binary(payload) < 0)
return -1;
break;
case detail::rfc6455::ping:
ship_pong(payload);
return true;
case detail::rfc6455::pong:
break;
default: // detail::rfc6455::pong
// nop
return true;
default:
// error
return false;
break;
}
return static_cast<ptrdiff_t>(frame_size);
}
void framing::ship_pong(byte_span payload) {
......
......@@ -60,6 +60,16 @@ TEST_CASE("masking") {
CHECK_EQ(masked_data, data);
}
TEST_CASE("decoding a frame with RSV bits fails") {
std::vector<uint8_t> data;
byte_buffer out = bytes({
0xF2, // FIN + RSV + binary frame opcode
0x00, // data size = 0
});
impl::header hdr;
CHECK_EQ(impl::decode_header(out, hdr), -1);
}
TEST_CASE("no mask key and no data") {
std::vector<uint8_t> data;
byte_buffer out;
......
......@@ -160,7 +160,6 @@ mock_web_socket_app::accept(const caf::net::http::request_header& hdr) {
void mock_web_socket_app::abort(const caf::error& reason) {
abort_reason = reason;
down->shutdown(reason);
}
ptrdiff_t mock_web_socket_app::consume_text(std::string_view text) {
......
......@@ -14,11 +14,15 @@ using namespace std::literals;
namespace {
struct fixture {
mock_web_socket_app* app;
net::web_socket::framing* uut;
mock_web_socket_app* app = nullptr;
net::web_socket::framing* uut = nullptr;
std::unique_ptr<mock_stream_transport> transport;
fixture() {
reset();
}
void reset() {
auto app_layer = mock_web_socket_app::make();
app = app_layer.get();
auto uut_layer
......@@ -28,12 +32,19 @@ struct fixture {
CHECK_EQ(transport->start(nullptr), error{});
transport->configure_read(caf::net::receive_policy::up_to(2048u));
}
byte_buffer make_test_data(size_t requested_size) {
return byte_buffer{requested_size, std::byte{0xFF}};
}
};
byte_buffer make_test_data(size_t requested_size) {
return byte_buffer{requested_size, std::byte{0xFF}};
}
auto bytes(std::initializer_list<uint8_t> xs) {
byte_buffer result;
for (auto x : xs)
result.emplace_back(static_cast<std::byte>(x));
return result;
}
} // namespace
BEGIN_FIXTURE_SCOPE(fixture)
......@@ -162,4 +173,312 @@ SCENARIO("the client closes the connection with a closing handshake") {
}
}
SCENARIO("ping messages may not be fragmented") {
GIVEN("a valid WebSocket connection") {
std::vector<std::byte> ping_frame;
WHEN("the client sends the first frame of a fragmented ping message") {
auto data = make_test_data(10);
detail::rfc6455::assemble_frame(detail::rfc6455::ping, 0x0, data,
ping_frame, 0);
transport->push(ping_frame);
THEN("the server aborts the application") {
CHECK_EQ(transport->handle_input(), 0);
CHECK(app->has_aborted());
CHECK_EQ(app->abort_reason, sec::protocol_error);
MESSAGE("Aborted with: " << app->abort_reason);
}
AND("the server closes the connection with a protocol error") {
detail::rfc6455::header hdr;
detail::rfc6455::decode_header(transport->output_buffer(), hdr);
MESSAGE("Buffer: " << transport->output_buffer());
CHECK_EQ(hdr.opcode, detail::rfc6455::connection_close);
CHECK(hdr.payload_len >= 2);
auto status = (std::to_integer<int>(transport->output_buffer()[2]) << 8)
+ std::to_integer<int>(transport->output_buffer()[3]);
CHECK_EQ(status,
static_cast<int>(net::web_socket::status::protocol_error));
}
}
}
}
SCENARIO("ping messages may arrive between message fragments") {
GIVEN("a valid WebSocket connection") {
WHEN("the frames arrive all at once") {
std::vector<std::byte> input;
auto fragment1 = "Hello"sv;
auto fragment2 = ", world!"sv;
auto data = as_bytes(make_span(fragment1));
detail::rfc6455::assemble_frame(detail::rfc6455::text_frame, 0x0, data,
input, 0);
transport->push(input);
input.clear();
detail::rfc6455::assemble_frame(detail::rfc6455::ping, 0x0, data, input);
transport->push(input);
input.clear();
data = as_bytes(make_span(fragment2));
detail::rfc6455::assemble_frame(detail::rfc6455::continuation_frame, 0x0,
data, input);
transport->push(input);
transport->handle_input();
THEN("the server responds with a pong") {
detail::rfc6455::header hdr;
auto hdr_len
= detail::rfc6455::decode_header(transport->output_buffer(), hdr);
MESSAGE("Payload: " << transport->output_buffer());
CHECK_EQ(hdr_len, 2u);
CHECK(hdr.fin);
CHECK_EQ(hdr.opcode, detail::rfc6455::pong);
CHECK_EQ(hdr.payload_len, 5u);
CHECK_EQ(hdr.mask_key, 0u);
}
AND("the server receives the full text message") {
CHECK_EQ(app->text_input, "Hello, world!"sv);
}
AND("the app is still running") {
CHECK(!app->has_aborted());
}
}
WHEN("the frames arrive separately") {
reset();
auto fragment1 = "Hello"sv;
auto fragment2 = ", world!"sv;
std::vector<std::byte> input;
auto data = as_bytes(make_span(fragment1));
detail::rfc6455::assemble_frame(detail::rfc6455::text_frame, 0x0, data,
input, 0);
transport->push(input);
transport->handle_input();
THEN("the server receives nothing") {
CHECK(app->text_input.empty());
CHECK(app->binary_input.empty());
}
input.clear();
detail::rfc6455::assemble_frame(detail::rfc6455::ping, 0x0, data, input);
transport->push(input);
transport->handle_input();
THEN("the server responds with a pong") {
detail::rfc6455::header hdr;
auto hdr_len
= detail::rfc6455::decode_header(transport->output_buffer(), hdr);
CHECK_EQ(hdr_len, 2u);
CHECK(hdr.fin);
CHECK_EQ(hdr.opcode, detail::rfc6455::pong);
CHECK_EQ(hdr.payload_len, 5u);
CHECK_EQ(hdr.mask_key, 0u);
}
input.clear();
data = as_bytes(make_span(fragment2));
detail::rfc6455::assemble_frame(detail::rfc6455::continuation_frame, 0x0,
data, input);
transport->push(input);
transport->handle_input();
THEN("the server receives the full text message") {
CHECK_EQ(app->text_input, "Hello, world!"sv);
}
AND("the client did not abort") {
CHECK(!app->has_aborted());
}
}
}
}
END_FIXTURE_SCOPE()
namespace {
// The following setup is a mock websocket application that rejects everything
struct rejecting_mock_web_socket_app : public mock_web_socket_app {
rejecting_mock_web_socket_app() : mock_web_socket_app(false) {
}
ptrdiff_t consume_text(std::string_view) override {
abort(make_error(sec::logic_error));
return -1;
}
ptrdiff_t consume_binary(caf::byte_span) override {
abort(make_error(sec::logic_error));
return -1;
}
};
struct rejecting_fixture {
rejecting_mock_web_socket_app* app = nullptr;
net::web_socket::framing* uut = nullptr;
std::unique_ptr<mock_stream_transport> transport;
void reset() {
auto app_layer = std::make_unique<rejecting_mock_web_socket_app>();
app = app_layer.get();
auto uut_layer
= net::web_socket::framing::make_server(std::move(app_layer));
uut = uut_layer.get();
transport = mock_stream_transport::make(std::move(uut_layer));
CHECK_EQ(transport->start(nullptr), error{});
transport->configure_read(caf::net::receive_policy::up_to(2048u));
}
};
} // namespace
BEGIN_FIXTURE_SCOPE(rejecting_fixture);
SCENARIO("apps can return errors to shut down the framing layer") {
GIVEN("an app that returns -1 for any frame it receives") {
WHEN("receiving a binary message") {
THEN("the framing layer closes the connection") {
reset();
byte_buffer input;
const auto data = make_test_data(4);
detail::rfc6455::assemble_frame(0x0, data, input);
transport->push(input);
CHECK_EQ(transport->handle_input(), 0);
CHECK(app->has_aborted());
}
}
WHEN("receiving a fragmented binary message") {
THEN("the framing layer closes the connection") {
reset();
byte_buffer frame1;
byte_buffer frame2;
const auto data = make_test_data(4);
detail::rfc6455::assemble_frame(detail::rfc6455::binary_frame, 0x0,
data, frame1, 0);
transport->push(frame1);
detail::rfc6455::assemble_frame(detail::rfc6455::continuation_frame,
0x0, data, frame2);
transport->push(frame2);
// We only handle one frame, the second one results in an abort
CHECK_EQ(transport->handle_input(),
static_cast<ptrdiff_t>(frame1.size()));
CHECK(app->has_aborted());
}
}
WHEN("receiving a text message") {
THEN("the framing layer closes the connection") {
reset();
byte_buffer input;
const auto msg = "Hello, world!"sv;
detail::rfc6455::assemble_frame(0x0, make_span(msg), input);
transport->push(input);
CHECK_EQ(transport->handle_input(), 0);
CHECK(app->has_aborted());
}
}
WHEN("receiving a fragmented text message") {
THEN("the framing layer closes the connection") {
reset();
byte_buffer frame1;
byte_buffer frame2;
const auto msg = "Hello, world!"sv;
detail::rfc6455::assemble_frame(detail::rfc6455::text_frame, 0x0,
as_bytes(make_span(msg)), frame1, 0);
transport->push(frame1);
detail::rfc6455::assemble_frame(detail::rfc6455::continuation_frame,
0x0, as_bytes(make_span(msg)), frame2);
transport->push(frame2);
CHECK_EQ(transport->handle_input(),
static_cast<ptrdiff_t>(frame1.size()));
CHECK(app->has_aborted());
}
}
}
}
SCENARIO("the application receives a pong") {
GIVEN("a valid WebSocket connection") {
WHEN("the client sends a pong") {
reset();
byte_buffer input;
const auto data = make_test_data(10);
detail::rfc6455::assemble_frame(detail::rfc6455::pong, 0x0, data, input);
transport->push(input);
THEN("the application handles the frame without actual input") {
CHECK_EQ(transport->handle_input(),
static_cast<ptrdiff_t>(input.size()));
CHECK(app->text_input.empty());
CHECK(app->binary_input.empty());
}
AND("the application is still working") {
CHECK(!app->has_aborted());
}
}
}
}
SCENARIO("apps reject frames whose payload exceeds maximum allowed size") {
GIVEN("a WebSocket app accepting payloads up to INT32_MAX") {
WHEN("receiving a single large frame") {
reset();
// Faking a large frame without the actual payload
auto frame
= bytes({0x82, // FIN + binary frame opcode
0x7F, // NO MASK + 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x80, 0x00, 0x00, 0x00, // 2 ^ 31
0xFF, 0xFF, 0xFF, 0xFF}); // first 4 bytes
transport->push(frame);
THEN("the app closes the connection with a protocol error") {
CHECK_EQ(transport->handle_input(), 0);
CHECK_EQ(app->abort_reason, sec::protocol_error);
MESSAGE("Aborted with: " << app->abort_reason);
}
}
WHEN("receiving fragmented frames whose combined payload is too large") {
reset();
byte_buffer frame;
auto data = make_test_data(256);
detail::rfc6455::assemble_frame(detail::rfc6455::binary_frame, 0x0, data,
frame, 0);
transport->push(frame);
CHECK_EQ(transport->handle_input(), static_cast<ptrdiff_t>(frame.size()));
frame.clear();
frame = bytes({0x82, // FIN + continuation frame opcode
0x7F, // NO MASK + 127 -> uint64 size
0x00, 0x00, 0x00, 0x00, 0x7f, 0xff, 0xff,
0x00, // 2 ^ 31 - 256
0xFF, 0xFF, 0xFF, 0xFF}); // first 4 masked bytes
transport->push(frame);
THEN("the app closes the connection with a protocol error") {
CHECK_EQ(transport->handle_input(), 0);
CHECK_EQ(app->abort_reason, sec::protocol_error);
MESSAGE("Aborted with: " << app->abort_reason);
}
}
}
}
SCENARIO("the application shuts down on invalid frame fragments") {
GIVEN("a client that sends invalid fragmented frames") {
WHEN("the first fragment is a continuation frame") {
reset();
byte_buffer input;
const auto data = make_test_data(10);
detail::rfc6455::assemble_frame(detail::rfc6455::continuation_frame, 0x0,
data, input, 0);
transport->push(input);
THEN("the app closes the connection with a protocol error") {
CHECK_EQ(transport->handle_input(), 0);
CHECK_EQ(app->abort_reason, sec::protocol_error);
}
}
WHEN("two starting fragments are received") {
reset();
byte_buffer input;
const auto data = make_test_data(10);
detail::rfc6455::assemble_frame(detail::rfc6455::binary_frame, 0x0, data,
input, 0);
THEN("the app accepts the first frame") {
transport->push(input);
CHECK_EQ(transport->handle_input(),
static_cast<ptrdiff_t>(input.size()));
}
AND("the app closes the connection after the second frame") {
transport->push(input);
CHECK_EQ(transport->handle_input(), 0);
CHECK_EQ(app->abort_reason, sec::protocol_error);
}
}
}
}
END_FIXTURE_SCOPE();
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment