MongoDB SSL Connection

src/Storages/StorageMongoDB.cpp

@@ -1,4 +1,5 @@
 #include "StorageMongoDB.h"
+#include "StorageMongoDBSocketFactory.h"
 
 #include <Poco/MongoDB/Connection.h>
 #include <Poco/MongoDB/Cursor.h>
@@ -33,6 +34,7 @@ StorageMongoDB::StorageMongoDB(
     const std::string & collection_name_,
     const std::string & username_,
     const std::string & password_,
+    const std::string & options_,
     const ColumnsDescription & columns_,
     const ConstraintsDescription & constraints_,
     const String & comment)
@@ -43,6 +45,8 @@ StorageMongoDB::StorageMongoDB(
     , collection_name(collection_name_)
     , username(username_)
     , password(password_)
+    , options(options_)
+    , uri("mongodb://" + username_ + ":" + password_ + "@" + host_ + ":" + std::to_string(port_) + "/" + database_name_ + "?" + options_)
 {
     StorageInMemoryMetadata storage_metadata;
     storage_metadata.setColumns(columns_);
@@ -56,7 +60,10 @@ void StorageMongoDB::connectIfNotConnected()
 {
     std::lock_guard lock{connection_mutex};
     if (!connection)
-        connection = std::make_shared<Poco::MongoDB::Connection>(host, port);
+    {
+        StorageMongoDBSocketFactory factory;
+        connection = std::make_shared<Poco::MongoDB::Connection>(uri, factory);
+    }
 
     if (!authenticated)
     {
@@ -102,9 +109,9 @@ void registerStorageMongoDB(StorageFactory & factory)
     {
         ASTs & engine_args = args.engine_args;
 
-        if (engine_args.size() != 5)
+        if (engine_args.size() < 5 || engine_args.size() > 6)
             throw Exception(
-                "Storage MongoDB requires 5 parameters: MongoDB('host:port', database, collection, 'user', 'password').",
+                "Storage MongoDB requires from 5 to 6 parameters: MongoDB('host:port', database, collection, 'user', 'password' [, 'options']).",
                 ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH);
 
         for (auto & engine_arg : engine_args)
@@ -118,6 +125,11 @@ void registerStorageMongoDB(StorageFactory & factory)
         const String & username = engine_args[3]->as<ASTLiteral &>().value.safeGet<String>();
         const String & password = engine_args[4]->as<ASTLiteral &>().value.safeGet<String>();
 
+        String options = "";
+
+        if (engine_args.size() >= 6)
+            options = engine_args[5]->as<ASTLiteral &>().value.safeGet<String>();
+
         return StorageMongoDB::create(
             args.table_id,
             parsed_host_port.first,
@@ -126,6 +138,7 @@ void registerStorageMongoDB(StorageFactory & factory)
             collection,
             username,
             password,
+            options,
             args.columns,
             args.constraints,
             args.comment);

src/Storages/StorageMongoDB.h

@@ -26,6 +26,7 @@ public:
         const std::string & collection_name_,
         const std::string & username_,
         const std::string & password_,
+        const std::string & options_,
         const ColumnsDescription & columns_,
         const ConstraintsDescription & constraints_,
         const String & comment);
@@ -50,6 +51,8 @@ private:
     const std::string collection_name;
     const std::string username;
     const std::string password;
+    const std::string options;
+    const std::string uri;
 
     std::shared_ptr<Poco::MongoDB::Connection> connection;
     bool authenticated = false;

src/Storages/StorageMongoDBSocketFactory.cpp

@@ -0,0 +1,36 @@
+#include "StorageMongoDBSocketFactory.h"
+
+#include <Poco/Net/IPAddress.h>
+#include <Poco/Net/SocketAddress.h>
+#include <Poco/Net/SecureStreamSocket.h>
+
+
+#pragma clang diagnostic ignored "-Wunused-parameter"
+#pragma GCC diagnostic ignored "-Wunused-parameter"
+
+
+namespace DB
+{
+
+Poco::Net::StreamSocket StorageMongoDBSocketFactory::createSocket(const std::string & host, int port, Poco::Timespan connectTimeout, bool secure)
+{
+    return secure ? createSecureSocket(host, port) : createPlainSocket(host, port);
+}
+
+Poco::Net::StreamSocket StorageMongoDBSocketFactory::createPlainSocket(const std::string & host, int port)
+{
+    Poco::Net::SocketAddress address(host, port);
+    Poco::Net::StreamSocket socket(address);
+
+    return socket;
+}
+
+Poco::Net::StreamSocket StorageMongoDBSocketFactory::createSecureSocket(const std::string & host, int port)
+{
+    Poco::Net::SocketAddress address(host, port);
+    Poco::Net::SecureStreamSocket socket(address, host);
+
+    return socket;
+}
+
+}

src/Storages/StorageMongoDBSocketFactory.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include <Poco/MongoDB/Connection.h>
+
+
+namespace DB
+{
+
+class StorageMongoDBSocketFactory : public Poco::MongoDB::Connection::SocketFactory
+{
+public:
+    virtual Poco::Net::StreamSocket createSocket(const std::string & host, int port, Poco::Timespan connectTimeout, bool secure) override;
+
+private:
+    Poco::Net::StreamSocket createPlainSocket(const std::string & host, int port);
+    Poco::Net::StreamSocket createSecureSocket(const std::string & host, int port);
+};
+
+}