// license:MIT // copyright-holders:Ole Christian Eidheim, Miodrag Milanovic /* Changes to the websocket client July 2022: (Michael Zapf, patch by Pete Eberlein [peberlein]) Symptom: During the initial handshake, the ws client reads the HTTP header from the server and then waits for more data to come. In cases where the websocket server sends data immediately after the handshake to the client, these data may already be available in the streambuf without triggering a new reception event, so the client may wait forever for a new reception. Correction: When going into the async_read, the client only waits for so many bytes as to satisfy the next read from the streambuf. For this end, the Message::needed_for method has been added. Also, leftover bytes are copied to the next message instance. */ #ifndef CLIENT_WS_HPP #define CLIENT_WS_HPP #include "asio.h" #include #include #include #include #include #include "crypto.hpp" #include "base64.hpp" #ifndef CASE_INSENSITIVE_EQUALS_AND_HASH #define CASE_INSENSITIVE_EQUALS_AND_HASH class case_insensitive_equals { public: bool operator()(const std::string &key1, const std::string &key2) const { return key1.size() == key2.size() && equal(key1.cbegin(), key1.cend(), key2.cbegin(), [](std::string::value_type key1v, std::string::value_type key2v) { return tolower(key1v) == tolower(key2v); }); } }; class case_insensitive_hash { public: size_t operator()(const std::string &key) const { size_t seed = 0; for (auto &c : key) { std::hash hasher; seed ^= hasher(std::tolower(c)) + 0x9e3779b9 + (seed << 6) + (seed >> 2); } return seed; } }; #endif namespace webpp { template class SocketClient; template class SocketClientBase { public: virtual ~SocketClientBase() { connection.reset(); } class SendStream : public std::iostream { friend class SocketClientBase; asio::streambuf streambuf; public: SendStream(): std::iostream(&streambuf) {} size_t size() const { return streambuf.size(); } }; class Connection { friend class SocketClientBase; friend class SocketClient; public: std::unordered_multimap header; std::string remote_endpoint_address; unsigned short remote_endpoint_port; private: explicit Connection(asio::io_context &context, socket_type* socket): remote_endpoint_port(0), socket(socket), strand(context), closed(false) { } class SendData { public: SendData(const std::shared_ptr &send_stream, const std::function &callback) : send_stream(send_stream), callback(callback) {} std::shared_ptr send_stream; std::function callback; }; std::unique_ptr socket; asio::io_context::strand strand; std::list send_queue; void send_from_queue() { strand.post([this]() { asio::async_write(*socket, send_queue.begin()->send_stream->streambuf, strand.wrap([this](const std::error_code& ec, size_t /*bytes_transferred*/) { auto send_queued=send_queue.begin(); if(send_queued->callback) send_queued->callback(ec); if(!ec) { send_queue.erase(send_queued); if(send_queue.size()>0) send_from_queue(); } else send_queue.clear(); })); }); } std::atomic closed; void read_remote_endpoint_data() { try { remote_endpoint_address=socket->lowest_layer().remote_endpoint().address().to_string(); remote_endpoint_port=socket->lowest_layer().remote_endpoint().port(); } catch(const std::exception& e) { std::cerr << e.what() << std::endl; } } }; std::shared_ptr connection; class Message : public std::istream { friend class SocketClientBase; public: unsigned char fin_rsv_opcode; size_t size() const { return length; } std::string string() { if (s.size() != length) { asio::streambuf::const_buffers_type bufs = streambuf.data(); s.assign(asio::buffers_begin(bufs), asio::buffers_begin(bufs)+length); streambuf.consume(length); } return s; } // Return the number of bytes that are missing in the streambuf // before n bytes can be read from it size_t needed_for(size_t n) const { if (streambuf.size() >= n) return 0; return n - streambuf.size(); } private: Message(): std::istream(&streambuf), fin_rsv_opcode(0), length(0) { } size_t length; asio::streambuf streambuf; std::string s; }; std::function on_open; std::function)> on_message; std::function on_close; std::function on_error; void start() { if(!io_context) { io_context=std::make_shared(); internal_io_context=true; } if(io_context->stopped()) io_context->reset(); if(!resolver) resolver= std::make_unique(*io_context); connect(); if(internal_io_context) io_context->run(); } void stop() const { resolver->cancel(); if(internal_io_context) io_context->stop(); } ///fin_rsv_opcode: 129=one fragment, text, 130=one fragment, binary, 136=close connection. ///See http://tools.ietf.org/html/rfc6455#section-5.2 for more information void send(const std::shared_ptr &message_stream, const std::function& callback=nullptr, unsigned char fin_rsv_opcode=129) { //Create mask std::vector mask; mask.resize(4); std::uniform_int_distribution dist(0,255); std::random_device rd; for(int c=0;c<4;c++) { mask[c]=static_cast(dist(rd)); } auto send_stream = std::make_shared(); size_t length=message_stream->size(); send_stream->put(fin_rsv_opcode); //masked (first length byte>=128) if(length>=126) { int num_bytes; if(length>0xffff) { num_bytes=8; send_stream->put(static_cast(127+128)); } else { num_bytes=2; send_stream->put(static_cast(126+128)); } for(int c=num_bytes-1;c>=0;c--) { send_stream->put((static_cast(length) >> (8 * c)) % 256); } } else send_stream->put(static_cast(length+128)); for(int c=0;c<4;c++) { send_stream->put(mask[c]); } for(size_t c=0;cput(message_stream->get()^mask[c%4]); } connection->strand.post([this, send_stream, callback]() { connection->send_queue.emplace_back(send_stream, callback); if(connection->send_queue.size()==1) connection->send_from_queue(); }); } void send_close(int status, const std::string& reason="", const std::function& callback=nullptr) { //Send close only once (in case close is initiated by client) if(connection->closed) return; connection->closed=true; auto send_stream=std::make_shared(); send_stream->put(status>>8); send_stream->put(status%256); *send_stream << reason; //fin_rsv_opcode=136: message close send(send_stream, callback, 136); } /// If you have your own asio::io_context, store its pointer here before running start(). std::shared_ptr io_context; protected: const std::string ws_magic_string="258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; bool internal_io_context=false; std::unique_ptr resolver; std::string host; unsigned short port; std::string path; SocketClientBase(const std::string& host_port_path, unsigned short default_port) { size_t host_end=host_port_path.find(':'); size_t host_port_end=host_port_path.find('/'); if(host_end==std::string::npos) { host_end=host_port_end; port=default_port; } else { if(host_port_end==std::string::npos) port=static_cast(stoul(host_port_path.substr(host_end + 1))); else port=static_cast(stoul(host_port_path.substr(host_end + 1, host_port_end - (host_end + 1)))); } if(host_port_end==std::string::npos) path="/"; else path=host_port_path.substr(host_port_end); if(host_end==std::string::npos) host=host_port_path; else host=host_port_path.substr(0, host_end); } virtual void connect()=0; void handshake() { connection->read_remote_endpoint_data(); auto write_buffer = std::make_shared(); std::ostream request(write_buffer.get()); request << "GET " << path << " HTTP/1.1" << "\r\n"; request << "Host: " << host << "\r\n"; request << "Upgrade: websocket\r\n"; request << "Connection: Upgrade\r\n"; //Make random 16-byte nonce std::string nonce; nonce.resize(16); std::uniform_int_distribution dist(0,255); std::random_device rd; for(int c=0;c<16;c++) nonce[c]=static_cast(dist(rd)); auto nonce_base64 = std::make_shared(base64_encode(nonce)); request << "Sec-WebSocket-Key: " << *nonce_base64 << "\r\n"; request << "Sec-WebSocket-Version: 13\r\n"; request << "\r\n"; asio::async_write(*connection->socket, *write_buffer, [this, write_buffer, nonce_base64] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { std::shared_ptr message(new Message()); asio::async_read_until(*connection->socket, message->streambuf, "\r\n\r\n", [this, message, nonce_base64] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { parse_handshake(*message); auto header_it = connection->header.find("Sec-WebSocket-Accept"); if (header_it != connection->header.end() && base64_decode(header_it->second) == sha1_encode(*nonce_base64 + ws_magic_string)) { if(on_open) on_open(); read_message(message); } else if(on_error) on_error(std::error_code(int(std::errc::protocol_error), std::generic_category())); } else if(on_error) on_error(ec); }); } else if(on_error) on_error(ec); }); } void parse_handshake(std::istream& stream) const { std::string line; getline(stream, line); //Not parsing the first line getline(stream, line); size_t param_end; while((param_end=line.find(':'))!=std::string::npos) { size_t value_start=param_end+1; if((value_start)header.emplace(line.substr(0, param_end), line.substr(value_start, line.size()-value_start-1)); } getline(stream, line); } } void read_message(const std::shared_ptr &message) { asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(2)), [this, message](const std::error_code& ec, size_t bytes_transferred) { if(!ec) { if(message->streambuf.size()+bytes_transferred==0) { //TODO: This might happen on server at least, might also happen here read_message(message); return; } std::vector first_bytes; first_bytes.resize(2); message->read(reinterpret_cast(&first_bytes[0]), 2); message->fin_rsv_opcode=first_bytes[0]; //Close connection if masked message from server (protocol error) if(first_bytes[1]>=128) { const std::string reason("message from server masked"); auto kept_connection=connection; send_close(1002, reason, [](const std::error_code& /*ec*/) {}); if(on_close) on_close(1002, reason); return; } size_t length=(first_bytes[1]&127); if(length==126) { //2 next bytes is the size of content asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(2)), [this, message] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { std::vector length_bytes; length_bytes.resize(2); message->read(reinterpret_cast(&length_bytes[0]), 2); size_t length=0; int num_bytes=2; for(int c=0;clength=length; read_message_content(message); } else if(on_error) on_error(ec); }); } else if(length==127) { //8 next bytes is the size of content asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(8)), [this, message] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { std::vector length_bytes; length_bytes.resize(8); message->read(reinterpret_cast(&length_bytes[0]), 8); size_t length=0; int num_bytes=8; for(int c=0;clength=length; read_message_content(message); } else if(on_error) on_error(ec); }); } else { message->length=length; read_message_content(message); } } else if(on_error) on_error(ec); }); } void read_message_content(const std::shared_ptr &message) { asio::async_read(*connection->socket, message->streambuf, asio::transfer_exactly(message->needed_for(message->length)), [this, message] (const std::error_code& ec, size_t /*bytes_transferred*/) { if(!ec) { //If connection close if((message->fin_rsv_opcode&0x0f)==8) { int status=0; if(message->length>=2) { unsigned char byte1=message->get(); unsigned char byte2=message->get(); status=(byte1<<8)+byte2; } auto reason=message->string(); auto kept_connection=connection; send_close(status, reason, [](const std::error_code& /*ec*/) {}); if(on_close) on_close(status, reason); return; } //If ping else if((message->fin_rsv_opcode&0x0f)==9) { //send pong auto empty_send_stream=std::make_shared(); send(empty_send_stream, nullptr, message->fin_rsv_opcode+1); } else if(on_message) { on_message(message); } //Next message std::shared_ptr next_message(new Message()); //Copy any leftover data in the streambuf into the next message next_message->streambuf.commit(buffer_copy( next_message->streambuf.prepare(message->streambuf.size()), message->streambuf.data())); read_message(next_message); } else if(on_error) on_error(ec); }); } }; template class SocketClient : public SocketClientBase { public: SocketClient(const std::string& host_port_path, unsigned short default_port) : SocketClientBase(host_port_path, default_port) { } }; using WS = asio::ip::tcp::socket; template<> class SocketClient : public SocketClientBase { public: explicit SocketClient(const std::string& server_port_path) : SocketClientBase(server_port_path, 80) {}; protected: void connect() override { asio::ip::tcp::resolver::query query(host, std::to_string(port)); resolver->async_resolve(query, [this] (const std::error_code &ec, asio::ip::tcp::resolver::iterator it){ if(!ec) { connection=std::shared_ptr(new Connection(*io_context, new WS(*io_context))); asio::async_connect(*connection->socket, it, [this] (const std::error_code &ec, asio::ip::tcp::resolver::iterator /*it*/){ if(!ec) { asio::ip::tcp::no_delay option(true); connection->socket->set_option(option); handshake(); } else if(on_error) on_error(ec); }); } else if(on_error) on_error(ec); }); } }; class ws_client : public SocketClient { public: explicit ws_client(const std::string& server_port_path) : SocketClient::SocketClient(server_port_path) {} }; } #endif /* CLIENT_WS_HPP */