ceph/src/Beast/include/beast/websocket/detail/stream_base.hpp

   1 //
   2 // Copyright (c) 2013-2017 Vinnie Falco (vinnie dot falco at gmail dot com)
   3 //
   4 // Distributed under the Boost Software License, Version 1.0. (See accompanying
   5 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   6 //
   7
   8 #ifndef BEAST_WEBSOCKET_DETAIL_STREAM_BASE_HPP
   9 #define BEAST_WEBSOCKET_DETAIL_STREAM_BASE_HPP
  10
  11 #include <beast/websocket/error.hpp>
  12 #include <beast/websocket/option.hpp>
  13 #include <beast/websocket/rfc6455.hpp>
  14 #include <beast/websocket/detail/decorator.hpp>
  15 #include <beast/websocket/detail/frame.hpp>
  16 #include <beast/websocket/detail/invokable.hpp>
  17 #include <beast/websocket/detail/mask.hpp>
  18 #include <beast/websocket/detail/pmd_extension.hpp>
  19 #include <beast/websocket/detail/utf8_checker.hpp>
  20 #include <beast/http/message.hpp>
  21 #include <beast/http/string_body.hpp>
  22 #include <beast/zlib/deflate_stream.hpp>
  23 #include <beast/zlib/inflate_stream.hpp>
  24 #include <boost/asio/error.hpp>
  25 #include <boost/assert.hpp>
  26 #include <cstdint>
  27 #include <memory>
  28
  29 namespace beast {
  30 namespace websocket {
  31 namespace detail {
  32
  33 /// Identifies the role of a WebSockets stream.
  34 enum class role_type
  35 {
  36     /// Stream is operating as a client.
  37     client,
  38
  39     /// Stream is operating as a server.
  40     server
  41 };
  42
  43 //------------------------------------------------------------------------------
  44
  45 struct stream_base
  46 {
  47 protected:
  48     friend class frame_test;
  49
  50     struct op {};
  51
  52     detail::maskgen maskgen_;               // source of mask keys
  53     decorator_type d_;                      // adorns http messages
  54     bool keep_alive_ = false;               // close on failed upgrade
  55     std::size_t rd_msg_max_ =
  56         16 * 1024 * 1024;                   // max message size
  57     bool wr_autofrag_ = true;               // auto fragment
  58     std::size_t wr_buf_size_ = 4096;        // write buffer size
  59     std::size_t rd_buf_size_ = 4096;        // read buffer size
  60     opcode wr_opcode_ = opcode::text;       // outgoing message type
  61     ping_cb ping_cb_;                       // ping callback
  62     role_type role_;                        // server or client
  63     bool failed_;                           // the connection failed
  64
  65     bool wr_close_;                         // sent close frame
  66     op* wr_block_;                          // op currenly writing
  67
  68     ping_data* ping_data_;                  // where to put the payload
  69     invokable rd_op_;                       // read parking
  70     invokable wr_op_;                       // write parking
  71     invokable ping_op_;                     // ping parking
  72     close_reason cr_;                       // set from received close frame
  73
  74     // State information for the message being received
  75     //
  76     struct rd_t
  77     {
  78         // opcode of current message being read
  79         opcode op;
  80
  81         // `true` if the next frame is a continuation.
  82         bool cont;
  83
  84         // Checks that test messages are valid utf8
  85         detail::utf8_checker utf8;
  86
  87         // Size of the current message so far.
  88         std::uint64_t size;
  89
  90         // Size of the read buffer.
  91         // This gets set to the read buffer size option at the
  92         // beginning of sending a message, so that the option can be
  93         // changed mid-send without affecting the current message.
  94         std::size_t buf_size;
  95
  96         // The read buffer. Used for compression and masking.
  97         std::unique_ptr<std::uint8_t[]> buf;
  98     };
  99
 100     rd_t rd_;
 101
 102     // State information for the message being sent
 103     //
 104     struct wr_t
 105     {
 106         // `true` if next frame is a continuation,
 107         // `false` if next frame starts a new message
 108         bool cont;
 109
 110         // `true` if this message should be auto-fragmented
 111         // This gets set to the auto-fragment option at the beginning
 112         // of sending a message, so that the option can be changed
 113         // mid-send without affecting the current message.
 114         bool autofrag;
 115
 116         // `true` if this message should be compressed.
 117         // This gets set to the compress option at the beginning of
 118         // of sending a message, so that the option can be changed
 119         // mid-send without affecting the current message.
 120         bool compress;
 121
 122         // Size of the write buffer.
 123         // This gets set to the write buffer size option at the
 124         // beginning of sending a message, so that the option can be
 125         // changed mid-send without affecting the current message.
 126         std::size_t buf_size;
 127
 128         // The write buffer. Used for compression and masking.
 129         // The buffer is allocated or reallocated at the beginning of
 130         // sending a message.
 131         std::unique_ptr<std::uint8_t[]> buf;
 132     };
 133
 134     wr_t wr_;
 135
 136     // State information for the permessage-deflate extension
 137     struct pmd_t
 138     {
 139         // `true` if current read message is compressed
 140         bool rd_set;
 141
 142         zlib::deflate_stream zo;
 143         zlib::inflate_stream zi;
 144     };
 145
 146     // If not engaged, then permessage-deflate is not
 147     // enabled for the currently active session.
 148     std::unique_ptr<pmd_t> pmd_;
 149
 150     // Local options for permessage-deflate
 151     permessage_deflate pmd_opts_;
 152
 153     // Offer for clients, negotiated result for servers
 154     pmd_offer pmd_config_;
 155
 156     stream_base(stream_base&&) = default;
 157     stream_base(stream_base const&) = delete;
 158     stream_base& operator=(stream_base&&) = default;
 159     stream_base& operator=(stream_base const&) = delete;
 160
 161     stream_base()
 162         : d_(detail::default_decorator{})
 163     {
 164     }
 165
 166     template<class = void>
 167     void
 168     open(role_type role);
 169
 170     template<class = void>
 171     void
 172     close();
 173
 174     template<class DynamicBuffer>
 175     std::size_t
 176     read_fh1(detail::frame_header& fh,
 177         DynamicBuffer& db, close_code& code);
 178
 179     template<class DynamicBuffer>
 180     void
 181     read_fh2(detail::frame_header& fh,
 182         DynamicBuffer& db, close_code& code);
 183
 184     // Called before receiving the first frame of each message
 185     template<class = void>
 186     void
 187     rd_begin();
 188
 189     // Called before sending the first frame of each message
 190     //
 191     template<class = void>
 192     void
 193     wr_begin();
 194
 195     template<class DynamicBuffer>
 196     void
 197     write_close(DynamicBuffer& db, close_reason const& rc);
 198
 199     template<class DynamicBuffer>
 200     void
 201     write_ping(DynamicBuffer& db, opcode op, ping_data const& data);
 202 };
 203
 204 template<class>
 205 void
 206 stream_base::
 207 open(role_type role)
 208 {
 209     // VFALCO TODO analyze and remove dupe code in reset()
 210     role_ = role;
 211     failed_ = false;
 212     rd_.cont = false;
 213     wr_close_ = false;
 214     wr_block_ = nullptr;    // should be nullptr on close anyway
 215     ping_data_ = nullptr;   // should be nullptr on close anyway
 216
 217     wr_.cont = false;
 218     wr_.buf_size = 0;
 219
 220     if(((role_ == role_type::client && pmd_opts_.client_enable) ||
 221         (role_ == role_type::server && pmd_opts_.server_enable)) &&
 222             pmd_config_.accept)
 223     {
 224         pmd_normalize(pmd_config_);
 225         pmd_.reset(new pmd_t);
 226         if(role_ == role_type::client)
 227         {
 228             pmd_->zi.reset(
 229                 pmd_config_.server_max_window_bits);
 230             pmd_->zo.reset(
 231                 pmd_opts_.compLevel,
 232                 pmd_config_.client_max_window_bits,
 233                 pmd_opts_.memLevel,
 234                 zlib::Strategy::normal);
 235         }
 236         else
 237         {
 238             pmd_->zi.reset(
 239                 pmd_config_.client_max_window_bits);
 240             pmd_->zo.reset(
 241                 pmd_opts_.compLevel,
 242                 pmd_config_.server_max_window_bits,
 243                 pmd_opts_.memLevel,
 244                 zlib::Strategy::normal);
 245         }
 246     }
 247 }
 248
 249 template<class>
 250 void
 251 stream_base::
 252 close()
 253 {
 254     rd_.buf.reset();
 255     wr_.buf.reset();
 256     pmd_.reset();
 257 }
 258
 259 // Read fixed frame header from buffer
 260 // Requires at least 2 bytes
 261 //
 262 template<class DynamicBuffer>
 263 std::size_t
 264 stream_base::
 265 read_fh1(detail::frame_header& fh,
 266     DynamicBuffer& db, close_code& code)
 267 {
 268     using boost::asio::buffer;
 269     using boost::asio::buffer_copy;
 270     using boost::asio::buffer_size;
 271     auto const err =
 272         [&](close_code cv)
 273         {
 274             code = cv;
 275             return 0;
 276         };
 277     std::uint8_t b[2];
 278     BOOST_ASSERT(buffer_size(db.data()) >= sizeof(b));
 279     db.consume(buffer_copy(buffer(b), db.data()));
 280     std::size_t need;
 281     fh.len = b[1] & 0x7f;
 282     switch(fh.len)
 283     {
 284         case 126: need = 2; break;
 285         case 127: need = 8; break;
 286         default:
 287             need = 0;
 288     }
 289     fh.mask = (b[1] & 0x80) != 0;
 290     if(fh.mask)
 291         need += 4;
 292     fh.op   = static_cast<opcode>(b[0] & 0x0f);
 293     fh.fin  = (b[0] & 0x80) != 0;
 294     fh.rsv1 = (b[0] & 0x40) != 0;
 295     fh.rsv2 = (b[0] & 0x20) != 0;
 296     fh.rsv3 = (b[0] & 0x10) != 0;
 297     switch(fh.op)
 298     {
 299     case opcode::binary:
 300     case opcode::text:
 301         if(rd_.cont)
 302         {
 303             // new data frame when continuation expected
 304             return err(close_code::protocol_error);
 305         }
 306         if((fh.rsv1 && ! pmd_) ||
 307             fh.rsv2 || fh.rsv3)
 308         {
 309             // reserved bits not cleared
 310             return err(close_code::protocol_error);
 311         }
 312         if(pmd_)
 313             pmd_->rd_set = fh.rsv1;
 314         break;
 315
 316     case opcode::cont:
 317         if(! rd_.cont)
 318         {
 319             // continuation without an active message
 320             return err(close_code::protocol_error);
 321         }
 322         if(fh.rsv1 || fh.rsv2 || fh.rsv3)
 323         {
 324             // reserved bits not cleared
 325             return err(close_code::protocol_error);
 326         }
 327         break;
 328
 329     default:
 330         if(is_reserved(fh.op))
 331         {
 332             // reserved opcode
 333             return err(close_code::protocol_error);
 334         }
 335         if(! fh.fin)
 336         {
 337             // fragmented control message
 338             return err(close_code::protocol_error);
 339         }
 340         if(fh.len > 125)
 341         {
 342             // invalid length for control message
 343             return err(close_code::protocol_error);
 344         }
 345         if(fh.rsv1 || fh.rsv2 || fh.rsv3)
 346         {
 347             // reserved bits not cleared
 348             return err(close_code::protocol_error);
 349         }
 350         break;
 351     }
 352     // unmasked frame from client
 353     if(role_ == role_type::server && ! fh.mask)
 354     {
 355         code = close_code::protocol_error;
 356         return 0;
 357     }
 358     // masked frame from server
 359     if(role_ == role_type::client && fh.mask)
 360     {
 361         code = close_code::protocol_error;
 362         return 0;
 363     }
 364     code = close_code::none;
 365     return need;
 366 }
 367
 368 // Decode variable frame header from buffer
 369 //
 370 template<class DynamicBuffer>
 371 void
 372 stream_base::
 373 read_fh2(detail::frame_header& fh,
 374     DynamicBuffer& db, close_code& code)
 375 {
 376     using boost::asio::buffer;
 377     using boost::asio::buffer_copy;
 378     using boost::asio::buffer_size;
 379     using namespace boost::endian;
 380     switch(fh.len)
 381     {
 382     case 126:
 383     {
 384         std::uint8_t b[2];
 385         BOOST_ASSERT(buffer_size(db.data()) >= sizeof(b));
 386         db.consume(buffer_copy(buffer(b), db.data()));
 387         fh.len = big_uint16_to_native(&b[0]);
 388         // length not canonical
 389         if(fh.len < 126)
 390         {
 391             code = close_code::protocol_error;
 392             return;
 393         }
 394         break;
 395     }
 396     case 127:
 397     {
 398         std::uint8_t b[8];
 399         BOOST_ASSERT(buffer_size(db.data()) >= sizeof(b));
 400         db.consume(buffer_copy(buffer(b), db.data()));
 401         fh.len = big_uint64_to_native(&b[0]);
 402         // length not canonical
 403         if(fh.len < 65536)
 404         {
 405             code = close_code::protocol_error;
 406             return;
 407         }
 408         break;
 409     }
 410     }
 411     if(fh.mask)
 412     {
 413         std::uint8_t b[4];
 414         BOOST_ASSERT(buffer_size(db.data()) >= sizeof(b));
 415         db.consume(buffer_copy(buffer(b), db.data()));
 416         fh.key = little_uint32_to_native(&b[0]);
 417     }
 418     else
 419     {
 420         // initialize this otherwise operator== breaks
 421         fh.key = 0;
 422     }
 423     if(! is_control(fh.op))
 424     {
 425         if(fh.op != opcode::cont)
 426         {
 427             rd_.size = 0;
 428             rd_.op = fh.op;
 429         }
 430         else
 431         {
 432             if(rd_.size > (std::numeric_limits<
 433                 std::uint64_t>::max)() - fh.len)
 434             {
 435                 code = close_code::too_big;
 436                 return;
 437             }
 438         }
 439         rd_.cont = ! fh.fin;
 440     }
 441     code = close_code::none;
 442 }
 443
 444 template<class>
 445 void
 446 stream_base::
 447 rd_begin()
 448 {
 449     // Maintain the read buffer
 450     if(pmd_)
 451     {
 452         if(! rd_.buf || rd_.buf_size != rd_buf_size_)
 453         {
 454             rd_.buf_size = rd_buf_size_;
 455             rd_.buf.reset(new std::uint8_t[rd_.buf_size]);
 456         }
 457     }
 458 }
 459
 460 template<class>
 461 void
 462 stream_base::
 463 wr_begin()
 464 {
 465     wr_.autofrag = wr_autofrag_;
 466     wr_.compress = static_cast<bool>(pmd_);
 467
 468     // Maintain the write buffer
 469     if( wr_.compress ||
 470         role_ == detail::role_type::client)
 471     {
 472         if(! wr_.buf || wr_.buf_size != wr_buf_size_)
 473         {
 474             wr_.buf_size = wr_buf_size_;
 475             wr_.buf.reset(new std::uint8_t[wr_.buf_size]);
 476         }
 477     }
 478     else
 479     {
 480         wr_.buf_size = wr_buf_size_;
 481         wr_.buf.reset();
 482     }
 483 }
 484
 485 template<class DynamicBuffer>
 486 void
 487 stream_base::
 488 write_close(DynamicBuffer& db, close_reason const& cr)
 489 {
 490     using namespace boost::endian;
 491     frame_header fh;
 492     fh.op = opcode::close;
 493     fh.fin = true;
 494     fh.rsv1 = false;
 495     fh.rsv2 = false;
 496     fh.rsv3 = false;
 497     fh.len = cr.code == close_code::none ?
 498         0 : 2 + cr.reason.size();
 499     fh.mask = role_ == detail::role_type::client;
 500     if(fh.mask)
 501         fh.key = maskgen_();
 502     detail::write(db, fh);
 503     if(cr.code != close_code::none)
 504     {
 505         detail::prepared_key key;
 506         if(fh.mask)
 507             detail::prepare_key(key, fh.key);
 508         {
 509             std::uint8_t b[2];
 510             ::new(&b[0]) big_uint16_buf_t{
 511                 (std::uint16_t)cr.code};
 512             auto d = db.prepare(2);
 513             boost::asio::buffer_copy(d,
 514                 boost::asio::buffer(b));
 515             if(fh.mask)
 516                 detail::mask_inplace(d, key);
 517             db.commit(2);
 518         }
 519         if(! cr.reason.empty())
 520         {
 521             auto d = db.prepare(cr.reason.size());
 522             boost::asio::buffer_copy(d,
 523                 boost::asio::const_buffer(
 524                     cr.reason.data(), cr.reason.size()));
 525             if(fh.mask)
 526                 detail::mask_inplace(d, key);
 527             db.commit(cr.reason.size());
 528         }
 529     }
 530 }
 531
 532 template<class DynamicBuffer>
 533 void
 534 stream_base::
 535 write_ping(
 536     DynamicBuffer& db, opcode op, ping_data const& data)
 537 {
 538     frame_header fh;
 539     fh.op = op;
 540     fh.fin = true;
 541     fh.rsv1 = false;
 542     fh.rsv2 = false;
 543     fh.rsv3 = false;
 544     fh.len = data.size();
 545     fh.mask = role_ == role_type::client;
 546     if(fh.mask)
 547         fh.key = maskgen_();
 548     detail::write(db, fh);
 549     if(data.empty())
 550         return;
 551     detail::prepared_key key;
 552     if(fh.mask)
 553         detail::prepare_key(key, fh.key);
 554     auto d = db.prepare(data.size());
 555     boost::asio::buffer_copy(d,
 556         boost::asio::const_buffers_1(
 557             data.data(), data.size()));
 558     if(fh.mask)
 559         detail::mask_inplace(d, key);
 560     db.commit(data.size());
 561 }
 562
 563 } // detail
 564 } // websocket
 565 } // beast
 566
 567 #endif