thomas.goyne_275

Committed 09 Apr 2024 03:33AM UTC coverage: 92.608% (+0.5%) from 92.088%

Build # thomas.goyne_275

Build Type

Pull #7300

Evergreen

Committed by

tgoyne

Commit Message

Extract some duplicated code in PushClient

Pull Request Pull Request #7300: Rework sync user handling and metadata storage

Run Details

102672 of 194970 branches covered (52.66%)

3165 of 3247 new or added lines in 46 files covered. (97.47%)

34 existing lines in 9 files now uncovered.

249420 of 269329 relevant lines covered (92.61%)

45087511.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.01

/src/realm/sync/network/websocket.cpp

#include <cctype>

#include <realm/sync/network/network.hpp>
#include <realm/sync/network/websocket.hpp>
#include <realm/util/buffer.hpp>
#include <realm/util/base64.hpp>
#include <realm/util/sha_crypto.hpp>

using namespace realm;
using namespace realm::sync;
using HttpError = websocket::HttpError;
using WebSocketError = websocket::WebSocketError;


namespace {

// case_insensitive_equal is used to compare, ignoring case, certain HTTP
// header values with their expected values.
bool case_insensitive_equal(StringData str1, StringData str2)
{
    if (str1.size() != str2.size())
        return false;

    for (size_t i = 0; i < str1.size(); ++i)
        if (std::tolower(str1[i], std::locale::classic()) != std::tolower(str2[i], std::locale::classic()))
            return false;

    return true;
}

// The WebSocket version is always 13 according to the standard.
const StringData sec_websocket_version = "13";

// The magic string is specified in the WebSocket protocol. It is used in the handshake.
const StringData websocket_magic_string = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";

// The Sec-WebSocket-Key is a header in the client HTTP request.
// It is the base64 encoding of 16 random bytes.
std::string make_random_sec_websocket_key(std::mt19937_64& random)
{
    char random_bytes[16];
    std::uniform_int_distribution<> dis(std::numeric_limits<char>::min(), std::numeric_limits<char>::max());
    for (int i = 0; i < 16; ++i) {
        random_bytes[i] = dis(random);
    }

    char out_buffer[24];
    size_t encoded_size = util::base64_encode(random_bytes, out_buffer);
    REALM_ASSERT(encoded_size == 24);

    return std::string{out_buffer, 24};
}

// Sec-WebSocket-Accept is a header in the server's HTTP response.
// It is calculated from the Sec-WebSocket-Key in the client's HHTP request
// as the base64 encoding of the sha1 of the concatenation of the
// Sec-Websocket-Key and the magic string.
std::string make_sec_websocket_accept(StringData sec_websocket_key)
{
    std::string sha1_input;
    sha1_input.reserve(sec_websocket_key.size() + websocket_magic_string.size());
    sha1_input.append(sec_websocket_key.data(), sec_websocket_key.size());
    sha1_input.append(websocket_magic_string.data(), websocket_magic_string.size());

    char sha1_output[20];
    util::sha1(sha1_input.data(), sha1_input.length(), reinterpret_cast<unsigned char*>(sha1_output));

    char base64_output[28];
    size_t base64_output_size = util::base64_encode(sha1_output, base64_output);
    REALM_ASSERT(base64_output_size == 28);

    return std::string(base64_output, 28);
}

// find_http_header_value() returns the value of the header \param header in the
// HTTP message \param message if present. If the header is absent, the function returns
// None.
util::Optional<StringData> find_http_header_value(const HTTPHeaders& headers, StringData header)
{
    auto it = headers.find(header);

    if (it != headers.end()) {
        return StringData(it->second);
    }
    return none;
}

// validate_websocket_upgrade() returns true if the HTTP \a headers
// contain the line
// Upgrade: websocket
bool validate_websocket_upgrade(const HTTPHeaders& headers)
{
    util::Optional<StringData> header_value_upgrade = find_http_header_value(headers, "Upgrade");

    return (header_value_upgrade && case_insensitive_equal(*header_value_upgrade, "websocket"));
}

// validate_websocket_connection() returns true if the HTTP \a headers
// contains the line:
// Connection: Upgrade
bool validate_websocket_connection(const HTTPHeaders& headers)
{
    util::Optional<StringData> header_value_connection = find_http_header_value(headers, "Connection");

    return (header_value_connection && case_insensitive_equal(*header_value_connection, "Upgrade"));
}

// validate_sec_websocket_version() returns true if the
// \param http_message contains a header with the
// correct sec_websocket_version.
bool validate_sec_websocket_version(const HTTPHeaders& headers)
{
    return find_http_header_value(headers, "Sec-WebSocket-Version") == sec_websocket_version;
}

// find_sec_websocket_key() returns true if the
// \param headers contains a Sec-Websocket-Key
// header, false otherwise. If the header is found, the
// argument sec_websocket_key is set to its value.
bool find_sec_websocket_key(const HTTPHeaders& headers, std::string& sec_websocket_key)
{
    util::Optional<StringData> header_value = find_http_header_value(headers, "Sec-WebSocket-Key");

    if (!header_value)
        return false;

    sec_websocket_key = *header_value;

    return true;
}

util::Optional<HTTPResponse> do_make_http_response(const HTTPRequest& request,
                                                   const std::string& sec_websocket_protocol, std::error_code& ec)
{
    std::string sec_websocket_key;

    if (!validate_websocket_upgrade(request.headers)) {
        ec = HttpError::bad_request_header_upgrade;
        return util::none;
    }

    if (!validate_websocket_connection(request.headers)) {
        ec = HttpError::bad_request_header_connection;
        return util::none;
    }

    if (!validate_sec_websocket_version(request.headers)) {
        ec = HttpError::bad_request_header_websocket_version;
        return util::none;
    }

    if (!find_sec_websocket_key(request.headers, sec_websocket_key)) {
        ec = HttpError::bad_request_header_websocket_key;
        return util::none;
    }

    std::string sec_websocket_accept = make_sec_websocket_accept(sec_websocket_key);

    HTTPResponse response;
    response.status = HTTPStatus::SwitchingProtocols;
    response.headers["Upgrade"] = "websocket";
    response.headers["Connection"] = "Upgrade";
    response.headers["Sec-WebSocket-Accept"] = sec_websocket_accept;
    response.headers["Sec-WebSocket-Protocol"] = sec_websocket_protocol;

    return response;
}

// mask_payload masks (and demasks) the payload sent from the client to the server.
void mask_payload(char* masking_key, const char* payload, size_t payload_len, char* output)
{
    for (size_t i = 0; i < payload_len; ++i) {
        output[i] = payload[i] ^ masking_key[i % 4];
    }
}

// make_frame() creates a WebSocket frame according to the WebSocket standard.
// \param fin indicates whether the frame is the final fragment in a message.
// Sync clients and servers will only send unfragmented messages, but they must be
// prepared to receive fragmented messages.
// \param opcode must be one of six values:
// 0  = continuation frame
// 1  = text frame
// 2  = binary frame
// 8  = ping frame
// 9  = pong frame
// 10 = close frame.
// Sync clients and server will only send the last four, but must be prepared to
// receive all.
// \param mask indicates whether the payload of the frame should be masked. Frames
// are masked if and only if they originate from the client.
// The payload is located in the buffer \param payload, and has size \param payload_size.
// \param output is the output buffer. It must be large enough to contain the frame.
// The frame size can at most be payload_size + 14.
// \param random is used to create a random masking key.
// The return value is the size of the frame.
size_t make_frame(bool fin, int opcode, bool mask, const char* payload, size_t payload_size, char* output,
                  std::mt19937_64& random)
{
    int index = 0; // used to keep track of position within the header.
    using uchar = unsigned char;
    output[0] = (fin ? char(uchar(128)) : 0) + opcode; // fin and opcode in the first byte.
    output[1] = (mask ? char(uchar(128)) : 0);         // First bit of the second byte is mask.
    if (payload_size <= 125) {                         // The payload length is contained in the second byte.
        output[1] += static_cast<char>(payload_size);
        index = 2;
    }
    else if (payload_size < 65536) { // The payload length is contained bytes 3-4.
        output[1] += 126;
        // FIXME: Verify that this code works even if one sync-client is on a platform where
        // a 'char' is signed by default and the other client is on a platform where char is
        // unsigned. Note that the result of payload_size / 256 may not fit in a signed char.
        output[2] = static_cast<char>(payload_size / 256);

        // FIXME: Verify that the modulo arithmetic is well defined
        output[3] = payload_size % 256;
        index = 4;
    }
    else { // The payload length is contained in bytes 3-10.
        output[1] += 127;
        size_t fraction = payload_size;
        int remainder = 0;
        for (int i = 0; i < 8; ++i) {
            remainder = fraction % 256;
            fraction /= 256;
            output[9 - i] = remainder;
        }
        index = 10;
    }
    if (mask) {
        char masking_key[4];
        std::uniform_int_distribution<> dis(0, 255);
        for (int i = 0; i < 4; ++i) {
            masking_key[i] = dis(random);
        }
        output[index++] = masking_key[0];
        output[index++] = masking_key[1];
        output[index++] = masking_key[2];
        output[index++] = masking_key[3];
        mask_payload(masking_key, payload, payload_size, output + index);
    }
    else {
        std::copy(payload, payload + payload_size, output + index);
    }

    return payload_size + index;
}

// class FrameReader takes care of parsing the incoming bytes and
// constructing the received WebSocket messages. FrameReader manages
// read buffers internally. FrameReader handles fragmented messages as
// well. FrameRader is used in the following way:
//
// After constructing a FrameReader, the user runs in a loop.
//
// Loop start:
//
// Call frame_reader.next().
// if (frame_reader.protocol_error) {
//      // report a protocol error and
//      // break out of this loop
// }
// else if (frame_reader.delivery_ready) {
//     // use the message
//     // in frame_reader.delivery_buffer
//     // of size
//     // frame_reader.delivery_size
//     // with opcode (type)
//     // frame_reader.delivery_opcode
// }
// else {
//    // read frame_reader.read_size
//    // bytes into the buffer
//    // frame_reader.read_buffer
// }
// Goto Loop start.
//
class FrameReader {
public:
    // FrameReader is owned by a websocket, so a shared_ptr is not needed
    util::Logger& logger;

