Crow  1.1
A C++ microframework for the web
 
Loading...
Searching...
No Matches
websocket.h
1#pragma once
2#include <array>
3#include <memory>
4#include <optional>
5#include <string>
6#include <thread>
7#include "crow/http_response.h"
8#include "crow/logging.h"
9#include "crow/socket_adaptors.h"
10#include "crow/http_request.h"
11#include "crow/tcp_socket_options.h"
12#include "crow/TinySHA1.hpp"
13#include "crow/utility.h"
14
15namespace crow // NOTE: Already documented in "crow/app.h"
16{
17#ifdef CROW_USE_BOOST
18 namespace asio = boost::asio;
19 using error_code = boost::system::error_code;
20#else
21 using error_code = asio::error_code;
22#endif
23
24 /**
25 * \namespace crow::websocket
26 * \brief Namespace that includes the \ref Connection class
27 * and \ref connection struct. Useful for WebSockets connection.
28 *
29 * Used specially in crow/websocket.h, crow/app.h and crow/routing.h
30 */
31 namespace websocket
32 {
33 enum class WebSocketReadState
34 {
35 MiniHeader,
36 Len16,
37 Len64,
38 Mask,
39 Payload,
40 };
41
42 // Codes taken from https://www.rfc-editor.org/rfc/rfc6455#section-7.4.1
43 enum CloseStatusCode : uint16_t {
44 NormalClosure = 1000,
45 EndpointGoingAway = 1001,
46 ProtocolError = 1002,
47 UnacceptableData = 1003,
48 InconsistentData = 1007,
49 PolicyViolated = 1008,
50 MessageTooBig = 1009,
51 ExtensionsNotNegotiated = 1010,
52 UnexpectedCondition = 1011,
53
54 // Reserved for applications only, should not send/receive these to/from clients
55 NoStatusCodePresent = 1005,
56 ClosedAbnormally = 1006,
57 TLSHandshakeFailure = 1015,
58
59 StartStatusCodesForLibraries = 3000,
60 StartStatusCodesForPrivateUse = 4000,
61 // Status code should be between 1000 and 4999 inclusive
62 StartStatusCodes = NormalClosure,
63 EndStatusCodes = 4999,
64 };
65
66 /// A base class for websocket connection.
67 struct connection
68 {
69 virtual void send_binary(std::string msg) = 0;
70 virtual void send_text(std::string msg) = 0;
71 virtual void send_ping(std::string msg) = 0;
72 virtual void send_pong(std::string msg) = 0;
73 virtual void close(std::string const& msg = "quit", uint16_t status_code = CloseStatusCode::NormalClosure) = 0;
74 virtual std::string get_remote_ip() = 0;
75 virtual uint16_t get_remote_port() = 0;
76 virtual std::string get_subprotocol() const = 0;
77 virtual ~connection() = default;
78
79 void userdata(void* u) { userdata_ = u; }
80 void* userdata() { return userdata_; }
81
82 private:
83 void* userdata_;
84 };
85
86 // Modified version of the illustration in RFC6455 Section-5.2
87 //
88 //
89 // 0 1 2 3 -byte
90 // 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 -bit
91 // +-+-+-+-+-------+-+-------------+-------------------------------+
92 // |F|R|R|R| opcode|M| Payload len | Extended payload length |
93 // |I|S|S|S| (4) |A| (7) | (16/64) |
94 // |N|V|V|V| |S| | (if payload len==126/127) |
95 // | |1|2|3| |K| | |
96 // +-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
97 // | Extended payload length continued, if payload len == 127 |
98 // + - - - - - - - - - - - - - - - +-------------------------------+
99 // | |Masking-key, if MASK set to 1 |
100 // +-------------------------------+-------------------------------+
101 // | Masking-key (continued) | Payload Data |
102 // +-------------------------------- - - - - - - - - - - - - - - - +
103 // : Payload Data continued ... :
104 // + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
105 // | Payload Data continued ... |
106 // +---------------------------------------------------------------+
107 //
108
109 /// A websocket connection.
110
111 template<typename Adaptor, typename Handler>
112 class Connection : public connection, public std::enable_shared_from_this<Connection<Adaptor, Handler>>
113 {
114 public:
115 /// Factory for a connection.
116 ///
117 /// Requires a request with an "Upgrade: websocket" header.<br>
118 /// Automatically handles the handshake.
119 static void create(const crow::request& req, Adaptor adaptor, Handler* handler,
120 uint64_t max_payload, const std::vector<std::string>& subprotocols,
121 std::function<void(crow::websocket::connection&)> open_handler,
122 std::function<void(crow::websocket::connection&, const std::string&, bool)> message_handler,
123 std::function<void(crow::websocket::connection&, const std::string&, uint16_t)> close_handler,
124 std::function<void(crow::websocket::connection&, const std::string&)> error_handler,
125 std::function<void(const crow::request&, std::optional<crow::response>&, void**)> accept_handler,
126 bool mirror_protocols,
127 const detail::socket::tcp_socket_options& tcp_options = {})
128 {
129 auto conn = std::shared_ptr<Connection>(new Connection(std::move(adaptor),
130 handler, max_payload,
131 std::move(open_handler),
132 std::move(message_handler),
133 std::move(close_handler),
134 std::move(error_handler),
135 std::move(accept_handler)));
136
137 // Apply TCP socket options to WebSocket connection
138 detail::socket::apply_tcp_socket_options(conn->adaptor_.socket(), tcp_options);
139
140 // Perform handshake validation
141 if (!utility::string_equals(req.get_header_value("upgrade"), "websocket"))
142 {
143 conn->adaptor_.close();
144 return;
145 }
146
147 std::string requested_subprotocols_header = req.get_header_value("Sec-WebSocket-Protocol");
148 if (!subprotocols.empty() || !requested_subprotocols_header.empty())
149 {
150 auto requested_subprotocols = utility::split(requested_subprotocols_header, ", ");
151 auto subprotocol = utility::find_first_of(subprotocols.begin(), subprotocols.end(), requested_subprotocols.begin(), requested_subprotocols.end());
152 if (subprotocol != subprotocols.end())
153 {
154 conn->subprotocol_ = *subprotocol;
155 }
156 }
157
158 if (mirror_protocols & !requested_subprotocols_header.empty())
159 {
160 conn->subprotocol_ = requested_subprotocols_header;
161 }
162
163 if (conn->accept_handler_)
164 {
165 void* ud = nullptr;
166 std::optional<crow::response> res;
167 conn->accept_handler_(req, res, &ud);
168 if (res)
169 {
170 std::vector<asio::const_buffer> buffers;
171 auto server_name = "";
172 std::string content_length_buffer;
173 res->write_header_into_buffer(buffers, content_length_buffer, req.keep_alive, server_name);
174 buffers.emplace_back(res->body.data(), res->body.size());
175 error_code ec;
176 asio::write(conn->adaptor_.socket(), buffers, ec);
177 conn->adaptor_.close();
178 return;
179 }
180 conn->userdata(ud);
181 }
182
183 // Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
184 // Sec-WebSocket-Version: 13
185 std::string magic = req.get_header_value("Sec-WebSocket-Key") + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
186 sha1::SHA1 s;
187 s.processBytes(magic.data(), magic.size());
188 uint8_t digest[20];
189 s.getDigestBytes(digest);
190
191 conn->handler_->add_websocket(conn);
192 conn->start(crow::utility::base64encode((unsigned char*)digest, 20));
193 }
194
195 ~Connection() noexcept override = default;
196
197 template<typename Callable>
198 struct WeakWrappedMessage
199 {
200 Callable callable;
201 std::weak_ptr<void> watch;
202
203 void operator()()
204 {
205 if (auto anchor = watch.lock())
206 {
207 std::move(callable)();
208 }
209 }
210 };
211
212 /// Send data through the socket.
213 template<typename CompletionHandler>
214 void dispatch(CompletionHandler&& handler)
215 {
216 asio::dispatch(adaptor_.get_io_context(),
217 WeakWrappedMessage<typename std::decay<CompletionHandler>::type>{
218 std::forward<CompletionHandler>(handler), anchor_});
219 }
220
221 /// Send data through the socket and return immediately.
222 template<typename CompletionHandler>
223 void post(CompletionHandler&& handler)
224 {
225 asio::post(adaptor_.get_io_context(),
226 WeakWrappedMessage<typename std::decay<CompletionHandler>::type>{
227 std::forward<CompletionHandler>(handler), anchor_});
228 }
229
230 /// Send a "Ping" message.
231
232 ///
233 /// Usually invoked to check if the other point is still online.
234 void send_ping(std::string msg) override
235 {
236 send_data(0x9, std::move(msg));
237 }
238
239 /// Send a "Pong" message.
240
241 ///
242 /// Usually automatically invoked as a response to a "Ping" message.
243 void send_pong(std::string msg) override
244 {
245 send_data(0xA, std::move(msg));
246 }
247
248 /// Send a binary encoded message.
249 void send_binary(std::string msg) override
250 {
251 send_data(0x2, std::move(msg));
252 }
253
254 /// Send a plaintext message.
255 void send_text(std::string msg) override
256 {
257 send_data(0x1, std::move(msg));
258 }
259
260 /// Send a close signal.
261
262 ///
263 /// Sets a flag to destroy the object once the message is sent.
264 void close(std::string const& msg, uint16_t status_code) override
265 {
266 dispatch([shared_this = this->shared_from_this(), msg, status_code]() mutable {
267 shared_this->has_sent_close_ = true;
268 if (shared_this->has_recv_close_ && !shared_this->is_close_handler_called_)
269 {
270 shared_this->is_close_handler_called_ = true;
271 if (shared_this->close_handler_)
272 shared_this->close_handler_(*shared_this, msg, status_code);
273 }
274
275 switch(status_code) {
276
277 case NoStatusCodePresent:
278 case ClosedAbnormally:
279 case TLSHandshakeFailure:
280 // These codes only report that no status code was received and must not
281 // be sent in a Close frame (RFC 6455 Section 7.4.1), so send none at all.
282 {
283 shared_this->write_buffers_.emplace_back(shared_this->build_header(0x8, 0));
284 }
285 default:
286 {
287 auto header = shared_this->build_header(0x8, msg.size() + 2);
288 char status_buf[2];
289 *(uint16_t*)(status_buf) = htons(status_code);
290
291 shared_this->write_buffers_.emplace_back(std::move(header));
292 shared_this->write_buffers_.emplace_back(std::string(status_buf, 2));
293 shared_this->write_buffers_.emplace_back(msg);
294 }
295 }
296 shared_this->do_write();
297 });
298 }
299
300 std::string get_remote_ip() override
301 {
302 return adaptor_.address();
303 }
304
305 /// Returns the TCP port of the remote endpoint, 0 when unavailable (Unix domain socket or socket already closed).
306 uint16_t get_remote_port() override
307 {
308 return adaptor_.remote_port();
309 }
310
311 void set_max_payload_size(uint64_t payload)
312 {
313 max_payload_bytes_ = payload;
314 }
315
316 /// Returns the matching client/server subprotocol, empty string if none matched.
317 std::string get_subprotocol() const override
318 {
319 return subprotocol_;
320 }
321
322 protected:
323 /// Generate the websocket headers using an opcode and the message size (in bytes).
324 std::string build_header(int opcode, size_t size)
325 {
326 char buf[2 + 8] = "\x80\x00";
327 buf[0] += opcode;
328 if (size < 126)
329 {
330 buf[1] += static_cast<char>(size);
331 return {buf, buf + 2};
332 }
333 else if (size < 0x10000)
334 {
335 buf[1] += 126;
336 *(uint16_t*)(buf + 2) = htons(static_cast<uint16_t>(size));
337 return {buf, buf + 4};
338 }
339 else
340 {
341 buf[1] += 127;
342 *reinterpret_cast<uint64_t*>(buf + 2) = ((1 == htonl(1)) ? static_cast<uint64_t>(size) : (static_cast<uint64_t>(htonl((size)&0xFFFFFFFF)) << 32) | htonl(static_cast<uint64_t>(size) >> 32));
343 return {buf, buf + 10};
344 }
345 }
346
347 /// Send the HTTP upgrade response.
348
349 ///
350 /// Finishes the handshake process, then starts reading messages from the socket.
351 void start(std::string&& hello)
352 {
353 static const std::string header =
354 "HTTP/1.1 101 Switching Protocols\r\n"
355 "Upgrade: websocket\r\n"
356 "Connection: Upgrade\r\n"
357 "Sec-WebSocket-Accept: ";
358 write_buffers_.emplace_back(header);
359 write_buffers_.emplace_back(std::move(hello));
360 write_buffers_.emplace_back(crlf);
361 if (!subprotocol_.empty())
362 {
363 write_buffers_.emplace_back("Sec-WebSocket-Protocol: ");
364 write_buffers_.emplace_back(subprotocol_);
365 write_buffers_.emplace_back(crlf);
366 }
367 write_buffers_.emplace_back(crlf);
368 do_write();
369 if (open_handler_)
370 open_handler_(*this);
371 do_read();
372 }
373
374 /// Read a websocket message.
375
376 ///
377 /// Involves:<br>
378 /// Handling headers (opcodes, size).<br>
379 /// Unmasking the payload.<br>
380 /// Reading the actual payload.<br>
381 void do_read()
382 {
383 if (has_sent_close_ && has_recv_close_)
384 {
385 close_connection_ = true;
386 adaptor_.shutdown_readwrite();
387 adaptor_.close();
388 check_destroy();
389 return;
390 }
391
392 is_reading = true;
393 switch (state_)
394 {
395 case WebSocketReadState::MiniHeader:
396 {
397 mini_header_ = 0;
398 asio::async_read(
399 adaptor_.socket(), asio::buffer(&mini_header_, 2),
400 [shared_this = this->shared_from_this()](const error_code& ec, std::size_t
401#ifdef CROW_ENABLE_DEBUG
402 bytes_transferred
403#endif
404 )
405
406 {
407 shared_this->is_reading = false;
408 shared_this->mini_header_ = ntohs(shared_this->mini_header_);
409#ifdef CROW_ENABLE_DEBUG
410
411 if (!ec && bytes_transferred != 2)
412 {
413 throw std::runtime_error("WebSocket:MiniHeader:async_read fail:asio bug?");
414 }
415#endif
416
417 if (!ec)
418 {
419 if ((shared_this->mini_header_ & 0x80) == 0x80)
420 shared_this->has_mask_ = true;
421 else //if the websocket specification is enforced and the message isn't masked, terminate the connection
422 {
423#ifndef CROW_ENFORCE_WS_SPEC
424 shared_this->has_mask_ = false;
425#else
426 shared_this->close_connection_ = true;
427 shared_this->adaptor_.shutdown_readwrite();
428 shared_this->adaptor_.close();
429 if (shared_this->error_handler_)
430 shared_this->error_handler_(*shared_this, "Client connection not masked.");
431 shared_this->check_destroy(CloseStatusCode::UnacceptableData);
432#endif
433 }
434
435 if ((shared_this->mini_header_ & 0x7f) == 127)
436 {
437 shared_this->state_ = WebSocketReadState::Len64;
438 }
439 else if ((shared_this->mini_header_ & 0x7f) == 126)
440 {
441 shared_this->state_ = WebSocketReadState::Len16;
442 }
443 else
444 {
445 shared_this->remaining_length_ = shared_this->mini_header_ & 0x7f;
446 shared_this->state_ = WebSocketReadState::Mask;
447 }
448 shared_this->do_read();
449 }
450 else
451 {
452 shared_this->close_connection_ = true;
453 shared_this->adaptor_.shutdown_readwrite();
454 shared_this->adaptor_.close();
455 if (shared_this->error_handler_)
456 shared_this->error_handler_(*shared_this, ec.message());
457 shared_this->check_destroy();
458 }
459 });
460 }
461 break;
462 case WebSocketReadState::Len16:
463 {
464 remaining_length_ = 0;
465 remaining_length16_ = 0;
466 asio::async_read(
467 adaptor_.socket(), asio::buffer(&remaining_length16_, 2),
468 [shared_this = this->shared_from_this()](const error_code& ec, std::size_t
469#ifdef CROW_ENABLE_DEBUG
470 bytes_transferred
471#endif
472 ) {
473 shared_this->is_reading = false;
474 shared_this->remaining_length16_ = ntohs(shared_this->remaining_length16_);
475 shared_this->remaining_length_ = shared_this->remaining_length16_;
476#ifdef CROW_ENABLE_DEBUG
477 if (!ec && bytes_transferred != 2)
478 {
479 throw std::runtime_error("WebSocket:Len16:async_read fail:asio bug?");
480 }
481#endif
482
483 if (!ec)
484 {
485 shared_this->state_ = WebSocketReadState::Mask;
486 shared_this->do_read();
487 }
488 else
489 {
490 shared_this->close_connection_ = true;
491 shared_this->adaptor_.shutdown_readwrite();
492 shared_this->adaptor_.close();
493 if (shared_this->error_handler_)
494 shared_this->error_handler_(*shared_this, ec.message());
495 shared_this->check_destroy();
496 }
497 });
498 }
499 break;
500 case WebSocketReadState::Len64:
501 {
502 asio::async_read(
503 adaptor_.socket(), asio::buffer(&remaining_length_, 8),
504 [shared_this = this->shared_from_this()](const error_code& ec, std::size_t
505#ifdef CROW_ENABLE_DEBUG
506 bytes_transferred
507#endif
508 ) {
509 shared_this->is_reading = false;
510 shared_this->remaining_length_ = ((1 == ntohl(1)) ? (shared_this->remaining_length_) : (static_cast<uint64_t>(ntohl((shared_this->remaining_length_)&0xFFFFFFFF)) << 32) | ntohl((shared_this->remaining_length_) >> 32));
511#ifdef CROW_ENABLE_DEBUG
512 if (!ec && bytes_transferred != 8)
513 {
514 throw std::runtime_error("WebSocket:Len16:async_read fail:asio bug?");
515 }
516#endif
517
518 if (!ec)
519 {
520 shared_this->state_ = WebSocketReadState::Mask;
521 shared_this->do_read();
522 }
523 else
524 {
525 shared_this->close_connection_ = true;
526 shared_this->adaptor_.shutdown_readwrite();
527 shared_this->adaptor_.close();
528 if (shared_this->error_handler_)
529 shared_this->error_handler_(*shared_this, ec.message());
530 shared_this->check_destroy();
531 }
532 });
533 }
534 break;
535 case WebSocketReadState::Mask:
536 if (message_.size() > max_payload_bytes_ ||
537 remaining_length_ > max_payload_bytes_ - message_.size())
538 {
539 close_connection_ = true;
540 adaptor_.close();
541 if (error_handler_)
542 error_handler_(*this, "Message length exceeds maximum payload.");
543 check_destroy(MessageTooBig);
544 }
545 else if (has_mask_)
546 {
547 asio::async_read(
548 adaptor_.socket(), asio::buffer((char*)&mask_, 4),
549 [shared_this = this->shared_from_this()](const error_code& ec, std::size_t
550#ifdef CROW_ENABLE_DEBUG
551 bytes_transferred
552#endif
553 ) {
554 shared_this->is_reading = false;
555#ifdef CROW_ENABLE_DEBUG
556 if (!ec && bytes_transferred != 4)
557 {
558 throw std::runtime_error("WebSocket:Mask:async_read fail:asio bug?");
559 }
560#endif
561
562 if (!ec)
563 {
564 shared_this->state_ = WebSocketReadState::Payload;
565 shared_this->do_read();
566 }
567 else
568 {
569 shared_this->close_connection_ = true;
570 if (shared_this->error_handler_)
571 shared_this->error_handler_(*shared_this, ec.message());
572 shared_this->adaptor_.shutdown_readwrite();
573 shared_this->adaptor_.close();
574 shared_this->check_destroy();
575 }
576 });
577 }
578 else
579 {
580 state_ = WebSocketReadState::Payload;
581 do_read();
582 }
583 break;
584 case WebSocketReadState::Payload:
585 {
586 auto to_read = static_cast<std::uint64_t>(buffer_.size());
587 if (remaining_length_ < to_read)
588 to_read = remaining_length_;
589 adaptor_.socket().async_read_some(
590 asio::buffer(buffer_, static_cast<std::size_t>(to_read)),
591 [shared_this = this->shared_from_this()](const error_code& ec, std::size_t bytes_transferred) {
592 shared_this->is_reading = false;
593
594 if (!ec)
595 {
596 shared_this->fragment_.insert(shared_this->fragment_.end(), shared_this->buffer_.begin(), shared_this->buffer_.begin() + bytes_transferred);
597 shared_this->remaining_length_ -= bytes_transferred;
598 if (shared_this->remaining_length_ == 0)
599 {
600 if (shared_this->handle_fragment())
601 {
602 shared_this->state_ = WebSocketReadState::MiniHeader;
603 shared_this->do_read();
604 }
605 }
606 else
607 shared_this->do_read();
608 }
609 else
610 {
611 shared_this->close_connection_ = true;
612 if (shared_this->error_handler_)
613 shared_this->error_handler_(*shared_this, ec.message());
614 shared_this->adaptor_.shutdown_readwrite();
615 shared_this->adaptor_.close();
616 shared_this->check_destroy();
617 }
618 });
619 }
620 break;
621 }
622 }
623
624 /// Check if the FIN bit is set.
625 bool is_FIN()
626 {
627 return mini_header_ & 0x8000;
628 }
629
630 /// Extract the opcode from the header.
631 int opcode()
632 {
633 return (mini_header_ & 0x0f00) >> 8;
634 }
635
636 /// Process the payload fragment.
637
638 ///
639 /// Unmasks the fragment, checks the opcode, merges fragments into 1 message body, and calls the appropriate handler.
640 bool handle_fragment()
641 {
642 if (has_mask_)
643 {
644 for (decltype(fragment_.length()) i = 0; i < fragment_.length(); i++)
645 {
646 fragment_[i] ^= ((char*)&mask_)[i % 4];
647 }
648 }
649 if ((mini_header_ & 0x7000) != 0)
650 {
651 if (error_handler_)
652 error_handler_(*this, "RSV bits set without a negotiated extension.");
653 close("", CloseStatusCode::ProtocolError);
654 check_destroy(CloseStatusCode::ProtocolError);
655 return false;
656 }
657 if ((opcode() & 0x8) != 0 && (!is_FIN() || fragment_.size() > 125))
658 {
659 if (error_handler_)
660 error_handler_(*this, "Fragmented or oversized control frame.");
661 close("", CloseStatusCode::ProtocolError);
662 check_destroy(CloseStatusCode::ProtocolError);
663 return false;
664 }
665 switch (opcode())
666 {
667 case 0: // Continuation
668 {
669 if (!has_unfinished_message_)
670 {
671 if (error_handler_)
672 error_handler_(*this, "Continuation frame without a message to continue.");
673 close("", CloseStatusCode::ProtocolError);
674 check_destroy(CloseStatusCode::ProtocolError);
675 return false;
676 }
677 message_ += fragment_;
678 if (is_FIN())
679 {
680 has_unfinished_message_ = false;
681 if (message_handler_)
682 message_handler_(*this, message_, is_binary_);
683 message_.clear();
684 }
685 }
686 break;
687 case 1: // Text
688 case 2: // Binary
689 {
690 if (has_unfinished_message_)
691 {
692 if (error_handler_)
693 error_handler_(*this, "New message before the fragmented message was finished.");
694 close("", CloseStatusCode::ProtocolError);
695 check_destroy(CloseStatusCode::ProtocolError);
696 return false;
697 }
698 is_binary_ = opcode() == 2;
699 message_ += fragment_;
700 if (is_FIN())
701 {
702 if (message_handler_)
703 message_handler_(*this, message_, is_binary_);
704 message_.clear();
705 }
706 else
707 {
708 has_unfinished_message_ = true;
709 }
710 }
711 break;
712 case 0x8: // Close
713 {
714 has_recv_close_ = true;
715 uint16_t status_code = NoStatusCodePresent;
716 std::string::size_type message_start = 2;
717 if (fragment_.size() >= 2)
718 {
719 status_code = ntohs(((uint16_t*)fragment_.data())[0]);
720 } else {
721 // no message will crash substr
722 message_start = 0;
723 }
724
725 if (!has_sent_close_)
726 {
727 if (fragment_.size() == 1 || (fragment_.size() >= 2 && !is_valid_close_code(status_code)))
728 {
729 if (error_handler_)
730 error_handler_(*this, "Invalid close frame.");
731 close("", CloseStatusCode::ProtocolError);
732 }
733 else
734 {
735 close(fragment_.substr(message_start), status_code);
736 }
737 }
738 else
739 {
740
741 close_connection_ = true;
742 if (!is_close_handler_called_)
743 {
744 if (close_handler_)
745 close_handler_(*this, fragment_.substr(message_start), status_code);
746 is_close_handler_called_ = true;
747 }
748 adaptor_.shutdown_readwrite();
749 adaptor_.close();
750
751 // Close handler must have been called at this point so code does not matter
752 check_destroy();
753 return false;
754 }
755 }
756 break;
757 case 0x9: // Ping
758 {
759 send_pong(fragment_);
760 }
761 break;
762 case 0xA: // Pong
763 {
764 pong_received_ = true;
765 }
766 break;
767 default:
768 {
769 if (error_handler_)
770 error_handler_(*this, "Reserved opcode.");
771 close("", CloseStatusCode::ProtocolError);
772 check_destroy(CloseStatusCode::ProtocolError);
773 return false;
774 }
775 }
776
777 fragment_.clear();
778 return true;
779 }
780
781 /// Send the buffers' data through the socket.
782
783 ///
784 /// Also destroys the object if the Close flag is set.
785 void do_write()
786 {
787 if (sending_buffers_.empty()) {
788 if (write_buffers_.empty()) return;
789
790 sending_buffers_.swap(write_buffers_);
791 std::vector<asio::const_buffer> buffers;
792 buffers.reserve(sending_buffers_.size());
793 for (auto &s: sending_buffers_)
794 {
795 buffers.emplace_back(asio::buffer(s));
796 }
797 auto watch = std::weak_ptr<void>{anchor_};
798 asio::async_write(
799 adaptor_.socket(), buffers,
800 [shared_this = this->shared_from_this(), watch](const error_code &ec, std::size_t /*bytes_transferred*/) {
801 auto anchor = watch.lock();
802 if (anchor == nullptr)
803 return;
804
805 if (!ec && !shared_this->close_connection_)
806 {
807 shared_this->sending_buffers_.clear();
808 if (!shared_this->write_buffers_.empty())
809 shared_this->do_write();
810 if (shared_this->has_sent_close_)
811 shared_this->close_connection_ = true;
812 }
813 else
814 {
815 shared_this->sending_buffers_.clear();
816 shared_this->close_connection_ = true;
817 shared_this->check_destroy();
818 }
819 });
820 }
821 }
822
823 /// Destroy the Connection.
824 /// Whether a peer may send the status code in a Close frame (RFC 6455 Section 7.4).
825 static bool is_valid_close_code(uint16_t code)
826 {
827 return (code >= 1000 && code <= 1003) || (code >= 1007 && code <= 1014) || (code >= 3000 && code <= 4999);
828 }
829
830 void check_destroy(websocket::CloseStatusCode code = CloseStatusCode::ClosedAbnormally)
831 {
832 // Note that if the close handler was not yet called at this point we did not receive a close packet (or send one)
833 // and thus we use ClosedAbnormally unless instructed otherwise
834 if (!is_close_handler_called_)
835 {
836 if (close_handler_)
837 {
838 close_handler_(*this, "uncleanly", code);
839 }
840 }
841
842 handler_->remove_websocket(this->shared_from_this());
843 }
844
845
846 struct SendMessageType
847 {
848 std::string payload;
849 Connection* self;
850 int opcode;
851
852 void operator()()
853 {
854 self->send_data_impl(this);
855 }
856 };
857
858 void send_data_impl(SendMessageType* s)
859 {
860 auto header = build_header(s->opcode, s->payload.size());
861 write_buffers_.emplace_back(std::move(header));
862 write_buffers_.emplace_back(std::move(s->payload));
863 do_write();
864 }
865
866 void send_data(int opcode, std::string&& msg)
867 {
868 SendMessageType event_arg{
869 std::move(msg),
870 this,
871 opcode};
872
873 post(std::move(event_arg));
874 }
875
876 private:
877 Connection(Adaptor&& adaptor, Handler* handler, uint64_t max_payload,
878 std::function<void(crow::websocket::connection&)> open_handler,
879 std::function<void(crow::websocket::connection&, const std::string&, bool)> message_handler,
880 std::function<void(crow::websocket::connection&, const std::string&, uint16_t)> close_handler,
881 std::function<void(crow::websocket::connection&, const std::string&)> error_handler,
882 std::function<void(const crow::request&, std::optional<crow::response>&, void**)> accept_handler):
883 adaptor_(std::move(adaptor)),
884 handler_(handler),
885 max_payload_bytes_(max_payload),
886 open_handler_(std::move(open_handler)),
887 message_handler_(std::move(message_handler)),
888 close_handler_(std::move(close_handler)),
889 error_handler_(std::move(error_handler)),
890 accept_handler_(std::move(accept_handler))
891 {}
892
893 Adaptor adaptor_;
894 Handler* handler_;
895
896 std::vector<std::string> sending_buffers_;
897 std::vector<std::string> write_buffers_;
898
899 std::array<char, 4096> buffer_;
900 bool is_binary_;
901 bool has_unfinished_message_{false};
902 std::string message_;
903 std::string fragment_;
904 WebSocketReadState state_{WebSocketReadState::MiniHeader};
905 uint16_t remaining_length16_{0};
906 uint64_t remaining_length_{0};
907 uint64_t max_payload_bytes_{UINT64_MAX};
908 std::string subprotocol_;
909 bool close_connection_{false};
910 bool is_reading{false};
911 bool has_mask_{false};
912 uint32_t mask_;
913 uint16_t mini_header_;
914 bool has_sent_close_{false};
915 bool has_recv_close_{false};
916 bool error_occurred_{false};
917 bool pong_received_{false};
918 bool is_close_handler_called_{false};
919
920 std::shared_ptr<void> anchor_ = std::make_shared<int>(); // Value is just for placeholding
921
922 std::function<void(crow::websocket::connection&)> open_handler_;
923 std::function<void(crow::websocket::connection&, const std::string&, bool)> message_handler_;
924 std::function<void(crow::websocket::connection&, const std::string&, uint16_t status_code)> close_handler_;
925 std::function<void(crow::websocket::connection&, const std::string&)> error_handler_;
926 std::function<void(const crow::request&, std::optional<crow::response>&, void**)> accept_handler_;
927 };
928 } // namespace websocket
929} // namespace crow
TinySHA1.hpp
TinySHA1 - a header only implementation of the SHA1 algorithm in C++. Based on the implementation in ...
crow::websocket::Connection
A websocket connection.
Definition websocket.h:113
crow::websocket::Connection::dispatch
void dispatch(CompletionHandler &&handler)
Send data through the socket.
Definition websocket.h:214
crow::websocket::Connection::do_read
void do_read()
Read a websocket message.
Definition websocket.h:381
crow::websocket::Connection::send_pong
void send_pong(std::string msg) override
Send a "Pong" message.
Definition websocket.h:243
crow::websocket::Connection::handle_fragment
bool handle_fragment()
Process the payload fragment.
Definition websocket.h:640
crow::websocket::Connection::build_header
std::string build_header(int opcode, size_t size)
Generate the websocket headers using an opcode and the message size (in bytes).
Definition websocket.h:324
crow::websocket::Connection::close
void close(std::string const &msg, uint16_t status_code) override
Send a close signal.
Definition websocket.h:264
crow::websocket::Connection::send_text
void send_text(std::string msg) override
Send a plaintext message.
Definition websocket.h:255
crow::websocket::Connection::get_subprotocol
std::string get_subprotocol() const override
Returns the matching client/server subprotocol, empty string if none matched.
Definition websocket.h:317
crow::websocket::Connection::opcode
int opcode()
Extract the opcode from the header.
Definition websocket.h:631
crow::websocket::Connection::do_write
void do_write()
Send the buffers' data through the socket.
Definition websocket.h:785
crow::websocket::Connection::start
void start(std::string &&hello)
Send the HTTP upgrade response.
Definition websocket.h:351
crow::websocket::Connection::is_valid_close_code
static bool is_valid_close_code(uint16_t code)
Definition websocket.h:825
crow::websocket::Connection::is_FIN
bool is_FIN()
Check if the FIN bit is set.
Definition websocket.h:625
crow::websocket::Connection::get_remote_port
uint16_t get_remote_port() override
Returns the TCP port of the remote endpoint, 0 when unavailable (Unix domain socket or socket already...
Definition websocket.h:306
crow::websocket::Connection::send_ping
void send_ping(std::string msg) override
Send a "Ping" message.
Definition websocket.h:234
crow::websocket::Connection::send_binary
void send_binary(std::string msg) override
Send a binary encoded message.
Definition websocket.h:249
crow::websocket::Connection::create
static void create(const crow::request &req, Adaptor adaptor, Handler *handler, uint64_t max_payload, const std::vector< std::string > &subprotocols, std::function< void(crow::websocket::connection &)> open_handler, std::function< void(crow::websocket::connection &, const std::string &, bool)> message_handler, std::function< void(crow::websocket::connection &, const std::string &, uint16_t)> close_handler, std::function< void(crow::websocket::connection &, const std::string &)> error_handler, std::function< void(const crow::request &, std::optional< crow::response > &, void **)> accept_handler, bool mirror_protocols, const detail::socket::tcp_socket_options &tcp_options={})
Definition websocket.h:119
crow::websocket::Connection::post
void post(CompletionHandler &&handler)
Send data through the socket and return immediately.
Definition websocket.h:223
sha1::SHA1
A tiny SHA1 algorithm implementation used internally in the Crow server (specifically in crow/websock...
Definition TinySHA1.hpp:48
crow
The main namespace of the library. In this namespace is defined the most important classes and functi...
crow::detail::socket::tcp_socket_options
Definition tcp_socket_options.h:29
crow::request
An HTTP request.
Definition http_request.h:47
crow::request::keep_alive
bool keep_alive
Whether or not the server should send a connection: Keep-Alive header to the client.
Definition http_request.h:57
crow::websocket::connection
A base class for websocket connection.
Definition websocket.h:68