release300.2.5

Author: dolphindb-jffu

include/DolphinDB.h

@@ -33,6 +33,12 @@
 #pragma warning( disable : 4251 )
 #endif
 
+#ifdef USE_OPENSSL
+constexpr bool HAS_OPENSSL{true};
+#else
+constexpr bool HAS_OPENSSL{false};
+#endif
+
 namespace dolphindb {
 
 class DBConnectionImpl;
@@ -95,7 +101,7 @@ class EXPORT_DECL DBConnection {
             throw RuntimeException("Failed to connect to server: " + host + ":" + std::to_string(port));
         }
         if (!userName.empty()) {
-            login(userName, password, true);
+            login(userName, password, HAS_OPENSSL);
         }
     }
     DBConnection(bool enableSSL = false, bool asyncTask = false, int keepAliveTime = 7200, bool compress = false, bool python = false, bool isReverseStreaming = false, bool enableSCRAM = false);

include/SysIO.h

@@ -19,11 +19,6 @@
 #include <openssl/err.h>
 #include <openssl/ssl.h>
 
-#else
-
-using SSL_CTX = int;
-using SSL = int;
-
 #endif
 
 #ifdef __linux__
@@ -75,19 +70,19 @@ class EXPORT_DECL Socket{
 	bool setBlocking();
 	bool setTcpNoDelay();
 	int getErrorCode();
-	SSL_CTX* initCTX();
+	void* initCTX();
 	bool sslInit();
 	IO_ERR sslConnect();
-	void showCerts(SSL* ssl);
+	void showCerts(void* ssl);
 private:
 	std::string host_;
 	int port_;
 	SOCKET handle_;
 	bool blocking_;
 	bool autoClose_;
 	bool enableSSL_;
-	SSL_CTX* ctx_;
-	SSL* ssl_;
+	void* ctx_;
+	void* ssl_;
 	int keepAliveTime_;
 };
 

src/Crypto.cpp