    char* delivery_buffer = nullptr;
    size_t delivery_size = 0;
    size_t read_size = 0;
    char* read_buffer = nullptr;
    bool protocol_error = false;
    bool delivery_ready = false;
    websocket::Opcode delivery_opcode = websocket::Opcode::continuation;

    FrameReader(util::Logger& logger, bool& is_client)
        : logger(logger)
        , m_is_client(is_client)
    {
    }

    // reset() resets the frame reader such that it is ready to work on
    // a new WebSocket connection.
    void reset()
    {
        m_stage = Stage::init;
    }

    // next() parses the new information and moves
    // one stage forward.
    void next()
    {
        switch (m_stage) {
            case Stage::init:
                stage_init();
                break;
            case Stage::header_beginning:
                stage_header_beginning();
                break;
            case Stage::header_end:
                stage_header_end();
                break;
            case Stage::payload:
                stage_payload();
                break;
            case Stage::delivery:
                stage_delivery();
                break;
            default:
                break;
        }
    }

private:
    bool& m_is_client;

    char header_buffer[14];
    char* m_masking_key;
    size_t m_payload_size;
    websocket::Opcode m_opcode = websocket::Opcode::continuation;
    bool m_fin = false;
    bool m_mask = false;
    char m_short_payload_size = 0;

