Line data Source code
1 : #ifndef THINGER_ASIO_WEBSOCKET_HPP
2 : #define THINGER_ASIO_WEBSOCKET_HPP
3 : #pragma once
4 :
5 : #include "socket.hpp"
6 : #include "../async_event.hpp"
7 : #include "../watchdog.hpp"
8 : #include "../../util/close_notifier.hpp"
9 : #include <boost/asio/any_io_executor.hpp>
10 : #include <boost/asio/steady_timer.hpp>
11 : #include <chrono>
12 : #include <deque>
13 : #include <memory>
14 : #include <optional>
15 : #include <string>
16 : #include <string_view>
17 : #include <vector>
18 :
19 : namespace thinger::asio {
20 :
21 : // How a WebSocket closed (see websocket::on_close())
22 : struct websocket_close {
23 : // Status code of the closing handshake (RFC 6455, 7.4), as websocket::get_close_code():
24 : // the one of the peer's close frame (1005 if it had none, 1002 if it was not valid), else
25 : // the one this end closed with; 1006 (websocket::CLOSE_ABNORMAL) if no close frame was
26 : // exchanged, e.g. when the connection was lost, or the one of close_graceful() was not
27 : // written whole by its deadline
28 : uint16_t code = 1006;
29 : // Reason of the peer's close frame, if any (not validated as UTF-8), never the one this end
30 : // sent
31 : std::string reason;
32 : // Whether the peer closed first (sent its close frame before this end did)
33 : bool remote = false;
34 : };
35 :
36 : // WebSocket (RFC 6455) over another socket, once the HTTP handshake is done (see
37 : // http::response::upgrade_websocket_stream() and the client websocket()).
38 : //
39 : // As a socket it is a byte stream, like TCP, TLS or Unix sockets, so a protocol running on it
40 : // (or a socket_pipe) cannot tell it runs over WebSocket:
41 : // - Reads return the payload of the data messages received, text or binary, as one continuous
42 : // stream: messages, and the frames of a fragmented message, are not delimited. read_some()
43 : // returns data from one frame at most, and never 0 bytes (empty frames are skipped).
44 : // - Each write sends one message (unfragmented), binary by default (see set_binary()). Writes
45 : // may be called while another one is pending, from several coroutines: they are sent one
46 : // after another, in call order. Writing 0 bytes sends nothing (write_message() sends an
47 : // empty message). A write interrupted in the
48 : // middle of its frame (cancel(), or the cancellation of the operation) leaves the rest of
49 : // the frame to be sent before anything else, so the framing is never broken.
50 : // - Pings are answered, pongs and unexpected control frames handled, while reading: some
51 : // read must be pending for them to be answered (as with a protocol reading its socket). The
52 : // pong is sent as soon as no frame is being written, without holding the reads.
53 : // - A close frame received is answered, the connection closed, and reads end with
54 : // boost::asio::error::eof (as a TCP socket closed by its peer). Writes then fail.
55 : // - close(), on the executor of the websocket, ends the pending operations and closes the
56 : // socket for its user right away; the closing handshake goes on by itself: the rest of a
57 : // frame being written, a close frame (code 1000), and the connection closed once the peer
58 : // answers or after CLOSE_TIMEOUT. Called anywhere else (e.g. by a synchronous client, while
59 : // nothing runs its io_context), it closes the connection at once instead. close(code) does
60 : // the same with another status code. close_graceful() does the handshake, waiting, with a
61 : // status code and a reason if given, within CLOSE_TIMEOUT whatever the peer does. A
62 : // websocket destroyed open on its executor closes with 1001. An executor that is neither an
63 : // io_context executor nor a strand cannot be told to be running (see
64 : // asio::running_in_this_thread()): close() then closes at once, without the handshake.
65 : // - An optional keepalive pings the peer (one ping at a time) and closes the connection
66 : // when it is idle (see set_keepalive()).
67 : // - on_close() tells once the connection closed, whatever closed it.
68 : // - Text is not validated as UTF-8 here, neither in text messages nor in close reasons:
69 : // the stream may cut a character anywhere (http::websocket_connection validates whole
70 : // messages).
71 : //
72 : // The message interface (read_message_some(), is_message_complete(), is_binary()) is for
73 : // message-oriented users, such as http::websocket_connection and http::websocket_client.
74 : //
75 : // It must be owned by a std::shared_ptr for the keepalive and to answer pings without
76 : // holding the reads. Not thread-safe, as any socket (see socket.hpp): used on its executor
77 : // (see get_executor()), where it also runs the work it does on its own (the keepalive, the
78 : // control frames it sends, the closing handshake), so never at the same time as its user.
79 : class websocket : public socket, public std::enable_shared_from_this<websocket> {
80 :
81 : public:
82 :
83 : static constexpr auto CONNECTION_TIMEOUT_SECONDS = std::chrono::seconds{60};
84 : // What the closing handshake may take, from the call that starts it to the socket closed
85 : static constexpr auto CLOSE_TIMEOUT = std::chrono::seconds{5};
86 : static constexpr int MASK_SIZE_BYTES = 4;
87 : static std::atomic<unsigned long> connections;
88 :
89 : // Close status codes (RFC 6455, 7.4.1)
90 : static constexpr uint16_t CLOSE_NORMAL = 1000;
91 : static constexpr uint16_t CLOSE_GOING_AWAY = 1001;
92 : static constexpr uint16_t CLOSE_PROTOCOL_ERROR = 1002;
93 : // Never sent: a close frame without a status code, and a connection closed without close frames
94 : static constexpr uint16_t CLOSE_NO_STATUS = 1005;
95 : static constexpr uint16_t CLOSE_ABNORMAL = 1006;
96 : static constexpr uint16_t CLOSE_INVALID_PAYLOAD = 1007;
97 : static constexpr uint16_t CLOSE_MESSAGE_TOO_BIG = 1009;
98 : static constexpr uint16_t CLOSE_INTERNAL_ERROR = 1011;
99 : // IANA registry: the server sheds a client (e.g. one that does not keep up); it may retry
100 : static constexpr uint16_t CLOSE_TRY_AGAIN_LATER = 1013;
101 :
102 : // Whether a close frame may carry `code` (RFC 6455, 7.4, and the IANA registry up to 1014, as
103 : // Node ws and Go accept): 1000-1014 but 1004, 1005 and 1006, and 3000-4999
104 : static bool valid_close_code(uint16_t code);
105 : // Largest reason of a close frame, in bytes: a control frame carries 125 (RFC 6455, 5.5.1)
106 : static constexpr size_t MAX_CLOSE_REASON_SIZE = 123;
107 :
108 : // `binary`: frames written are binary (otherwise text); `server`: the server role (frames
109 : // received must be masked, frames sent are not); `executor`: where its user runs it (e.g. a
110 : // strand of an io_context run by several threads), the io_context of the socket by default
111 : websocket(std::shared_ptr<socket> socket, bool binary = true, bool server = true,
112 : boost::asio::any_io_executor executor = {});
113 : ~websocket() override;
114 :
115 : // Socket control
116 : awaitable<boost::system::error_code> connect(
117 : const std::string &host,
118 : const std::string &port,
119 : std::chrono::seconds timeout) override;
120 : void close() override;
121 : // Close the connection with `code`: the closing handshake on its own when called on the
122 : // executor, waiting for the close frame of the peer if `expect_reply`. On the executor, as
123 : // any operation. A code that may not be sent (RFC 6455, 7.4: e.g. 1005, 1006 or 1015) is
124 : // sent as 1000
125 : void close(uint16_t code, bool expect_reply = true);
126 : // Send a close frame with `code` and `reason` (waiting for a pending write), wait for the
127 : // close frame of the peer, and close; the keepalive stops at once. All of it within
128 : // CLOSE_TIMEOUT of the first call, its wait for the turn to write included: at the deadline
129 : // the socket is closed without the handshake, whatever is pending on it, and on_close() told
130 : // (see get_close_code()). While that frame waits for its turn, it is what any other close
131 : // sends, and another close_graceful() only waits for the same handshake. A code that may not
132 : // be sent is sent as 1000, a reason above MAX_CLOSE_REASON_SIZE is left out; the reason is
133 : // not validated as UTF-8 here
134 : awaitable<void> close_graceful(uint16_t code = CLOSE_NORMAL, std::string reason = {});
135 : // The pending read and writes end with boost::asio::error::operation_aborted; the
136 : // websocket stays usable. A read that received data meanwhile may return it instead, but
137 : // never reads on: e.g. after a control frame received just before
138 : void cancel() override;
139 : bool requires_handshake() const override;
140 : awaitable<boost::system::error_code> handshake(const std::string& host = "") override;
141 :
142 : // Read operations: read(), read(streambuf) and read_until() are built on read_some()
143 : awaitable<io_result> read_some(uint8_t buffer[], size_t max_size) override;
144 :
145 : // Write operations: one message each
146 : awaitable<io_result> write(const uint8_t buffer[], size_t size) override;
147 : awaitable<io_result> write(std::string_view str) override;
148 : awaitable<io_result> write(const std::vector<boost::asio::const_buffer> &buffers) override;
149 : // One message of the type set with set_binary(), as write(), but an empty one is sent too
150 : // (a frame without payload), for message-oriented users
151 : awaitable<io_result> write_message(std::string_view data);
152 : // The same, a frame larger than WRITE_SLICE_SIZE written in slices (still one frame),
153 : // `progress` called after each one (see socket::write_slices()), e.g. to move a write deadline
154 : awaitable<io_result> write_message(std::string_view data, const sync_callback<void()>& progress);
155 :
156 : // Wait
157 : awaitable<boost::system::error_code> wait(boost::asio::socket_base::wait_type type) override;
158 :
159 : // Message interface: up to max_size bytes of the frame being received, which may be
160 : // empty (0 bytes for an empty frame, without error). The message is complete when
161 : // is_message_complete() and remaining_in_frame() is 0; is_binary() tells its type.
162 : awaitable<io_result> read_message_some(uint8_t buffer[], size_t max_size);
163 : uint64_t remaining_in_frame() const;
164 : bool is_message_complete() const;
165 9991 : bool is_binary() const { return message_opcode_ == 0x2; }
166 :
167 : // Control frames, sent before the writes waiting (after the one in progress), waiting
168 : // until sent
169 : awaitable<void> send_ping(uint8_t buffer[] = nullptr, size_t size = 0);
170 : awaitable<void> send_pong(uint8_t buffer[] = nullptr, size_t size = 0);
171 :
172 : // Closing handshake: the status code of the close frame received from the peer
173 : // (CLOSE_NO_STATUS if it had none, CLOSE_PROTOCOL_ERROR if it was not valid), else of the
174 : // one this end sent; CLOSE_ABNORMAL if neither was, e.g. when the connection was lost, or
175 : // the frame of close_graceful() was not written whole by its deadline
176 : uint16_t get_close_code() const;
177 : // The reason of the close frame received, if any (not validated as UTF-8)
178 693 : const std::string& get_close_reason() const { return close_reason_; }
179 : // Whether the peer started the closing handshake (its close frame came first)
180 693 : bool closed_by_peer() const { return closed_by_peer_; }
181 :
182 : // Called once the connection closed, with how (see websocket_close): close() (by its user,
183 : // the keepalive, a protocol error, or the close frame of the peer answered), the deadline of
184 : // close_graceful(), or the connection lost (an error reading or writing other than a
185 : // cancellation, the end of the stream without a close frame, or the socket closed under it,
186 : // e.g. by a server at the end of its shutdown): then with 1006 unless a close frame was
187 : // sent or received. Once, posted to the
188 : // executor (never from within the call that closed it), what it throws logged; set once
189 : // closed, it is posted at once. A websocket destroyed open does not call it. The connection
190 : // is noticed lost by the operations running on it (a pending read, the keepalive). Unlike
191 : // http::websocket_connection, nothing breaks a cycle here: the callback must not own the
192 : // websocket (capture a std::weak_ptr). On the executor
193 : void on_close(sync_callback<void(const websocket_close&)> callback);
194 :
195 : // State getters
196 : bool is_open() const override;
197 : bool is_secure() const override;
198 : size_t available() const override;
199 : // The one of the socket it runs on
200 : size_t send_backlog() const override;
201 : std::string get_remote_ip() const override;
202 : std::string get_local_port() const override;
203 : std::string get_remote_port() const override;
204 :
205 : // Type of the messages written: binary (true) or text
206 27462 : void set_binary(bool binary) { binary_ = binary; }
207 : bool get_binary() const { return binary_; }
208 :
209 : // Subprotocol negotiated in the handshake (Sec-WebSocket-Protocol), if any
210 63 : const std::string& get_protocol() const { return protocol_; }
211 3805 : void set_protocol(std::string protocol) { protocol_ = std::move(protocol); }
212 :
213 : // Socket the WebSocket runs on
214 : std::shared_ptr<socket> get_socket() const { return socket_; }
215 : // The TLS details of the socket it runs on
216 : std::optional<tls_session> tls() const override;
217 :
218 : // Executor the WebSocket is used on, and runs its own work on: the one it was created with
219 4852847 : const boost::asio::any_io_executor& get_executor() const { return executor_; }
220 :
221 : // Keepalive: ping the peer once nothing was received from it for `ping_interval`, and
222 : // close the connection (1001) once nothing was received for `idle_timeout` (pongs
223 : // included). 0 disables each one; calling it again replaces the previous settings, and
224 : // counts the idle time from then; it does not start once close_graceful() was called.
225 : // Requires the websocket to be owned by a std::shared_ptr.
226 : // In the client role, for a user that does not read all the time (e.g. a synchronous
227 : // client between its calls): bytes that reached the socket while nothing read count as
228 : // received, once its check sees them; and, with a ping interval under the idle timeout, a
229 : // check that runs late (nothing ran the executor) pings first, and closes only once that
230 : // ping went unanswered for idle_timeout - ping_interval. Run on time, the instants are the
231 : // same; in the server role nothing of this applies
232 : void set_keepalive(std::chrono::milliseconds ping_interval, std::chrono::milliseconds idle_timeout);
233 90 : std::chrono::milliseconds get_ping_interval() const { return ping_interval_; }
234 90 : std::chrono::milliseconds get_idle_timeout() const { return idle_timeout_; }
235 :
236 : // Keepalive of the message interface: ping after CONNECTION_TIMEOUT_SECONDS without
237 : // receiving anything, close after twice that
238 : void start_timeout();
239 :
240 : // Bytes already read from the socket, read before anything else from it (e.g. frames
241 : // received along with the handshake)
242 : void set_read_ahead(std::string data);
243 :
244 : private:
245 : struct frame_header {
246 : bool fin = false;
247 : uint8_t rsv = 0;
248 : uint8_t opcode = 0;
249 : bool masked = false;
250 : uint64_t payload_size = 0;
251 : uint8_t mask[MASK_SIZE_BYTES]{};
252 : size_t size = 0; // bytes of the header
253 : };
254 :
255 : // Parse a frame header from `data`: false if more bytes are needed
256 : static bool parse_header(const uint8_t* data, size_t size, frame_header& header);
257 : // Header of a frame sent (with its mask key in the client role); its size
258 : size_t build_header(uint8_t opcode, uint64_t payload_size, uint8_t header[14], uint8_t mask[MASK_SIZE_BYTES]) const;
259 :
260 : // Read side
261 507397 : size_t buffered() const { return in_end_ - in_begin_; }
262 : // Wait until the input buffer holds `size` bytes
263 : awaitable<boost::system::error_code> fill(size_t size);
264 : // Read up to max_size bytes of the current data frame payload (at least one)
265 : awaitable<io_result> read_payload(uint8_t buffer[], size_t max_size);
266 : // Handle the control frame whose header and payload are buffered
267 : awaitable<boost::system::error_code> handle_control(const frame_header& header);
268 : awaitable<io_result> read_frame_data(uint8_t buffer[], size_t max_size, bool skip_empty);
269 : void unmask(uint8_t buffer[], size_t size);
270 : // XOR `data` with the mask key, starting at byte `offset` of the key
271 : static void apply_mask(uint8_t* data, size_t size, const uint8_t mask[MASK_SIZE_BYTES], size_t offset);
272 : // Fail the connection on a protocol error: close frame (1002, best effort) and close
273 : boost::system::error_code fail(const char* reason);
274 : // How it closed, for on_close()
275 : websocket_close close_status() const;
276 : // An operation on the socket ended with `ec`: the connection is lost if it is an error
277 : // other than a cancellation, or the socket was closed under it (told to on_close())
278 : void check_lost(const boost::system::error_code& ec);
279 :
280 : // Write side: one frame at a time (the writer holding the turn), in order: the rest of an
281 : // interrupted frame first, then the pong and ping queued, then the writes waiting
282 : awaitable<bool> acquire_write(bool control);
283 : // Pass the turn on: to the closing handshake once closed, to the control frames queued,
284 : // or to the next writer waiting
285 : void release_write();
286 : void wake_writers();
287 : awaitable<io_result> send_frame(uint8_t opcode, const std::vector<boost::asio::const_buffer>& payload,
288 : bool control, const sync_callback<void()>* progress = nullptr);
289 : // Write a frame, holding the turn: the rest of an interrupted frame first, and what this
290 : // one leaves if interrupted. In slices with a `progress` (see write_message())
291 : awaitable<boost::system::error_code> write_frame(uint8_t opcode, const std::vector<boost::asio::const_buffer>& payload,
292 : const sync_callback<void()>* progress = nullptr);
293 : // The close frame close_graceful() asked for, waiting for the turn
294 : awaitable<boost::system::error_code> send_close();
295 : // Payload of the close frame sent: its code and reason
296 : std::vector<uint8_t> close_payload() const;
297 : // Queue a pong (answering the last ping) or a ping, sent without waiting for it
298 : void queue_control();
299 : // Send the control frames queued, holding the turn
300 : awaitable<void> flush_control();
301 : // Close the connection when idle for too long, or queue a ping when due, and arm the
302 : // keepalive for the next check
303 : void check_keepalive();
304 :
305 : // Closed for its user: pending operations told, timers and control frames dropped
306 : void mark_closed();
307 : // The handshake of close_graceful() ends within CLOSE_TIMEOUT: at the deadline the socket is
308 : // closed without it, whatever is pending on it
309 : void start_close_deadline();
310 : void close_expired();
311 :
312 : void start_linger();
313 : // The rest of a frame and a close frame, the bytes still buffered and the payload
314 : // remaining of the frame being received: what the closing handshake needs once this
315 : // websocket is closed, and when it ends at the latest
316 : static awaitable<void> linger(std::shared_ptr<socket> socket, std::vector<uint8_t> output,
317 : std::string input, uint64_t skip, bool expect_reply,
318 : std::chrono::steady_clock::time_point deadline);
319 :
320 : std::shared_ptr<socket> socket_;
321 : boost::asio::any_io_executor executor_;
322 : bool binary_;
323 : bool server_role_;
324 : std::string protocol_;
325 :
326 : // Input buffer: bytes read from the socket not processed yet
327 : std::vector<uint8_t> in_;
328 : size_t in_begin_ = 0;
329 : size_t in_end_ = 0;
330 :
331 : // Frame being received
332 : bool new_message_ = true;
333 : uint8_t message_opcode_ = 0;
334 : uint64_t frame_remaining_ = 0;
335 : bool frame_fin_ = false;
336 : bool frame_masked_ = false;
337 : uint8_t mask_[MASK_SIZE_BYTES]{};
338 : size_t mask_offset_ = 0;
339 : bool reading_ = false;
340 : // cancel() called during the pending read: it ends before reading from the socket again.
341 : // Cancelling the socket is not enough: a read of it that completed already (its handler
342 : // waiting to run on the executor) is not cancelled, and a control frame received then
343 : // is handled and the socket read again
344 : bool read_cancelled_ = false;
345 :
346 : // Writes waiting for their turn
347 : struct write_waiter {
348 868 : explicit write_waiter(const boost::asio::any_io_executor& executor) : turn(executor) {}
349 : // Set when the turn is granted, or the writers are woken (closed, cancelled)
350 : async_event turn;
351 : bool control = false;
352 : bool granted = false;
353 : };
354 : bool writing_ = false;
355 : std::deque<std::shared_ptr<write_waiter>> write_waiters_;
356 : // Rest of a frame whose write was interrupted, sent before anything else
357 : std::vector<uint8_t> tail_;
358 : // Error that broke the connection while writing
359 : boost::system::error_code write_error_;
360 : // Control frames queued: the payload of the last ping received, and a keepalive ping
361 : std::optional<std::string> pending_pong_;
362 : bool pending_ping_ = false;
363 : bool ping_in_flight_ = false;
364 :
365 : // Connection state
366 : bool closed_ = false;
367 : bool close_received_ = false;
368 : bool close_sent_ = false;
369 : // The close frame being written by close_graceful() is not written whole (yet): it counts as
370 : // not sent for get_close_code()
371 : bool close_unwritten_ = false;
372 : // Closing handshake to start once the write in progress ends
373 : bool linger_pending_ = false;
374 : bool linger_expect_reply_ = true;
375 : uint16_t close_code_ = CLOSE_NORMAL;
376 : // Asked for by close_graceful(): that code and this reason are the ones of the close frame
377 : bool close_requested_ = false;
378 : std::string close_reason_sent_;
379 : // Close frame received: its code and reason, and whether it came first
380 : uint16_t received_close_code_ = CLOSE_NO_STATUS;
381 : std::string close_reason_;
382 : bool closed_by_peer_ = false;
383 : std::shared_ptr<boost::asio::steady_timer> close_timer_;
384 : util::close_notifier<websocket_close> on_close_;
385 :
386 : // Keepalive
387 : watchdog keepalive_;
388 3919 : std::chrono::milliseconds ping_interval_{0};
389 3919 : std::chrono::milliseconds idle_timeout_{0};
390 : std::chrono::steady_clock::time_point last_received_;
391 : std::chrono::steady_clock::time_point last_ping_;
392 : // In the client role: the first ping sent since the last byte received, and the bytes
393 : // available (unread) at the previous check
394 : std::chrono::steady_clock::time_point first_ping_;
395 : size_t available_seen_ = 0;
396 : };
397 :
398 : }
399 :
400 : #endif
|
