mysql_filter: add handling for partial messages

source/extensions/filters/network/mysql_proxy/mysql_codec.h

@@ -99,7 +99,7 @@ class MySQLCodec : public Logger::Loggable<Logger::Id::filter> {
 
   PACKED_STRUCT(struct header_fields {
     uint32_t length_ : 24;
-    uint32_t seq_ : 8;
+    uint8_t seq_ : 8;
   });
 
   union MySQLHeader {
@@ -109,20 +109,17 @@ class MySQLCodec : public Logger::Loggable<Logger::Id::filter> {
 
   virtual ~MySQLCodec() {}
 
-  int decode(Buffer::Instance& data, uint64_t& offset, int seq, int len) {
+  int decode(Buffer::Instance& data, uint8_t seq, uint32_t len) {
     seq_ = seq;
-    const uint64_t prev_offset = offset;
-    int result = parseMessage(data, offset, len);
-    offset = prev_offset + len; // Ensure that the whole message was consumed
-    return result;
+    return parseMessage(data, len);
   }
 
   virtual std::string encode() PURE;
 
 protected:
-  virtual int parseMessage(Buffer::Instance& data, uint64_t& offset, int len) PURE;
+  virtual int parseMessage(Buffer::Instance& data, uint32_t len) PURE;
 
-  int seq_;
+  uint8_t seq_;
 };
 
 } // namespace MySQLProxy

source/extensions/filters/network/mysql_proxy/mysql_codec_clogin.cc

@@ -44,21 +44,21 @@ bool ClientLogin::isClientSecureConnection() const {
   return extended_client_cap_ & MYSQL_EXT_CL_SECURE_CONNECTION;
 }
 
-int ClientLogin::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int) {
+int ClientLogin::parseMessage(Buffer::Instance& buffer, uint32_t) {
   uint16_t client_cap = 0;
-  if (BufferHelper::peekUint16(buffer, offset, client_cap) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, client_cap) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing client_cap in mysql ClientLogin msg");
     return MYSQL_FAILURE;
   }
   setClientCap(client_cap);
   uint16_t extended_client_cap = 0;
-  if (BufferHelper::peekUint16(buffer, offset, extended_client_cap) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, extended_client_cap) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing extended_client_cap in mysql ClientLogin msg");
     return MYSQL_FAILURE;
   }
   setExtendedClientCap(extended_client_cap);
   uint32_t max_packet = 0;
-  if (BufferHelper::peekUint32(buffer, offset, max_packet) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint32(buffer, max_packet) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing max_packet in mysql ClientLogin msg");
     return MYSQL_FAILURE;
   }
@@ -68,52 +68,52 @@ int ClientLogin::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int) {
     return MYSQL_SUCCESS;
   }
   uint8_t charset = 0;
-  if (BufferHelper::peekUint8(buffer, offset, charset) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(buffer, charset) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing charset in mysql ClientLogin msg");
     return MYSQL_FAILURE;
   }
   setCharset(charset);