    char control_buffer[125]; // close, ping, pong.

    // A text or binary message can be fragmented. The
    // message is built up in m_message_buffer.
    std::vector<char> m_message_buffer;

    // The opcode of the message.
    websocket::Opcode m_message_opcode = websocket::Opcode::continuation;

    // The size of the stored Websocket message.
    // This size is not the same as the size of the buffer.
    size_t m_message_size = 0;

    // The message buffer has a minimum size,
    // and is extended when a large message arrives.
    // The message_buffer is resized to this value after
    // a larger message has been delivered.
    static const size_t s_message_buffer_min_size = 2048;

    enum class Stage { init, header_beginning, header_end, payload, delivery };
    Stage m_stage = Stage::init;

    void set_protocol_error()
    {
        protocol_error = true;
    }

    void set_payload_buffer()
    {
        read_size = m_payload_size;

        if (m_opcode == websocket::Opcode::close || m_opcode == websocket::Opcode::ping ||
            m_opcode == websocket::Opcode::pong) {
            read_buffer = control_buffer;
        }
        else {
            size_t required_size = m_message_size + m_payload_size;
            if (m_message_buffer.size() < required_size)
                m_message_buffer.resize(required_size);

            read_buffer = m_message_buffer.data() + m_message_size;
        }
    }

    void reset_message_buffer()
    {
        if (m_message_buffer.size() != s_message_buffer_min_size)
            m_message_buffer.resize(s_message_buffer_min_size);
        m_message_opcode = websocket::Opcode::continuation;
        m_message_size = 0;
    }


    // stage_init() is only called once for a FrameReader.
    // It just moves the stage to header_beginning and
    // asks for two bytes to be read into the header_buffer.
    void stage_init()
    {
        protocol_error = false;
        delivery_ready = false;
        delivery_buffer = nullptr;
        delivery_size = 0;
        delivery_opcode = websocket::Opcode::continuation;
        m_stage = Stage::header_beginning;
        reset_message_buffer();
        read_buffer = header_buffer;
        read_size = 2;
    }

    // When stage_header_beginning() is called, the
    // first two bytes in the header_buffer are
    // the first two bytes of an incoming WebSocket frame.
    void stage_header_beginning()
    {
        // bit 1.
        m_fin = ((header_buffer[0] & 128) == 128);

        // bit 2,3, and 4.
        char rsv = (header_buffer[0] & 112) >> 4;
        if (rsv != 0)
            return set_protocol_error();

        // bit 5, 6, 7, and 8.
        char op = (header_buffer[0] & 15);

        if (op != 0 && op != 1 && op != 2 && op != 8 && op != 9 && op != 10)
            return set_protocol_error();

        m_opcode = websocket::Opcode(op);

        // bit 9.
        m_mask = ((header_buffer[1] & 128) == 128);
        if ((m_mask && m_is_client) || (!m_mask && !m_is_client))
            return set_protocol_error();

        // Remainder of second byte.
        m_short_payload_size = (header_buffer[1] & 127);

        if (m_opcode == websocket::Opcode::continuation) {
            if (m_message_opcode == websocket::Opcode::continuation)
                return set_protocol_error();
        }
        else if (m_opcode == websocket::Opcode::text || m_opcode == websocket::Opcode::binary) {
            if (m_message_opcode != websocket::Opcode::continuation)
                return set_protocol_error();

            m_message_opcode = m_opcode;
        }
        else { // close, ping, pong.
            if (!m_fin || m_short_payload_size > 125)
                return set_protocol_error();
        }

        if (m_short_payload_size <= 125 && m_mask) {
            m_stage = Stage::header_end;
            m_payload_size = m_short_payload_size;
            read_size = 4;
            read_buffer = header_buffer + 2;
        }
        else if (m_short_payload_size <= 125 && !m_mask) {
            m_stage = Stage::payload;
            m_payload_size = m_short_payload_size;
            set_payload_buffer();
        }
        else if (m_short_payload_size == 126 && m_mask) {
            m_stage = Stage::header_end;
            read_size = 6;
            read_buffer = header_buffer + 2;
        }
        else if (m_short_payload_size == 126 && !m_mask) {
            m_stage = Stage::header_end;
            read_size = 2;
            read_buffer = header_buffer + 2;
        }
        else if (m_short_payload_size == 127 && m_mask) {
            m_stage = Stage::header_end;
            read_size = 12;
            read_buffer = header_buffer + 2;
        }
        else if (m_short_payload_size == 127 && !m_mask) {
            m_stage = Stage::header_end;
            read_size = 8;
            read_buffer = header_buffer + 2;
        }
    }

    void stage_header_end()
    {
        if (m_short_payload_size <= 125) {
            m_masking_key = header_buffer + 2;
        }
        else if (m_short_payload_size == 126) {
            const unsigned char* bytes = reinterpret_cast<unsigned char*>(header_buffer + 2);
            m_payload_size = bytes[0] * 256 + bytes[1];

            if (m_mask)
                m_masking_key = header_buffer + 4;
        }
        else if (m_short_payload_size == 127) {
            if (header_buffer[2] != 0 || header_buffer[3] != 0 || header_buffer[4] != 0 || header_buffer[5] != 0) {
                // Message should be smaller than 4GB
                // FIXME: We should introduce a maximum size for messages.
                set_protocol_error();
                return;
            }

            // Assume size_t is at least 4 bytes wide.
            const unsigned char* bytes = reinterpret_cast<unsigned char*>(header_buffer + 6);
            m_payload_size = bytes[0];
            m_payload_size = 256 * m_payload_size + bytes[1];
            m_payload_size = 256 * m_payload_size + bytes[2];
            m_payload_size = 256 * m_payload_size + bytes[3];

            if (m_mask)
                m_masking_key = header_buffer + 10;
        }

        m_stage = Stage::payload;
        set_payload_buffer();
    }

