[FEAT MERGE] [CP] query_interface 422

deps/oblib/src/rpc/CMakeLists.txt

@@ -56,6 +56,7 @@ ob_set_subtarget(oblib_rpc obmysql_packet
   obmysql/packet/ompk_resheader.cpp
   obmysql/packet/ompk_row.cpp
   obmysql/packet/ompk_ssl_request.cpp
+  obmysql/packet/ompk_auth_switch.cpp
 )
 
 ob_set_subtarget(oblib_rpc obrpc

deps/oblib/src/rpc/ob_packet.h

@@ -36,7 +36,8 @@ enum class ConnectionPhaseEnum
 {
   CPE_CONNECTED = 0,//server will send handshake pkt
   CPE_SSL_CONNECT,  //server will do ssl connect
-  CPE_AUTHED        //server will do auth check and send ok pkt
+  CPE_AUTHED,       //server will do auth check and send ok pkt
+  CPE_AUTH_SWITCH,  //server will do auth switch check and send ok pkt
 };
 
 

deps/oblib/src/rpc/obmysql/ob_2_0_protocol_processor.cpp

@@ -68,6 +68,7 @@ int Ob20ProtocolProcessor::do_decode(ObSMConnection& conn, ObICSMemPool& pool, c
   const uint32_t sessid = conn.sessid_;
   // together with mysql compress header, all treat as packet header
   const int64_t header_size = OB20_PROTOCOL_HEADER_LENGTH + OB_MYSQL_COMPRESSED_HEADER_SIZE;
+  conn.mysql_pkt_context_.is_auth_switch_ = conn.is_in_auth_switch_phase();
 
   // no need duplicated check 'm' valid, ObMySQLHandler::process() has already checked
   if ((end - start) >= header_size) {

deps/oblib/src/rpc/obmysql/ob_mysql_compress_protocol_processor.cpp

@@ -34,6 +34,7 @@ int ObMysqlCompressProtocolProcessor::do_decode(ObSMConnection& conn, ObICSMemPo
   pkt = NULL;
   const uint32_t sessid = conn.sessid_;
   const int64_t header_size = OB_MYSQL_COMPRESSED_HEADER_SIZE;
+  conn.mysql_pkt_context_.is_auth_switch_ = conn.is_in_auth_switch_phase();
   // no need duplicated check 'm' valid, ObMySQLHandler::process() has already checked
   if ((end - start) >= header_size) {
     //1. decode length from net buffer

deps/oblib/src/rpc/obmysql/ob_mysql_packet.h

@@ -84,6 +84,7 @@ enum ObMySQLCmd
   // COM_LOGIN represents client---->hand shake response && observer---> ok or error
   COM_HANDSHAKE,
   COM_LOGIN,
+  COM_AUTH_SWITCH_RESPONSE,
 
   COM_STMT_PREXECUTE = PREXECUTE_CMD,
   COM_STMT_SEND_PIECE_DATA,
@@ -106,6 +107,7 @@ enum class ObMySQLPacketType
   PKT_RESHEAD,   // 10 -> result header packet
   PKT_PREXEC,    // 11 -> prepare execute packet;
   PKT_FILENAME,  // 12 -> send file name to client(load local infile)
+  PKT_AUTH_SWITCH,// 13 -> auth switch request packet;
   PKT_END        // 13 -> end of packet type
 };
 

deps/oblib/src/rpc/obmysql/ob_mysql_protocol_processor.cpp

@@ -35,6 +35,7 @@ int ObMysqlProtocolProcessor::do_decode(ObSMConnection& conn, ObICSMemPool& pool
   const uint32_t sessid = conn.sessid_;
   const int64_t header_size = OB_MYSQL_HEADER_LENGTH;
   // no need duplicated check 'm' valid, ObMySQLHandler::process() has already checked
+  conn.mysql_pkt_context_.is_auth_switch_ = conn.is_in_auth_switch_phase();
   if ((end - start) >= header_size) {
     // 1. decode length from net buffer
     // 2. decode seq from net buffer
@@ -469,8 +470,12 @@ int ObMysqlProtocolProcessor::process_one_mysql_packet(
       }
       if (OB_SUCC(ret)) {
         uint8_t cmd = 0;
-        ObMySQLUtil::get_uint1(payload, cmd);
-        raw_pkt->set_cmd(static_cast<ObMySQLCmd>(cmd));
+        if (context.is_auth_switch_) {
+          raw_pkt->set_cmd(ObMySQLCmd::COM_AUTH_SWITCH_RESPONSE);
+        } else {
+          ObMySQLUtil::get_uint1(payload, cmd);
+          raw_pkt->set_cmd(static_cast<ObMySQLCmd>(cmd));
+        }
         raw_pkt->set_content(payload, static_cast<uint32_t>(total_data_len));
         // no need set seq again
         need_decode_more = false;

deps/oblib/src/rpc/obmysql/ob_mysql_request_utils.h

@@ -70,6 +70,7 @@ public:
     next_read_step_ = READ_HEADER;
     raw_pkt_.reset();
     is_multi_pkt_ = false;
+    is_auth_switch_ = false;
     arena_.reset(); //fast free memory
   }
 
@@ -92,7 +93,7 @@ public:
   TO_STRING_KV(K_(header_buffered_len), K_(payload_buffered_len), K_(payload_buffered_total_len),
                K_(last_pkt_seq), K_(payload_len), K_(curr_pkt_seq), K_(payload_buf_alloc_len),
                "next_read_step", get_read_step_str(next_read_step_), K_(raw_pkt),
-               "used", arena_.used(), "total", arena_.total(), K_(is_multi_pkt));
+               "used", arena_.used(), "total", arena_.total(), K_(is_multi_pkt), K_(is_auth_switch));
 
 public:
   char header_buf_[common::OB_MYSQL_HEADER_LENGTH];
@@ -108,6 +109,7 @@ public:
   ObMySQLRawPacket raw_pkt_;
   bool is_multi_pkt_;
   common::ObArenaAllocator arena_;
+  bool is_auth_switch_;
 
 private:
   DISALLOW_COPY_AND_ASSIGN(ObMysqlPktContext);

deps/oblib/src/rpc/obmysql/ob_sql_sock_processor.cpp

@@ -70,7 +70,7 @@ int ObSqlSockProcessor::decode_sql_packet(ObICSMemPool& mem_pool,
         LOG_WARN("sql nio enable ssl for server failed", K(ret));
       }
       break;
-    } else if (!conn.is_in_authed_phase()) {
+    } else if (!conn.is_in_authed_phase() && !conn.is_in_auth_switch_phase()) {
       ret_pkt = pkt;
       sess.set_last_pkt_sz(consume_sz);
     } else if (OB_FAIL(processor->do_splice(conn, mem_pool, (void*&)pkt, need_read_more))) {

deps/oblib/src/rpc/obmysql/ob_virtual_cs_protocol_processor.cpp

@@ -51,7 +51,7 @@ int ObVirtualCSProtocolProcessor::easy_process(easy_request_t *r, bool &need_rea
   ObSMConnection *conn = reinterpret_cast<ObSMConnection*>(r->ms->c->user_data);
   ObCSEasyMemPool pool(r->ms->pool);
   need_read_more = true;
-  if (!conn->is_in_authed_phase()) {
+  if (!conn->is_in_authed_phase() && !conn->is_in_auth_switch_phase()) {
     need_read_more = false;
   } else if (OB_FAIL(do_splice(*conn, pool, r->ipacket, need_read_more))) {
     LOG_ERROR("fail to splice mysql packet", K(ret));

deps/oblib/src/rpc/obmysql/obsm_struct.h

@@ -152,9 +152,15 @@ public:
     return type;
   }
 
+  bool is_support_plugin_auth() const {
+    return (1 == cap_flags_.cap_flags_.OB_CLIENT_PLUGIN_AUTH);
+  }
+
   inline bool is_in_connected_phase() { return rpc::ConnectionPhaseEnum::CPE_CONNECTED == connection_phase_; }
   inline bool is_in_ssl_connect_phase() { return rpc::ConnectionPhaseEnum::CPE_SSL_CONNECT == connection_phase_; }
   inline bool is_in_authed_phase() { return rpc::ConnectionPhaseEnum::CPE_AUTHED == connection_phase_; }
+  inline bool is_in_auth_switch_phase() { return rpc::ConnectionPhaseEnum::CPE_AUTH_SWITCH == connection_phase_; }
+  inline void set_auth_switch_phase() { connection_phase_ = rpc::ConnectionPhaseEnum::CPE_AUTH_SWITCH; }
   inline void set_ssl_connect_phase() { connection_phase_ = rpc::ConnectionPhaseEnum::CPE_SSL_CONNECT; }
   inline void set_auth_phase() { connection_phase_ = rpc::ConnectionPhaseEnum::CPE_AUTHED; }
   inline void set_connect_phase() { connection_phase_ = rpc::ConnectionPhaseEnum::CPE_CONNECTED; }

deps/oblib/src/rpc/obmysql/packet/ompk_auth_switch.cpp

@@ -0,0 +1,70 @@
+/**
+ * Copyright (c) 2021 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#define USING_LOG_PREFIX RPC_OBMYSQL
+
+#include "rpc/obmysql/packet/ompk_auth_switch.h"
+
+#include "rpc/obmysql/ob_mysql_util.h"
+
+using namespace oceanbase::common;
+using namespace oceanbase::obmysql;
+
+OMPKAuthSwitch::OMPKAuthSwitch()
+    : status_(0xfe),
+      plugin_name_(0),
+      scramble_()
+{}
+
+int OMPKAuthSwitch::serialize(char *buffer, int64_t len, int64_t &pos) const
+{
+  int ret = OB_SUCCESS;
+
+  if (OB_ISNULL(buffer) || OB_UNLIKELY(len - pos < 0)) {
+    ret = OB_INVALID_ARGUMENT;
+    LOG_WARN("invalid argument", KP(buffer), K(len), K(pos), K(ret));
+  } else if (OB_UNLIKELY(len - pos < static_cast<int64_t>(get_serialize_size()))) {
+    ret = OB_SIZE_OVERFLOW;
+    LOG_WARN("invalid argument", K(len), K(pos), "need_size", get_serialize_size());
+  } else {
+    if (OB_FAIL(ObMySQLUtil::store_int1(buffer, len, status_, pos))) {
+      LOG_WARN("store fail", KP(buffer), K(len), K(pos), K(ret));
+    } else if (OB_FAIL(ObMySQLUtil::store_obstr_zt(buffer, len, plugin_name_, pos))) {
+      LOG_WARN("store fail", KP(buffer), K(len), K(pos), K(ret));
+    } else if (OB_FAIL(ObMySQLUtil::store_obstr_nzt(buffer, len, scramble_, pos))) {
+      LOG_WARN("store fail", KP(buffer), K(len), K(pos), K(ret));
+    }
+  }
+
+  return ret;
+}
+
+int64_t OMPKAuthSwitch::get_serialize_size() const
+{
+  int64_t len = 0;
+  len += 1;                 // field_count_
+  len += plugin_name_.length() + 1;
+  len += scramble_.length() + 1;
+  return len;
+}
+
+int64_t OMPKAuthSwitch::to_string(char *buf, const int64_t buf_len) const
+{
+  int64_t pos = 0;
+  J_OBJ_START();
+  J_KV("header", hdr_,
+       K_(status),
+       K_(plugin_name),
+       K_(scramble));
+  J_OBJ_END();
+  return pos;
+}

deps/oblib/src/rpc/obmysql/packet/ompk_auth_switch.h

@@ -0,0 +1,55 @@
+/**
+ * Copyright (c) 2021 OceanBase
+ * OceanBase CE is licensed under Mulan PubL v2.
+ * You can use this software according to the terms and conditions of the Mulan PubL v2.
+ * You may obtain a copy of Mulan PubL v2 at:
+ *          http://license.coscl.org.cn/MulanPubL-2.0
+ * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+ * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+ * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+ * See the Mulan PubL v2 for more details.
+ */
+
+#ifndef _OMPK_AUTH_SWITCH_H_
+#define _OMPK_AUTH_SWITCH_H_
+
+#include "lib/string/ob_string.h"
+#include "rpc/obmysql/ob_mysql_packet.h"
+#include "lib/container/ob_se_array.h"
+
+namespace oceanbase
+{
+namespace obmysql
+{
+
+class OMPKAuthSwitch : public ObMySQLPacket
+{
+public:
+  OMPKAuthSwitch();
+  virtual ~OMPKAuthSwitch() {}
+
+  // serialize all data into thread buffer not include packet header
+  // Attention!! before called serialize or get_serialize_size, must set capability
+  virtual int serialize(char *buffer, const int64_t length, int64_t &pos) const;
+  virtual int64_t get_serialize_size() const;
+
+  // shadow copy
+  void set_plugin_name(const common::ObString &plugin_name) { plugin_name_ = plugin_name; }
+  void set_scramble(const common::ObString &scramble) { scramble_ = scramble; }
+
+  inline const common::ObString &get_plugin_name() const { return plugin_name_; }
+  inline const common::ObString &get_scramble() const { return scramble_; };
+  inline ObMySQLPacketType get_mysql_packet_type() { return ObMySQLPacketType::PKT_AUTH_SWITCH; }
+
+  virtual int64_t to_string(char *buf, const int64_t buf_len) const;
+private:
+  DISALLOW_COPY_AND_ASSIGN(OMPKAuthSwitch);
+
+  uint8_t status_;    // always 0xfe
+  common::ObString plugin_name_;
+  common::ObString scramble_;
+};
+
+} // end namespace obmysql
+} // end namespace oceanbase
+#endif /* _OMPK_AUTH_SWITCH_H_ */