Commit 5881b5c7 authored by Samir Halilcevic's avatar Samir Halilcevic

Use delta buffer in web_socket framing

parent b24836ec
...@@ -11,6 +11,23 @@ ...@@ -11,6 +11,23 @@
namespace caf::net::web_socket { namespace caf::net::web_socket {
namespace {
/// Checks whether the current input is valid UTF-8. Stores the last position
/// while scanning in order to avoid validating the same bytes again.
bool payload_valid(const_byte_span payload, size_t& offset) noexcept {
// validate from the index where we left off last time
auto [index, incomplete] = detail::rfc3629::validate(payload.subspan(offset));
offset += index;
if (offset == payload.size())
return true;
// incomplete will be true if the last code point is missing continuation
// bytes but might be valid
return incomplete;
}
}
// -- static utility functions ------------------------------------------------- // -- static utility functions -------------------------------------------------
error framing::validate_closing_payload(const_byte_span payload) { error framing::validate_closing_payload(const_byte_span payload) {
...@@ -104,67 +121,51 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) { ...@@ -104,67 +121,51 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) {
return -1; return -1;
} }
} }
// Calculate at what point of the received buffer the delta payload begins.
size_t frame_size = hdr_bytes + hdr.payload_len; auto offset = static_cast<ptrdiff_t>((buffer.size() - delta.size()))
// unmask the arrived data
if (hdr.mask_key != 0) {
// leave out the header part
// leave out the already part
auto offset = static_cast<ptrdiff_t>(buffer.size() - delta.size())
- hdr_bytes; - hdr_bytes;
// if the delta buffer is empty this means that we got called to the framing // Offset < zero - the delta buffer contains header bytes.
// layer, consumed one frame and returned with unconsumed bytes that // Delta is empty - we didn't process the whole input last time we got called.
// represent the second frame and the transport layer called us again, so if (delta.empty() || offset < 0)
// even though we saw the bytes beofre, we didn't unmask them.
if (delta.empty())
offset = 0; offset = 0;
if (offset < 0) // We read frame_size at most. This can leave unprocessed input.
offset = 0; auto payload = buffer.subspan(hdr_bytes, std::min(buffer.size() - hdr_bytes,
auto payload = buffer; hdr.payload_len));
if (buffer.size() > frame_size) // Unmask the arrived data.
payload = buffer.subspan(0, frame_size); if (hdr.mask_key != 0) {
detail::rfc6455::mask_data(hdr.mask_key, payload.subspan(hdr_bytes), detail::rfc6455::mask_data(hdr.mask_key, payload, offset);
offset);
} }
size_t frame_size = hdr_bytes + hdr.payload_len;
// Wait for more data if necessary. // In case of text message, we want to validate the UTF-8 encoding early.
if (buffer.size() < frame_size) {
// when handling a text frame we want to fail early on invalid UTF-8
if (hdr.opcode == detail::rfc6455::text_frame if (hdr.opcode == detail::rfc6455::text_frame
|| opcode_ == detail::rfc6455::text_frame) { || (hdr.opcode == detail::rfc6455::continuation_frame
down_->configure_read(receive_policy::up_to(frame_size)); && opcode_ == detail::rfc6455::text_frame)) {
if (hdr.opcode == detail::rfc6455::text_frame) { if (hdr.opcode == detail::rfc6455::text_frame && hdr.fin) {
auto arrived_payload if (!payload_valid(payload, validation_offset_)) {
= std::vector<std::byte>{buffer.begin() + hdr_bytes, buffer.end()};
if (auto [index, incomplete]
= detail::rfc3629::validate(arrived_payload);
index != arrived_payload.size() && !incomplete) {
abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence"); abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence");
return -1; return -1;
} }
} else { } else {
auto unvalidated_payload = std::vector<std::byte>{ payload_buf_.insert(payload_buf_.end(), payload.begin() + offset,
payload_buf_.begin() + static_cast<ptrdiff_t>(validation_offset_), payload.end());
payload_buf_.end()}; if (!payload_valid(payload_buf_, validation_offset_)) {
unvalidated_payload.insert(unvalidated_payload.end(),
buffer.begin() + hdr_bytes, buffer.end());
if (auto [index, incomplete]
= detail::rfc3629::validate(unvalidated_payload);
index != unvalidated_payload.size() && !incomplete) {
abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence"); abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence");
return -1; return -1;
} }
} }
// Wait for more data if necessary.
if (buffer.size() < frame_size) {
down_->configure_read(receive_policy::up_to(frame_size));
return 0;
}
} else { } else {
// Wait for more data if necessary.
if (buffer.size() < frame_size) {
down_->configure_read(receive_policy::exactly(frame_size)); down_->configure_read(receive_policy::exactly(frame_size));
}
return 0; return 0;
} }
}
// Decode frame. // At this point the frame is guaranteed to have arrived completely.
auto payload_len = static_cast<size_t>(hdr.payload_len);
auto payload = buffer.subspan(hdr_bytes, payload_len);
// Handle control frames first, since these may not me fragmented, // Handle control frames first, since these may not me fragmented,
// and can arrive between regular message fragments. // and can arrive between regular message fragments.
if (detail::rfc6455::is_control_frame(hdr.opcode)) { if (detail::rfc6455::is_control_frame(hdr.opcode)) {
...@@ -185,12 +186,12 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) { ...@@ -185,12 +186,12 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) {
} }
if (hdr.fin) { if (hdr.fin) {
if (opcode_ == nil_code) { if (opcode_ == nil_code) {
// Call upper layer.
if (hdr.opcode == detail::rfc6455::text_frame if (hdr.opcode == detail::rfc6455::text_frame
&& !detail::rfc3629::valid(payload)) { && validation_offset_ != payload.size()) {
abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence"); abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence");
return -1; return -1;
} }
// Call upper layer.
return handle(hdr.opcode, payload, frame_size); return handle(hdr.opcode, payload, frame_size);
} }
if (hdr.opcode != detail::rfc6455::continuation_frame) { if (hdr.opcode != detail::rfc6455::continuation_frame) {
...@@ -200,8 +201,11 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) { ...@@ -200,8 +201,11 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) {
return -1; return -1;
} }
// End of fragmented input. // End of fragmented input.
if (opcode_ != detail::rfc6455::text_frame) {
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end()); payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
if (opcode_ == detail::rfc6455::text_frame && !payload_valid()) { }
if (opcode_ == detail::rfc6455::text_frame
&& validation_offset_ != payload_buf_.size()) {
abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence"); abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence");
return -1; return -1;
} }
...@@ -220,8 +224,11 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) { ...@@ -220,8 +224,11 @@ ptrdiff_t framing::consume(byte_span buffer, byte_span delta) {
abort_and_shutdown(sec::protocol_error, "expected a continuation frame"); abort_and_shutdown(sec::protocol_error, "expected a continuation frame");
return -1; return -1;
} }
if (opcode_ != detail::rfc6455::text_frame) {
payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end()); payload_buf_.insert(payload_buf_.end(), payload.begin(), payload.end());
if (opcode_ == detail::rfc6455::text_frame && !payload_valid()) { }
if (opcode_ == detail::rfc6455::text_frame
&& !payload_valid(payload_buf_, validation_offset_)) {
abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence"); abort_and_shutdown(sec::malformed_message, "invalid UTF-8 sequence");
return -1; return -1;
} }
...@@ -312,6 +319,7 @@ ptrdiff_t framing::handle(uint8_t opcode, byte_span payload, ...@@ -312,6 +319,7 @@ ptrdiff_t framing::handle(uint8_t opcode, byte_span payload,
payload.size()}; payload.size()};
if (up_->consume_text(text) < 0) if (up_->consume_text(text) < 0)
return -1; return -1;
validation_offset_ = 0;
break; break;
} }
case detail::rfc6455::binary_frame: case detail::rfc6455::binary_frame:
...@@ -372,16 +380,4 @@ void framing::ship_closing_message(status code, std::string_view msg) { ...@@ -372,16 +380,4 @@ void framing::ship_closing_message(status code, std::string_view msg) {
down_->end_output(); down_->end_output();
} }
bool framing::payload_valid() noexcept {
// validate from the index where we left off last time
auto [index, incomplete] = detail::rfc3629::validate(
make_span(payload_buf_).subspan(validation_offset_));
validation_offset_ += index;
if (validation_offset_ == payload_buf_.size())
return true;
// incomplete will be true if the last code point is missing continuation
// bytes but might be valid
return incomplete;
}
} // namespace caf::net::web_socket } // namespace caf::net::web_socket
...@@ -164,10 +164,6 @@ private: ...@@ -164,10 +164,6 @@ private:
shutdown(err); shutdown(err);
} }
/// Checks whether the current input is valid UTF-8. Stores the last position
/// while scanning in order to avoid validating the same bytes again.
bool payload_valid() noexcept;
// -- member variables ------------------------------------------------------- // -- member variables -------------------------------------------------------
/// Points to the transport layer below. /// Points to the transport layer below.
......
...@@ -46,6 +46,8 @@ auto bytes(std::initializer_list<uint8_t> xs) { ...@@ -46,6 +46,8 @@ auto bytes(std::initializer_list<uint8_t> xs) {
} }
int fetch_status(const_byte_span payload) { int fetch_status(const_byte_span payload) {
if (payload.size() < 4)
return -1;
return (std::to_integer<int>(payload[2]) << 8) return (std::to_integer<int>(payload[2]) << 8)
+ std::to_integer<int>(payload[3]); + std::to_integer<int>(payload[3]);
} }
...@@ -333,6 +335,20 @@ SCENARIO("the application shuts down on invalid UTF-8 message") { ...@@ -333,6 +335,20 @@ SCENARIO("the application shuts down on invalid UTF-8 message") {
MESSAGE("Aborted with: " << app->abort_reason); MESSAGE("Aborted with: " << app->abort_reason);
} }
} }
WHEN("the client sends a message ending with a incomplete codepoint") {
reset();
byte_buffer frame;
detail::rfc6455::assemble_frame(detail::rfc6455::text_frame, 0x0,
data_span.first(6), frame);
transport->push(frame);
THEN("the server aborts the application") {
CHECK_EQ(transport->handle_input(), 0);
CHECK_EQ(app->abort_reason, sec::malformed_message);
CHECK_EQ(fetch_status(transport->output_buffer()),
static_cast<int>(net::web_socket::status::inconsistent_data));
MESSAGE("Aborted with: " << app->abort_reason);
}
}
WHEN("the client sends the first part of the message") { WHEN("the client sends the first part of the message") {
reset(); reset();
byte_buffer frame; byte_buffer frame;
......
...@@ -185,7 +185,12 @@ TEST_CASE("data may arrive fragmented") { ...@@ -185,7 +185,12 @@ TEST_CASE("data may arrive fragmented") {
rfc6455_append(detail::rfc6455::continuation_frame, "Socket!\n"sv, buf); rfc6455_append(detail::rfc6455::continuation_frame, "Socket!\n"sv, buf);
transport->push(buf); transport->push(buf);
CHECK_EQ(transport->handle_input(), static_cast<ptrdiff_t>(buf.size())); CHECK_EQ(transport->handle_input(), static_cast<ptrdiff_t>(buf.size()));
CHECK_EQ(app->text_input, "Hello WebSocket!\nBye WebSocket!\n"); auto expected = "Hello WebSocket!\nBye WebSocket!\n"sv;
CHECK_EQ(app->text_input.size(), expected.size());
for (auto i = 0ul; i < expected.size(); i++) {
CHECK_EQ(app->text_input.at(i), expected.at(i));
}
MESSAGE(app->text_input);
CHECK(!app->has_aborted()); CHECK(!app->has_aborted());
} }
......
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