    void stage_payload()
    {
        if (m_mask)
            mask_payload(m_masking_key, read_buffer, m_payload_size, read_buffer);

        if (m_opcode == websocket::Opcode::close || m_opcode == websocket::Opcode::ping ||
            m_opcode == websocket::Opcode::pong) {
            m_stage = Stage::delivery;
            delivery_ready = true;
            delivery_opcode = m_opcode;
            delivery_buffer = control_buffer;
            delivery_size = m_payload_size;
        }
        else {
            m_message_size += m_payload_size;
            if (m_fin) {
                m_stage = Stage::delivery;
                delivery_ready = true;
                delivery_opcode = m_message_opcode;
                delivery_buffer = m_message_buffer.data();
                delivery_size = m_message_size;
            }
            else {
                m_stage = Stage::header_beginning;
                read_buffer = header_buffer;
                read_size = 2;
            }
        }
    }

    void stage_delivery()
    {
        m_stage = Stage::header_beginning;
        read_buffer = header_buffer;
        read_size = 2;
        delivery_ready = false;
        delivery_buffer = nullptr;
        delivery_size = 0;
        delivery_opcode = websocket::Opcode::continuation;

        if (m_opcode == websocket::Opcode::continuation || m_opcode == websocket::Opcode::text ||
            m_opcode == websocket::Opcode::binary)
            reset_message_buffer();
    }
};


class WebSocket {
public:
    WebSocket(websocket::Config& config)
        : m_config(config)
        , m_logger_ptr(config.websocket_get_logger())
        , m_logger{*m_logger_ptr}
        , m_frame_reader(m_logger, m_is_client)
    {
        m_logger.debug(util::LogCategory::network, "WebSocket::Websocket()");
    }

    void initiate_client_handshake(const std::string& request_uri, const std::string& host,
                                   const std::string& sec_websocket_protocol, HTTPHeaders headers)
    {
        m_logger.debug(util::LogCategory::network, "WebSocket::initiate_client_handshake()");

        m_stopped = false;
        m_is_client = true;

        m_sec_websocket_key = make_random_sec_websocket_key(m_config.websocket_get_random());

        m_http_client.reset(new HTTPClient<websocket::Config>(m_config, m_logger_ptr));
        m_frame_reader.reset();
        HTTPRequest req;
        req.method = HTTPMethod::Get;
        req.path = std::move(request_uri);
        req.headers = std::move(headers);
        req.headers["Host"] = std::move(host);
        req.headers["Upgrade"] = "websocket";
        req.headers["Connection"] = "Upgrade";
        req.headers["Sec-WebSocket-Key"] = m_sec_websocket_key;
        req.headers["Sec-WebSocket-Version"] = sec_websocket_version;
        req.headers["Sec-WebSocket-Protocol"] = sec_websocket_protocol;

        m_logger.trace(util::LogCategory::network, "HTTP request =\n%1", req);

        auto handler = [this](HTTPResponse response, std::error_code ec) {
            // If the operation is aborted, the WebSocket object may have been destroyed.
            if (ec != util::error::operation_aborted) {
                if (ec == HTTPParserError::MalformedResponse) {
                    error_client_malformed_response();
                    return;
                }
                if (ec) {
                    stop();

                    // FIXME: Should be read instaed of write???
                    m_config.websocket_write_error_handler(ec);

                    return;
                }
                if (m_stopped)
                    return;
                handle_http_response_received(std::move(response)); // Throws
            }
        };

        m_http_client->async_request(req, std::move(handler));
    }

    void initiate_server_websocket_after_handshake()
    {
        m_stopped = false;
        m_is_client = false;
        m_frame_reader.reset();
        frame_reader_loop(); // Throws
    }

    void initiate_server_handshake()
    {
        m_logger.debug(util::LogCategory::network, "WebSocket::initiate_server_handshake()");

        m_stopped = false;
        m_is_client = false;
        m_http_server.reset(new HTTPServer<websocket::Config>(m_config, m_logger_ptr));
        m_frame_reader.reset();

        auto handler = [this](HTTPRequest request, std::error_code ec) {
            if (ec != util::error::operation_aborted) {
                if (ec == HTTPParserError::MalformedRequest) {
                    error_server_malformed_request();
                    return;
                }
                if (ec) {
                    stop();
                    m_config.websocket_read_error_handler(ec);
                    return;
                }
                if (m_stopped)
                    return;
                handle_http_request_received(std::move(request));
            }
        };

        m_http_server->async_receive_request(std::move(handler));
    }