@@ -45,7 +45,7 @@ auto Crypto::RSAEncrypt(const std::string &text) const -> std::string
 {
     int rsa_size = EVP_PKEY_size(rsa_);
     std::vector<unsigned char> encrypted(rsa_size);
-    size_t encrypted_len;
+    size_t encrypted_len = rsa_size;
     if (EVP_PKEY_encrypt(ctx_, encrypted.data(), &encrypted_len, (const unsigned char*)text.c_str(), text.size()) != 1) {
         printOpenSSLError();
         throw RuntimeException("Failed to encrypt userId or password.");

src/DolphinDBImp.cpp

@@ -115,7 +115,7 @@ bool DBConnectionImpl::connect() {
 
     if (!userId_.empty()) {
         try {
-            login(userId_, pwd_, false);
+            login(userId_, pwd_, HAS_OPENSSL);
         } catch (...) {
             close();
             throw;

src/Streaming.cpp

@@ -1402,7 +1402,7 @@ void StreamingClientImpl::sendPublishRequest(DBConnection &conn, SubscribeParam
 			info.offset_, info.filter_, info.allowExists_);
 	}
 	else {
-		conn.login(info.userName_, info.password_, false);
+		conn.login(info.userName_, info.password_, HAS_OPENSSL);
 		re = run(conn, "publishTable", getLocalIP(), listeningPort_, info.info_.tableName, info.info_.actionName,
 			info.offset_, info.filter_, info.allowExists_);
 	}
@@ -1421,7 +1421,7 @@ void StreamingClientImpl::sendPublishRequest(DBConnection &conn, SubscribeParam
 
 string StreamingClientImpl::subscribeInternal(DBConnection &conn, SubscribeParam &info) {
 	if (info.userName_.empty() == false)
-		conn.login(info.userName_, info.password_, false);
+		conn.login(info.userName_, info.password_, HAS_OPENSSL);
 	ConstantSP result = run(conn, "getSubscriptionTopic", info.info_.tableName, info.info_.actionName);
 	auto topic = result->get(0)->getString();
 	ConstantSP colLabels = result->get(1);
@@ -2029,7 +2029,7 @@ ThreadSP ThreadedClient::subscribe(string host, int port, const MessageHandler &
         }
         SubscribeConfig config { offset, msgAsTable, allowExists, false };
         if (!userName.empty()) {
-            conn->login(userName, password, false);
+            conn->login(userName, password, HAS_OPENSSL);
         }
         udpImpl_->subscribe(info, config, conn, handler, blobDeserializer);
         return nullptr;

src/SysIO.cpp

@@ -118,25 +118,23 @@ IO_ERR Socket::read(char* buffer, size_t length, size_t& actualLength, bool msgP
 	//RecordTime record("Socket.read");
 	if (!enableSSL_) {
 #ifdef _WIN32
-		actualLength = recv(handle_, buffer, static_cast<int>(length), msgPeek ? MSG_PEEK : 0);
-		RECORD_READ(buffer, actualLength);
-		if (actualLength < 0) {
+		actualLength = 0;
+		int ret = recv(handle_, buffer, static_cast<int>(length), msgPeek ? MSG_PEEK : 0);
+		if (ret == SOCKET_ERROR) {
 			DLogger::Error("socket read error", actualLength);
-		}
-		if (actualLength == 0)
-			return DISCONNECTED;
-		else if (actualLength != (size_t)SOCKET_ERROR)
-			return OK;
-		else {
-			actualLength = 0;
 			int error = WSAGetLastError();
 			if (error == WSAENOTCONN || error == WSAESHUTDOWN || error == WSAENETRESET)
 				return DISCONNECTED;
-			else if (error == WSAEWOULDBLOCK)
+			if (error == WSAEWOULDBLOCK)
 				return NODATA;
-			else
-				return OTHERERR;
+			return OTHERERR;
 		}
+		if (ret == 0) {
+			return DISCONNECTED;
+		}
+		actualLength = ret;
+		RECORD_READ(buffer, ret);
+		return OK;
 #else //Linux
 readdata:
 		actualLength = recv(handle_, (void*)buffer, length, (blocking_ ? 0 : MSG_DONTWAIT) | (msgPeek ? MSG_PEEK : 0));
@@ -156,20 +154,20 @@ IO_ERR Socket::read(char* buffer, size_t length, size_t& actualLength, bool msgP
 	} else {
 #ifdef USE_OPENSSL
 readdata2:
-        int ret = SSL_read(ssl_, buffer, static_cast<int>(length));
+        int ret = SSL_read((SSL*)ssl_, buffer, static_cast<int>(length));
         if (ret > 0) {
             actualLength = ret;
             RECORD_READ(buffer, ret);
         } else {
             DLogger::Error("socket read error", ret);
-            ret = SSL_get_error(ssl_, ret);
+            ret = SSL_get_error((SSL*)ssl_, ret);
             if(ret == SSL_ERROR_WANT_READ) goto readdata2;
             LOG_ERR("Socket(SSL)::read err =" + std::to_string(ret));
             return OTHERERR;
         }
 #endif
-        return OK;
     }
+    return OK;
 }
 
 IO_ERR Socket::write(const char* buffer, size_t length, size_t& actualLength){
@@ -218,12 +216,12 @@ IO_ERR Socket::write(const char* buffer, size_t length, size_t& actualLength){
 	else {
 #ifdef USE_OPENSSL
 		senddata2:
-        int ret = SSL_write(ssl_, (const void*)buffer, static_cast<int>(length));
+        int ret = SSL_write((SSL*)ssl_, (const void*)buffer, static_cast<int>(length));
         if (ret > 0) {
             actualLength = ret;
             RECORD_WRITE(buffer, ret);
         } else {
-            int err = SSL_get_error(ssl_, ret);
+            int err = SSL_get_error((SSL*)ssl_, ret);
             if (err == SSL_ERROR_WANT_WRITE) goto senddata2;
             DLogger::Error("socket write error", err);
             LOG_ERR("Socket(SSL)::write err =" + std::to_string(err));
@@ -387,10 +385,10 @@ IO_ERR Socket::close(){
 #ifdef USE_OPENSSL
 	if(ssl_ != nullptr) {
 		//shutdown until it done.
-		while (SSL_shutdown(ssl_) == 0) {
+		while (SSL_shutdown((SSL*)ssl_) == 0) {
 			Util::sleep(10);
 		}
-		SSL_free(ssl_);
+		SSL_free((SSL*)ssl_);
 		ssl_ = nullptr;
 	}
 #endif
@@ -407,7 +405,7 @@ IO_ERR Socket::close(){
 	}
 #ifdef USE_OPENSSL
 	if (ctx_ != nullptr) {
-		SSL_CTX_free(ctx_);
+		SSL_CTX_free((SSL_CTX*)ctx_);
 		ctx_ = nullptr;
 	}
 #endif
@@ -552,7 +550,7 @@ int Socket::getErrorCode(){
 }
 
 #ifdef USE_OPENSSL
-SSL_CTX* Socket::initCTX(){
+void* Socket::initCTX(){
     const SSL_METHOD* method;
     SSL_CTX* ctx;
 
@@ -576,11 +574,11 @@ bool Socket::sslInit() {
     if (ctx_ == nullptr) {
         return false;
     }
-    ssl_ = SSL_new(ctx_);
+    ssl_ = SSL_new((SSL_CTX*)ctx_);
     if (ssl_ == nullptr) {
         return false;
     }
-    SSL_set_fd(ssl_, static_cast<int>(handle_));
+    SSL_set_fd((SSL*)ssl_, static_cast<int>(handle_));
     return true;
 }
 
@@ -589,7 +587,7 @@ IO_ERR Socket::sslConnect() {
     if (!sslInit()) {
         return OTHERERR;
     }
-    if (SSL_connect(ssl_) == -1) {
+    if (SSL_connect((SSL*)ssl_) == -1) {
         if (!blocking_) {
             //TODO: solve error
         }
@@ -599,10 +597,10 @@ IO_ERR Socket::sslConnect() {
     return OK;
 }
 
-void Socket::showCerts(SSL *ssl) {
+void Socket::showCerts(void *ssl) {
     X509 *cert;
     char *line;
-    cert = SSL_get_peer_certificate(ssl); /* get the server's certificate */
+    cert = SSL_get_peer_certificate((SSL*)ssl); /* get the server's certificate */
     if ( cert != NULL ){
         std::cout << "Server certificates:\n";
         line = X509_NAME_oneline(X509_get_subject_name(cert), 0, 0);

src/Util.cpp

@@ -40,12 +40,12 @@ using std::vector;
 
 static SmartPointer<ConstantFactory> s_constFactory(new ConstantFactory());
 const bool Util::LITTLE_ENDIAN_ORDER = isLittleEndian();
-string Util::VER = "3.00.2.4";
+string Util::VER = "3.00.2.5";
 #ifndef _MSC_VER
 constexpr int Util::BUF_SIZE;
 #endif
-int Util::VERNUM = 30024;
-string Util::BUILD = "2025.03.11";
+int Util::VERNUM = 30025;
+string Util::BUILD = "2025.05.19";
 
 int Util::SEQUENCE_SEARCH_NUM_THRESHOLD = 10;
 int Util::MAX_LENGTH_FOR_ANY_VECTOR = 1048576;

test/DBConnectionTest_gtest.cpp

@@ -57,18 +57,10 @@ void StopCurNode(string cur_node)
 
     conn1.run("try{stopDataNode(\"" + cur_node + "\")}catch(ex){};");
     cout << cur_node + " has stopped..." << endl;
-    std::this_thread::sleep_for(std::chrono::seconds(5));
-    // std::this_thread::yield();
-    conn1.run("try{startDataNode(\"" + cur_node + "\")}catch(ex){};");
-    bool state = conn1.run("exec state from getClusterPerf() where name = `" + cur_node)->getBool();
-    int wait_time = 0;
-    while (!state && wait_time < 60){
-        conn1.run("try{startDataNode(\"" + cur_node + "\")}catch(ex){};");
-        state = conn1.run("exec state from getClusterPerf() where name = `" + cur_node)->getBool();
-        cout << "waiting for " + cur_node + " to start..." << endl;
-        Util::sleep(1000);
-        wait_time += 1;
-    }
+    Util::sleep(5000);
+    cout << "restarting " + cur_node + "..." << endl;
+    conn1.run("startDataNode(exec name from getClusterPerf() where mode !=1 and state != 1);go;"
+    "do{sleep(1000);}while((exec distinct state from getClusterPerf()).size() !=1);");
 }
 
 bool assertUnConnect()
@@ -856,6 +848,21 @@ TEST_F(DBConnectionTest, test_connection_SCRAM){
         _c.close();
         cout << "test_connection_SCRAM2 passed" << endl;
     }
+    { // login with no scram auth user admin
+        DBConnection _c = DBConnection(false, false, 7200, false, false, false, true);
+        _c.connect(hostName, port, "admin", "123456"); // stdout: [debug] [140591328847872] : user 'admin' doesn't support scram authMode.
+    }
+    { // enableEncryption with scram user
+        DBConnection _c = DBConnection(hostName, port);
+        _c.connect();
+        _c.login("scramUser", "123456", true);
+        ConstantSP res = _c.run("getCurrentSessionAndUser()[1]");
+        EXPECT_EQ(res->getString(), "scramUser");
+        _c.close();
+    }
+    { // async login with scram user
+        EXPECT_ANY_THROW(DBConnection(false, true, 7200, false, false, false, true));
+    }
 }
 
 TEST_F(DBConnectionTest, test_connectionPool_SCRAM){

test/StreamingSubscribeHAstreamTableTest_gtest.cpp

@@ -1409,6 +1409,7 @@ TEST_P(StreamingSubscribeHighAvailableTest, test_Pollingclient_stopDataNode)
                 if (msg1_total == insert_total_rows)
                 {
                     cout << "get all msg successfully!" << endl;
+                    break;
                 }
             }
         } };