-  if (BufferHelper::peekBytes(buffer, offset, UNSET_BYTES) != MYSQL_SUCCESS) {
+  if (BufferHelper::readBytes(buffer, UNSET_BYTES) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error skipping unset bytes in mysql ClientLogin msg");
     return MYSQL_FAILURE;
   }
   std::string username;
-  if (BufferHelper::peekString(buffer, offset, username) != MYSQL_SUCCESS) {
+  if (BufferHelper::readString(buffer, username) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing username in mysql ClientLogin msg");
     return MYSQL_FAILURE;
   }
   setUsername(username);
   std::string auth_resp;
   if (isClientAuthLenClData()) {
     uint64_t auth_resp_len = 0;
-    if (BufferHelper::peekLengthEncodedInteger(buffer, offset, auth_resp_len) != MYSQL_SUCCESS) {
+    if (BufferHelper::readLengthEncodedInteger(buffer, auth_resp_len) != MYSQL_SUCCESS) {
       ENVOY_LOG(info, "error parsing LengthEncodedInteger in mysql ClientLogin msg");
       return MYSQL_FAILURE;
     }
-    if (BufferHelper::peekStringBySize(buffer, offset, auth_resp_len, auth_resp) != MYSQL_SUCCESS) {
+    if (BufferHelper::readStringBySize(buffer, auth_resp_len, auth_resp) != MYSQL_SUCCESS) {
       ENVOY_LOG(info, "error parsing auth_resp in mysql ClientLogin msg");
       return MYSQL_FAILURE;
     }
   } else if (isClientSecureConnection()) {
     uint8_t auth_resp_len = 0;
-    if (BufferHelper::peekUint8(buffer, offset, auth_resp_len) != MYSQL_SUCCESS) {
+    if (BufferHelper::readUint8(buffer, auth_resp_len) != MYSQL_SUCCESS) {
       ENVOY_LOG(info, "error parsing auth_resp_len in mysql ClientLogin msg");
       return MYSQL_FAILURE;
     }
-    if (BufferHelper::peekStringBySize(buffer, offset, auth_resp_len, auth_resp) != MYSQL_SUCCESS) {
+    if (BufferHelper::readStringBySize(buffer, auth_resp_len, auth_resp) != MYSQL_SUCCESS) {
       ENVOY_LOG(info, "error parsing auth_resp in mysql ClientLogin msg");
       return MYSQL_FAILURE;
     }
   } else {
-    if (BufferHelper::peekString(buffer, offset, auth_resp) != MYSQL_SUCCESS) {
+    if (BufferHelper::readString(buffer, auth_resp) != MYSQL_SUCCESS) {
       ENVOY_LOG(info, "error parsing auth_resp in mysql ClientLogin msg");
       return MYSQL_FAILURE;
     }
   }
   setAuthResp(auth_resp);
   if (isConnectWithDb()) {
     std::string db;
-    if (BufferHelper::peekString(buffer, offset, db) != MYSQL_SUCCESS) {
+    if (BufferHelper::readString(buffer, db) != MYSQL_SUCCESS) {
       ENVOY_LOG(info, "error parsing auth_resp in mysql ClientLogin msg");
       return MYSQL_FAILURE;
     }

source/extensions/filters/network/mysql_proxy/mysql_codec_clogin.h

@@ -12,7 +12,7 @@ constexpr int UNSET_BYTES = 23;
 class ClientLogin : public MySQLCodec {
 public:
   // MySQLCodec
-  int parseMessage(Buffer::Instance& buffer, uint64_t& offset, int len) override;
+  int parseMessage(Buffer::Instance& buffer, uint32_t len) override;
   std::string encode() override;
 
   int getClientCap() const { return client_cap_; }

source/extensions/filters/network/mysql_proxy/mysql_codec_clogin_resp.cc

@@ -20,37 +20,37 @@ void ClientLoginResponse::setServerStatus(uint16_t status) { server_status_ = st
 
 void ClientLoginResponse::setWarnings(uint16_t warnings) { warnings_ = warnings; }
 
-int ClientLoginResponse::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int) {
+int ClientLoginResponse::parseMessage(Buffer::Instance& buffer, uint32_t) {
   uint8_t resp_code = 0;
-  if (BufferHelper::peekUint8(buffer, offset, resp_code) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(buffer, resp_code) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing response code in mysql Login Ok msg");
     return MYSQL_FAILURE;
   }
   setRespCode(resp_code);
-  if ((resp_code == MYSQL_RESP_AUTH_SWITCH) && BufferHelper::endOfBuffer(buffer, offset)) {
+  if ((resp_code == MYSQL_RESP_AUTH_SWITCH) && BufferHelper::endOfBuffer(buffer)) {
     // OldAuthSwitchRequest
     return MYSQL_SUCCESS;
   }
   uint8_t affected_rows = 0;
-  if (BufferHelper::peekUint8(buffer, offset, affected_rows) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(buffer, affected_rows) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing affected_rows in mysql Login Ok msg");
     return MYSQL_FAILURE;
   }
   setAffectedRows(affected_rows);
   uint8_t last_insert_id = 0;
-  if (BufferHelper::peekUint8(buffer, offset, last_insert_id) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(buffer, last_insert_id) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing last_insert_id in mysql Login Ok msg");
     return MYSQL_FAILURE;
   }
   setLastInsertId(last_insert_id);
   uint16_t server_status = 0;
-  if (BufferHelper::peekUint16(buffer, offset, server_status) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, server_status) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing server_status in mysql Login Ok msg");
     return MYSQL_FAILURE;
   }
   setServerStatus(server_status);
   uint16_t warnings = 0;
-  if (BufferHelper::peekUint16(buffer, offset, warnings) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, warnings) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing warnings in mysql Login Ok msg");
     return MYSQL_FAILURE;
   }

source/extensions/filters/network/mysql_proxy/mysql_codec_clogin_resp.h

@@ -13,7 +13,7 @@ namespace MySQLProxy {
 class ClientLoginResponse : public MySQLCodec {
 public:
   // MySQLCodec
-  int parseMessage(Buffer::Instance& buffer, uint64_t& offset, int len) override;
+  int parseMessage(Buffer::Instance& buffer, uint32_t len) override;
   std::string encode() override;
 
   uint8_t getRespCode() const { return resp_code_; }

source/extensions/filters/network/mysql_proxy/mysql_codec_command.cc

@@ -8,9 +8,9 @@ namespace Extensions {
 namespace NetworkFilters {
 namespace MySQLProxy {
 
-Command::Cmd Command::parseCmd(Buffer::Instance& data, uint64_t& offset) {
+Command::Cmd Command::parseCmd(Buffer::Instance& data) {
   uint8_t cmd;
-  if (BufferHelper::peekUint8(data, offset, cmd) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(data, cmd) != MYSQL_SUCCESS) {
     return Command::Cmd::COM_NULL;
   }
   return static_cast<Command::Cmd>(cmd);
@@ -20,8 +20,8 @@ void Command::setCmd(Command::Cmd cmd) { cmd_ = cmd; }
 
 void Command::setDb(std::string db) { db_ = db; }
 
-int Command::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int len) {
-  Command::Cmd cmd = parseCmd(buffer, offset);
+int Command::parseMessage(Buffer::Instance& buffer, uint32_t len) {
+  Command::Cmd cmd = parseCmd(buffer);
   setCmd(cmd);
   if (cmd == Command::Cmd::COM_NULL) {
     return MYSQL_FAILURE;
@@ -32,15 +32,15 @@ int Command::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int len) {
   case Command::Cmd::COM_CREATE_DB:
   case Command::Cmd::COM_DROP_DB: {
     std::string db = "";
-    BufferHelper::peekStringBySize(buffer, offset, len - 1, db);
+    BufferHelper::readStringBySize(buffer, len - 1, db);
     setDb(db);
     break;
   }
 
   case Command::Cmd::COM_QUERY:
     is_query_ = true;
     // query string starts after one byte for comm type
-    BufferHelper::peekStringBySize(buffer, offset, len - 1, data_);
+    BufferHelper::readStringBySize(buffer, len - 1, data_);
     setDb("");
     break;
 

source/extensions/filters/network/mysql_proxy/mysql_codec_command.h

@@ -35,10 +35,10 @@ class Command : public MySQLCodec {
   };
 
   // MySQLCodec
-  int parseMessage(Buffer::Instance&, uint64_t& offset, int len) override;
+  int parseMessage(Buffer::Instance&, uint32_t len) override;
   std::string encode() override;
 
-  Cmd parseCmd(Buffer::Instance& data, uint64_t& offset);
+  Cmd parseCmd(Buffer::Instance& data);
   Cmd getCmd() const { return cmd_; }
   const std::string& getData() const { return data_; }
   std::string& getDb() { return db_; }
@@ -57,7 +57,7 @@ class Command : public MySQLCodec {
 class CommandResponse : public MySQLCodec {
 public:
   // MySQLCodec
-  int parseMessage(Buffer::Instance&, uint64_t&, int) override { return MYSQL_SUCCESS; }
+  int parseMessage(Buffer::Instance&, uint32_t) override { return MYSQL_SUCCESS; }
   std::string encode() override { return ""; }
 
   uint16_t getServerStatus() const { return server_status_; }

source/extensions/filters/network/mysql_proxy/mysql_codec_greeting.cc

@@ -24,27 +24,27 @@ void ServerGreeting::setServerStatus(int server_status) { server_status_ = serve
 
 void ServerGreeting::setExtServerCap(int ext_server_cap) { ext_server_cap_ = ext_server_cap; }
 
-int ServerGreeting::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int) {
+int ServerGreeting::parseMessage(Buffer::Instance& buffer, uint32_t) {
   uint8_t protocol = 0;
-  if (BufferHelper::peekUint8(buffer, offset, protocol) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(buffer, protocol) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing protocol in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
   setProtocol(protocol);
   std::string version;
-  if (BufferHelper::peekString(buffer, offset, version) != MYSQL_SUCCESS) {
+  if (BufferHelper::readString(buffer, version) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing version in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
   setVersion(version);
   uint32_t thread_id = 0;
-  if (BufferHelper::peekUint32(buffer, offset, thread_id) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint32(buffer, thread_id) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing thread_id in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
   setThreadId(thread_id);
   std::string salt;
-  if (BufferHelper::peekString(buffer, offset, salt) != MYSQL_SUCCESS) {
+  if (BufferHelper::readString(buffer, salt) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing salt in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
@@ -54,29 +54,29 @@ int ServerGreeting::parseMessage(Buffer::Instance& buffer, uint64_t& offset, int
     return MYSQL_SUCCESS;
   }
   uint16_t server_cap = 0;
-  if (BufferHelper::peekUint16(buffer, offset, server_cap) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, server_cap) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing server_cap in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
   setServerCap(server_cap);
-  if (BufferHelper::endOfBuffer(buffer, offset) == true) {
+  if (BufferHelper::endOfBuffer(buffer)) {
     // HandshakeV10 can terminate after Server Capabilities
     return MYSQL_SUCCESS;
   }
   uint8_t server_language = 0;
-  if (BufferHelper::peekUint8(buffer, offset, server_language) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint8(buffer, server_language) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing server_language in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
   setServerLanguage(server_language);
   uint16_t server_status = 0;
-  if (BufferHelper::peekUint16(buffer, offset, server_status) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, server_status) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing server_status in mysql Greeting msg");
     return MYSQL_FAILURE;
   }
   setServerStatus(server_status);
   uint16_t ext_server_cap = 0;
-  if (BufferHelper::peekUint16(buffer, offset, ext_server_cap) != MYSQL_SUCCESS) {
+  if (BufferHelper::readUint16(buffer, ext_server_cap) != MYSQL_SUCCESS) {
     ENVOY_LOG(info, "error parsing ext_server_cap in mysql Greeting msg");
     return MYSQL_FAILURE;
   }

source/extensions/filters/network/mysql_proxy/mysql_codec_greeting.h

@@ -11,7 +11,7 @@ namespace MySQLProxy {
 class ServerGreeting : public MySQLCodec {
 public:
   // MySQLCodec
-  int parseMessage(Buffer::Instance& buffer, uint64_t& offset, int len) override;
+  int parseMessage(Buffer::Instance& buffer, uint32_t len) override;
   std::string encode() override;
 
   int getProtocol() const { return protocol_; }

source/extensions/filters/network/mysql_proxy/mysql_codec_switch_resp.cc

@@ -12,7 +12,7 @@ void ClientSwitchResponse::setAuthPluginResp(std::string& auth_plugin_resp) {
   auth_plugin_resp_.assign(auth_plugin_resp);
 }
 
-int ClientSwitchResponse::parseMessage(Buffer::Instance&, uint64_t&, int) { return MYSQL_SUCCESS; }
+int ClientSwitchResponse::parseMessage(Buffer::Instance&, uint32_t) { return MYSQL_SUCCESS; }
 
 std::string ClientSwitchResponse::encode() {
   Buffer::InstancePtr buffer(new Buffer::OwnedImpl());

source/extensions/filters/network/mysql_proxy/mysql_codec_switch_resp.h

@@ -11,7 +11,7 @@ namespace MySQLProxy {
 class ClientSwitchResponse : public MySQLCodec {
 public:
   // MySQLCodec
-  int parseMessage(Buffer::Instance& buffer, uint64_t& offset, int len) override;
+  int parseMessage(Buffer::Instance& buffer, uint32_t len) override;
   std::string encode() override;
 
   const std::string& getAuthPluginResp() const { return auth_plugin_resp_; }