    void async_write_frame(bool fin, int opcode, const char* data, size_t size,
                           sync::websocket::WriteCompletionHandler write_completion_handler)
    {
        REALM_ASSERT(!m_stopped);

        bool mask = m_is_client;

        // 14 is the maximum header length of a Websocket frame.
        size_t required_size = size + 14;
        if (m_write_buffer.size() < required_size)
            m_write_buffer.resize(required_size);

        size_t message_size =
            make_frame(fin, opcode, mask, data, size, m_write_buffer.data(), m_config.websocket_get_random());

        auto handler = [this, handler = std::move(write_completion_handler)](std::error_code ec, size_t) mutable {
            // If the operation is aborted, then the write operation was canceled and we should ignore this callback.
            if (ec == util::error::operation_aborted) {
                return handler(ec, 0);
            }

            auto is_socket_closed_err = (ec == util::error::make_error_code(util::error::connection_reset) ||
                                         ec == util::error::make_error_code(util::error::broken_pipe) ||
                                         ec == util::make_error_code(util::MiscExtErrors::end_of_input));
            // If the socket has been closed then we should continue to read from it until we've drained
            // the receive buffer. Eventually we will either receive an in-band error message from the
            // server about why we got disconnected or we'll receive ECONNRESET on the receive side as well.
            if (is_socket_closed_err) {
                return;
            }

            // Otherwise we've got some other I/O error that we should surface to the sync client.
            if (ec) {
                stop();
                return m_config.websocket_write_error_handler(ec);
            }

            handle_write_message(std::move(handler));
        };

        m_config.async_write(m_write_buffer.data(), message_size, std::move(handler));
    }

    void handle_write_message(sync::websocket::WriteCompletionHandler write_handler)
    {
        if (m_write_buffer.size() > s_write_buffer_stable_size) {
            m_write_buffer.resize(s_write_buffer_stable_size);
            m_write_buffer.shrink_to_fit();
        }

        write_handler(std::error_code(), m_write_buffer.size());
    }

    void stop() noexcept
    {
        m_stopped = true;
        m_frame_reader.reset();
    }

    void force_handshake_response_for_testing(int status_code, std::string body)
    {
        m_test_handshake_response.emplace(status_code);
        m_test_handshake_response_body = body;
    }

private:
    websocket::Config& m_config;
    const std::shared_ptr<util::Logger> m_logger_ptr;
    util::Logger& m_logger;
    FrameReader m_frame_reader;

    bool m_stopped = false;
    bool m_is_client;

    // Allocated on demand.
    std::unique_ptr<HTTPClient<websocket::Config>> m_http_client;
    std::unique_ptr<HTTPServer<websocket::Config>> m_http_server;

    std::string m_sec_websocket_key;
    std::string m_sec_websocket_accept;

    std::vector<char> m_write_buffer;
    static const size_t s_write_buffer_stable_size = 2048;

    std::optional<int> m_test_handshake_response;
    std::string m_test_handshake_response_body;

    void error_client_malformed_response()
    {
        m_stopped = true;
        m_logger.error(util::LogCategory::network, "WebSocket: Received malformed HTTP response");
        std::error_code ec = HttpError::bad_response_invalid_http;
        m_config.websocket_handshake_error_handler(ec, nullptr, nullptr); // Throws
    }

    void error_client_response_not_101(const HTTPResponse& response)
    {
        m_stopped = true;

        m_logger.error(util::LogCategory::network,
                       "Websocket: Expected HTTP response 101 Switching Protocols, "
                       "but received:\n%1",
                       response);

        int status_code = int(response.status);
        std::error_code ec;

        if (m_test_handshake_response)
            status_code = *m_test_handshake_response;

        if (status_code == 200)
            ec = HttpError::bad_response_200_ok;
        else if (status_code >= 200 && status_code < 300)
            ec = HttpError::bad_response_2xx_successful;
        else if (status_code == 301)
            ec = HttpError::bad_response_301_moved_permanently;
        else if (status_code == 308)
            ec = HttpError::bad_response_308_permanent_redirect;
        else if (status_code >= 300 && status_code < 400)
            ec = HttpError::bad_response_3xx_redirection;
        else if (status_code == 401)
            ec = HttpError::bad_response_401_unauthorized;
        else if (status_code == 403)
            ec = HttpError::bad_response_403_forbidden;
        else if (status_code == 404)
            ec = HttpError::bad_response_404_not_found;
        else if (status_code == 410)
            ec = HttpError::bad_response_410_gone;
        else if (status_code >= 400 && status_code < 500)
            ec = HttpError::bad_response_4xx_client_errors;
        else if (status_code == 500)
            ec = HttpError::bad_response_500_internal_server_error;
        else if (status_code == 502)
            ec = HttpError::bad_response_502_bad_gateway;
        else if (status_code == 503)
            ec = HttpError::bad_response_503_service_unavailable;
        else if (status_code == 504)
            ec = HttpError::bad_response_504_gateway_timeout;
        else if (status_code >= 500 && status_code < 600)
            ec = HttpError::bad_response_5xx_server_error;
        else
            ec = HttpError::bad_response_unexpected_status_code;

        std::string_view body;
        std::string_view* body_ptr = nullptr;
        if (m_test_handshake_response) {
            body = m_test_handshake_response_body;
            body_ptr = &body;
        }
        else if (response.body) {
            body = *response.body;
            body_ptr = &body;
        }
        m_config.websocket_handshake_error_handler(ec, &response.headers, body_ptr); // Throws
    }

    void error_client_response_websocket_headers_invalid(const HTTPResponse& response)
    {
        m_stopped = true;

        m_logger.error(util::LogCategory::network,
                       "Websocket: HTTP response has invalid websocket headers."
                       "HTTP response = \n%1",
                       response);
        std::error_code ec = HttpError::bad_response_header_protocol_violation;
        std::string_view body;
        std::string_view* body_ptr = nullptr;
        if (response.body) {
            body = *response.body;
            body_ptr = &body;
        }
        m_config.websocket_handshake_error_handler(ec, &response.headers, body_ptr); // Throws
    }

