Commit 11998e7b authored by Samir Halilcevic's avatar Samir Halilcevic

Refactor framing::consume and add coverage tests

parent 0a3666fc
...@@ -132,7 +132,8 @@ private: ...@@ -132,7 +132,8 @@ private:
// nop // 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); void ship_pong(byte_span payload);
......
...@@ -65,46 +65,41 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span) { ...@@ -65,46 +65,41 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span) {
"received a fragmented WebSocket control message"); "received a fragmented WebSocket control message");
return -1; return -1;
} }
if (hdr.opcode == detail::rfc6455::connection_close) { return handle(hdr.opcode, payload, frame_size);
abort_and_shutdown(sec::connection_closed); }
return -1; // Reject any payload that exceed max_frame_size. This covers assembled
} else if (!handle(hdr.opcode, payload)) { // payloads as well by including payload_buf_
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; return -1;
} }
} else if (hdr.fin) { if (hdr.fin) {
if (opcode_ == nil_code) { if (opcode_ == nil_code) {
// Call upper layer. // Call upper layer.
if (!handle(hdr.opcode, payload)) { return handle(hdr.opcode, payload, frame_size);
return -1;
} }
} else if (hdr.opcode != detail::rfc6455::continuation_frame) { if (hdr.opcode != detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("expected a WebSocket continuation_frame"); CAF_LOG_DEBUG("expected a WebSocket continuation_frame");
abort_and_shutdown(sec::protocol_error, abort_and_shutdown(sec::protocol_error,
"expected a WebSocket continuation_frame"); "expected a WebSocket continuation_frame");
return -1; 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 {
// End of fragmented input. // End of fragmented input.
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end()); payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
if (!handle(opcode_, payload_buf_)) { auto result = handle(opcode_, payload_buf_, frame_size);
return -1;
}
opcode_ = nil_code; opcode_ = nil_code;
payload_buf_.clear(); payload_buf_.clear();
return result;
} }
} else {
// The first frame must not be a continuation frame. Any frame that is not // The first frame must not be a continuation frame. Any frame that is not
// the first frame must be a continuation frame. // the first frame must be a continuation frame.
if (opcode_ == nil_code) { if (opcode_ == nil_code) {
if (hdr.opcode == detail::rfc6455::continuation_frame) { if (hdr.opcode == detail::rfc6455::continuation_frame) {
CAF_LOG_DEBUG("received WebSocket continuation " CAF_LOG_DEBUG("received WebSocket continuation "
"frame without prior opcode"); "frame without prior opcode");
abort_and_shutdown(sec::protocol_error, abort_and_shutdown(sec::protocol_error, "received WebSocket continuation "
"received WebSocket continuation "
"frame without prior opcode"); "frame without prior opcode");
return -1; return -1;
} }
...@@ -113,15 +108,8 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span) { ...@@ -113,15 +108,8 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span) {
CAF_LOG_DEBUG("expected a continuation frame"); CAF_LOG_DEBUG("expected a continuation frame");
abort_and_shutdown(sec::protocol_error, "expected a continuation frame"); abort_and_shutdown(sec::protocol_error, "expected a continuation frame");
return -1; 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");
return -1;
} }
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end()); payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
}
return static_cast<ptrdiff_t>(frame_size); return static_cast<ptrdiff_t>(frame_size);
} }
...@@ -193,27 +181,34 @@ bool framing::end_text_message() { ...@@ -193,27 +181,34 @@ bool framing::end_text_message() {
// -- implementation details --------------------------------------------------- // -- implementation details ---------------------------------------------------
bool framing::handle(uint8_t opcode, byte_span payload) { ptrdiff_t framing::handle(uint8_t opcode, byte_span payload,
// Code rfc6455::connection_close must be treated separately. size_t frame_size) {
CAF_ASSERT(opcode != detail::rfc6455::connection_close);
switch (opcode) { switch (opcode) {
case detail::rfc6455::connection_close:
abort_and_shutdown(sec::connection_closed);
return -1;
case detail::rfc6455::text_frame: { case detail::rfc6455::text_frame: {
std::string_view text{reinterpret_cast<const char*>(payload.data()), std::string_view text{reinterpret_cast<const char*>(payload.data()),
payload.size()}; payload.size()};
return up_->consume_text(text) >= 0; if (up_->consume_text(text) < 0)
return -1;
break;
} }
case detail::rfc6455::binary_frame: case detail::rfc6455::binary_frame:
return up_->consume_binary(payload) >= 0; if (up_->consume_binary(payload) < 0)
return -1;
break;
case detail::rfc6455::ping: case detail::rfc6455::ping:
ship_pong(payload); ship_pong(payload);
return true; break;
case detail::rfc6455::pong: case detail::rfc6455::pong:
// nop // nop
return true; break;
default: default:
// error // error
return false; return -1;
} }
return static_cast<ptrdiff_t>(frame_size);
} }
void framing::ship_pong(byte_span payload) { void framing::ship_pong(byte_span payload) {
......
...@@ -160,7 +160,6 @@ mock_web_socket_app::accept(const caf::net::http::request_header& hdr) { ...@@ -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) { void mock_web_socket_app::abort(const caf::error& reason) {
abort_reason = reason; abort_reason = reason;
down->shutdown(reason);
} }
ptrdiff_t mock_web_socket_app::consume_text(std::string_view text) { ptrdiff_t mock_web_socket_app::consume_text(std::string_view text) {
......
...@@ -29,18 +29,11 @@ struct fixture { ...@@ -29,18 +29,11 @@ struct fixture {
CHECK_EQ(transport->start(nullptr), error{}); CHECK_EQ(transport->start(nullptr), error{});
transport->configure_read(caf::net::receive_policy::up_to(2048u)); transport->configure_read(caf::net::receive_policy::up_to(2048u));
} }
};
byte_buffer make_test_data(size_t requested_size) { byte_buffer make_test_data(size_t requested_size) {
return byte_buffer{requested_size, std::byte{0xFF}}; 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 } // namespace
...@@ -289,3 +282,86 @@ SCENARIO("the client sends an fragmented text message with a ping in-between " ...@@ -289,3 +282,86 @@ SCENARIO("the client sends an fragmented text message with a ping in-between "
} }
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
// 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;
net::web_socket::framing* uut;
std::unique_ptr<mock_stream_transport> transport;
rejecting_fixture() {
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));
}
};
BEGIN_FIXTURE_SCOPE(rejecting_fixture);
TEST_CASE("send a binary message that gets rejected") {
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());
}
TEST_CASE("send a framed binary message gets rejected") {
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());
}
TEST_CASE("send a text message that gets rejected") {
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());
}
TEST_CASE("send a framed text message gets rejected") {
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());
}
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