    void error_server_malformed_request()
    {
        m_stopped = true;
        m_logger.error(util::LogCategory::network, "WebSocket: Received malformed HTTP request");
        std::error_code ec = HttpError::bad_request_malformed_http;
        m_config.websocket_handshake_error_handler(ec, nullptr, nullptr); // Throws
    }

    void error_server_request_header_protocol_violation(std::error_code ec, const HTTPRequest& request)
    {
        m_stopped = true;

        m_logger.error(util::LogCategory::network,
                       "Websocket: HTTP request has invalid websocket headers."
                       "HTTP request = \n%1",
                       request);
        m_config.websocket_handshake_error_handler(ec, &request.headers, nullptr); // Throws
    }

    void protocol_error(std::error_code ec)
    {
        m_stopped = true;
        m_config.websocket_protocol_error_handler(ec);
    }

    // The client receives the HTTP response.
    void handle_http_response_received(HTTPResponse response)
    {
        m_logger.debug(util::LogCategory::network, "WebSocket::handle_http_response_received()");
        m_logger.trace(util::LogCategory::network, "HTTP response = %1", response);

        if (response.status != HTTPStatus::SwitchingProtocols ||
            (m_test_handshake_response && *m_test_handshake_response != 101)) {
            error_client_response_not_101(response);
            return;
        }

        bool valid = (find_sec_websocket_accept(response.headers) &&
                      m_sec_websocket_accept == make_sec_websocket_accept(m_sec_websocket_key));
        if (!valid) {
            error_client_response_websocket_headers_invalid(response);
            return;
        }

        m_config.websocket_handshake_completion_handler(response.headers);

        if (m_stopped)
            return;

        frame_reader_loop();
    }

    void handle_http_request_received(HTTPRequest request)
    {
        m_logger.trace(util::LogCategory::network, "WebSocket::handle_http_request_received()");

        util::Optional<std::string> sec_websocket_protocol = websocket::read_sec_websocket_protocol(request);

        std::error_code ec;
        util::Optional<HTTPResponse> response =
            do_make_http_response(request, sec_websocket_protocol ? *sec_websocket_protocol : "realm.io", ec);

        if (ec) {
            error_server_request_header_protocol_violation(ec, request);
            return;
        }
        REALM_ASSERT(response);

        auto handler = [request, this](std::error_code ec) {
            // If the operation is aborted, the socket object may have been destroyed.
            if (ec != util::error::operation_aborted) {
                if (ec) {
                    stop();
                    m_config.websocket_write_error_handler(ec);
                    return;
                }

                if (m_stopped)
                    return;

                m_config.websocket_handshake_completion_handler(request.headers);

                if (m_stopped)
                    return;

                frame_reader_loop(); // Throws
            }
        };
        m_http_server->async_send_response(*response, std::move(handler));
    }

    // find_sec_websocket_accept is similar to
    // find_sec_websockey_key.
    bool find_sec_websocket_accept(const HTTPHeaders& headers)
    {
        util::Optional<StringData> header_value = find_http_header_value(headers, "Sec-WebSocket-Accept");

        if (!header_value)
            return false;

        m_sec_websocket_accept = *header_value;

        return true;
    }

    std::pair<WebSocketError, std::string_view> parse_close_message(const char* data, size_t size)
    {
        uint16_t error_code;
        std::string_view error_message;
        if (size < 2) {
            // Error code 1005 is defined as
            //     1005 is a reserved value and MUST NOT be set as a status code in a
            //     Close control frame by an endpoint.  It is designated for use in
            //     applications expecting a status code to indicate that no status
            //     code was actually present.
            // See https://tools.ietf.org/html/rfc6455#section-7.4.1 for more details
            error_code = 1005;
        }
        else {
            // Otherwise, the error code is the first two bytes of the body as a uint16_t in
            // network byte order. See https://tools.ietf.org/html/rfc6455#section-5.5.1 for more
            // details.
            error_code = ntohs((uint8_t(data[1]) << 8) | uint8_t(data[0]));
            error_message = std::string_view(data + 2, size - 2);
        }

        switch (static_cast<WebSocketError>(error_code)) {
            case WebSocketError::websocket_ok:
            case WebSocketError::websocket_going_away:
            case WebSocketError::websocket_protocol_error:
            case WebSocketError::websocket_unsupported_data:
            case WebSocketError::websocket_reserved:
            case WebSocketError::websocket_no_status_received:
            case WebSocketError::websocket_abnormal_closure:
            case WebSocketError::websocket_invalid_payload_data:
            case WebSocketError::websocket_policy_violation:
            case WebSocketError::websocket_message_too_big:
            case WebSocketError::websocket_invalid_extension:
            case WebSocketError::websocket_internal_server_error:
            case WebSocketError::websocket_tls_handshake_failed:

            case WebSocketError::websocket_unauthorized:
            case WebSocketError::websocket_forbidden:
            case WebSocketError::websocket_moved_permanently:
            case WebSocketError::websocket_client_too_old:
            case WebSocketError::websocket_client_too_new:
            case WebSocketError::websocket_protocol_mismatch:
                break;
            default:
                error_code = 1008;
        }

        return std::make_pair(static_cast<WebSocketError>(error_code), error_message);
    }

    // frame_reader_loop() uses the frame_reader to read and process the incoming
    // WebSocket messages.
    void frame_reader_loop()
    {
        // Advance parsing stage.
        m_frame_reader.next();

        if (m_frame_reader.protocol_error) {
            protocol_error(HttpError::bad_message);
            return;
        }

        if (m_frame_reader.delivery_ready) {
            bool should_continue = true;

            switch (m_frame_reader.delivery_opcode) {
                case websocket::Opcode::text:
                    should_continue = m_config.websocket_text_message_received(m_frame_reader.delivery_buffer,
                                                                               m_frame_reader.delivery_size);
                    break;
                case websocket::Opcode::binary:
                    should_continue = m_config.websocket_binary_message_received(m_frame_reader.delivery_buffer,
                                                                                 m_frame_reader.delivery_size);
                    break;
                case websocket::Opcode::close: {
                    auto [error_code, error_message] =
                        parse_close_message(m_frame_reader.delivery_buffer, m_frame_reader.delivery_size);
                    should_continue = m_config.websocket_close_message_received(error_code, error_message);
                    break;
                }
                case websocket::Opcode::ping:
                    should_continue = m_config.websocket_ping_message_received(m_frame_reader.delivery_buffer,
                                                                               m_frame_reader.delivery_size);
                    break;
                case websocket::Opcode::pong:
                    should_continue = m_config.websocket_pong_message_received(m_frame_reader.delivery_buffer,
                                                                               m_frame_reader.delivery_size);
                    break;
                default:
                    break;
            }

            // The websocket object might not even exist anymore
            if (!should_continue)
                return;

            if (m_stopped)
                return;

            // recursion is harmless, since the depth will be at most 2.
            frame_reader_loop();
            return;
        }

        auto handler = [this](std::error_code ec, size_t) {
            // If the operation is aborted, the socket object may have been destroyed.
            if (ec != util::error::operation_aborted) {
                if (ec) {
                    stop();
                    m_config.websocket_read_error_handler(ec);
                    return;
                }

                if (m_stopped)
                    return;

                frame_reader_loop();
            }
        };

        m_config.async_read(m_frame_reader.read_buffer, m_frame_reader.read_size, std::move(handler));
    }
};


const char* get_error_message(HttpError error_code)
{
    switch (error_code) {
        case HttpError::bad_request_malformed_http:
            return "Bad WebSocket request malformed HTTP";
        case HttpError::bad_request_header_upgrade:
            return "Bad WebSocket request header: Upgrade";
        case HttpError::bad_request_header_connection:
            return "Bad WebSocket request header: Connection";
        case HttpError::bad_request_header_websocket_version:
            return "Bad WebSocket request header: Sec-Websocket-Version";
        case HttpError::bad_request_header_websocket_key:
            return "Bad WebSocket request header: Sec-Websocket-Key";
        case HttpError::bad_response_invalid_http:
            return "Bad WebSocket response invalid HTTP";
        case HttpError::bad_response_2xx_successful:
            return "Bad WebSocket response 2xx successful";
        case HttpError::bad_response_200_ok:
            return "Bad WebSocket response 200 ok";
        case HttpError::bad_response_3xx_redirection:
            return "Bad WebSocket response 3xx redirection";
        case HttpError::bad_response_301_moved_permanently:
            return "Bad WebSocket response 301 moved permanently";
        case HttpError::bad_response_308_permanent_redirect:
            return "Bad WebSocket response 308 permanent redirect";
        case HttpError::bad_response_4xx_client_errors:
            return "Bad WebSocket response 4xx client errors";
        case HttpError::bad_response_401_unauthorized:
            return "Bad WebSocket response 401 unauthorized";
        case HttpError::bad_response_403_forbidden:
            return "Bad WebSocket response 403 forbidden";
        case HttpError::bad_response_404_not_found:
            return "Bad WebSocket response 404 not found";
        case HttpError::bad_response_410_gone:
            return "Bad WebSocket response 410 gone";
        case HttpError::bad_response_5xx_server_error:
            return "Bad WebSocket response 5xx server error";
        case HttpError::bad_response_500_internal_server_error:
            return "Bad WebSocket response 500 internal server error";
        case HttpError::bad_response_502_bad_gateway:
            return "Bad WebSocket response 502 bad gateway";
        case HttpError::bad_response_503_service_unavailable:
            return "Bad WebSocket response 503 service unavailable";
        case HttpError::bad_response_504_gateway_timeout:
            return "Bad WebSocket response 504 gateway timeout";
        case HttpError::bad_response_unexpected_status_code:
            return "Bad Websocket response unexpected status code";
        case HttpError::bad_response_header_protocol_violation:
            return "Bad WebSocket response header protocol violation";
        case HttpError::bad_message:
            return "Ill-formed WebSocket message";
    }
    return nullptr;
}


class HttpErrorCategory : public std::error_category {
public:
    const char* name() const noexcept override final
    {
        return "realm::sync::websocket::HttpError";
    }
    std::string message(int error_code) const override final
    {
        const char* msg = get_error_message(HttpError(error_code));
        if (!msg)
            msg = "Unknown error";
        std::string msg_2{msg}; // Throws (copy)
        return msg_2;
    }
};

} // unnamed namespace

namespace realm::sync::websocket {

std::ostream& operator<<(std::ostream& os, WebSocketError code)
{
    /// WebSocket error codes
    auto str = [&]() -> const char* {
        switch (code) {
            case WebSocketError::websocket_ok:
                return "WebSocket: OK";
            case WebSocketError::websocket_going_away:
                return "WebSocket: Going Away";
            case WebSocketError::websocket_protocol_error:
                return "WebSocket: Protocol Error";
            case WebSocketError::websocket_unsupported_data:
                return "WebSocket: Unsupported Data";
            case WebSocketError::websocket_reserved:
                return "WebSocket: Reserved";
            case WebSocketError::websocket_no_status_received:
                return "WebSocket: No Status Received";
            case WebSocketError::websocket_abnormal_closure:
                return "WebSocket: Abnormal Closure";
            case WebSocketError::websocket_invalid_payload_data:
                return "WebSocket: Invalid Payload Data";
            case WebSocketError::websocket_policy_violation:
                return "WebSocket: Policy Violation";
            case WebSocketError::websocket_message_too_big:
                return "WebSocket: Message Too Big";
            case WebSocketError::websocket_invalid_extension:
                return "WebSocket: Invalid Extension";
            case WebSocketError::websocket_internal_server_error:
                return "WebSocket: Internal Server Error";
            case WebSocketError::websocket_tls_handshake_failed:
                return "WebSocket: TLS Handshake Failed";

            /// WebSocket Errors - reported by server
            case WebSocketError::websocket_unauthorized:
                return "WebSocket: Unauthorized";
            case WebSocketError::websocket_forbidden:
                return "WebSocket: Forbidden";
            case WebSocketError::websocket_moved_permanently:
                return "WebSocket: Moved Permanently";
            case WebSocketError::websocket_client_too_old:
                return "WebSocket: Client Too Old";
            case WebSocketError::websocket_client_too_new:
                return "WebSocket: Client Too New";
            case WebSocketError::websocket_protocol_mismatch:
                return "WebSocket: Protocol Mismatch";

            case WebSocketError::websocket_resolve_failed:
                return "WebSocket: Resolve Failed";
            case WebSocketError::websocket_connection_failed:
                return "WebSocket: Connection Failed";
            case WebSocketError::websocket_read_error:
                return "WebSocket: Read Error";
            case WebSocketError::websocket_write_error:
                return "WebSocket: Write Error";
            case WebSocketError::websocket_retry_error:
                return "WebSocket: Retry Error";
            case WebSocketError::websocket_fatal_error:
                return "WebSocket: Fatal Error";
        }
        return nullptr;
    }();

    if (str == nullptr) {
        os << "WebSocket: Unknown Error (" << static_cast<std::underlying_type_t<WebSocketError>>(code) << ")";
    }
    else {
        os << str;
    }
    return os;
}

} // namespace realm::sync::websocket

bool websocket::Config::websocket_text_message_received(const char*, size_t)
{
    return true;
}

bool websocket::Config::websocket_binary_message_received(const char*, size_t)
{
    return true;
}

bool websocket::Config::websocket_close_message_received(WebSocketError, std::string_view)
{
    return true;
}

bool websocket::Config::websocket_ping_message_received(const char*, size_t)
{
    return true;
}

bool websocket::Config::websocket_pong_message_received(const char*, size_t)
{
    return true;
}


class websocket::Socket::Impl : public WebSocket {
public:
    Impl(Config& config)
        : WebSocket(config) // Throws
    {
    }
};

websocket::Socket::Socket(Config& config)
    : m_impl(new Impl{config})
{
}

websocket::Socket::Socket(Socket&& socket) noexcept
    : m_impl(std::move(socket.m_impl))
{
}

websocket::Socket::~Socket() noexcept {}

void websocket::Socket::initiate_client_handshake(const std::string& request_uri, const std::string& host,
                                                  const std::string& sec_websocket_protocol, HTTPHeaders headers)
{
    m_impl->initiate_client_handshake(request_uri, host, sec_websocket_protocol, std::move(headers));
}

void websocket::Socket::initiate_server_handshake()
{
    m_impl->initiate_server_handshake();
}

void websocket::Socket::initiate_server_websocket_after_handshake()
{
    m_impl->initiate_server_websocket_after_handshake();
}

void websocket::Socket::async_write_frame(bool fin, Opcode opcode, const char* data, size_t size,
                                          WriteCompletionHandler handler)
{
    m_impl->async_write_frame(fin, int(opcode), data, size, std::move(handler));
}

void websocket::Socket::async_write_text(const char* data, size_t size, WriteCompletionHandler handler)
{
    async_write_frame(true, Opcode::text, data, size, std::move(handler));
}

void websocket::Socket::async_write_binary(const char* data, size_t size, WriteCompletionHandler handler)
{
    async_write_frame(true, Opcode::binary, data, size, std::move(handler));
}

void websocket::Socket::async_write_close(const char* data, size_t size, WriteCompletionHandler handler)
{
    async_write_frame(true, Opcode::close, data, size, std::move(handler));
}

void websocket::Socket::async_write_ping(const char* data, size_t size, WriteCompletionHandler handler)
{
    async_write_frame(true, Opcode::ping, data, size, std::move(handler));
}

void websocket::Socket::async_write_pong(const char* data, size_t size, WriteCompletionHandler handler)
{
    async_write_frame(true, Opcode::pong, data, size, std::move(handler));
}

void websocket::Socket::stop() noexcept
{
    m_impl->stop();
}

void websocket::Socket::force_handshake_response_for_testing(int status_code, std::string body)
{
    m_impl->force_handshake_response_for_testing(status_code, body);
}

util::Optional<std::string> websocket::read_sec_websocket_protocol(const HTTPRequest& request)
{
    const HTTPHeaders& headers = request.headers;
    const StringData header = "Sec-WebSocket-Protocol";
    util::Optional<StringData> value = find_http_header_value(headers, header);
    return value ? util::Optional<std::string>(std::string(*value)) : util::none;
}

util::Optional<HTTPResponse> websocket::make_http_response(const HTTPRequest& request,
                                                           const std::string& sec_websocket_protocol,
                                                           std::error_code& ec)
{
    return do_make_http_response(request, sec_websocket_protocol, ec);
}

const std::error_category& websocket::http_error_category() noexcept
{
    static const HttpErrorCategory category = {};
    return category;
}

std::error_code websocket::make_error_code(HttpError error_code) noexcept
{
    return std::error_code{int(error_code), http_error_category()};
}

realm / realm-core / thomas.goyne_275

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous