Support client certificate authentication

README.md

@@ -73,6 +73,12 @@ The following environment variables are used by migration script:
 * `TARGET_SERVICE_URI` - service URI to the target MySQL database with admin credentials, which will be used for dump import.
 * `TARGET_MASTER_SERVICE_URI` - service URI for managing replication while migrating, omitting this variable will
 lead to fall-back to dump solution.
+SOURCE_SSL_* are optional, when provided it uses client certificate authentication.  
+* `SOURCE_SSL_CA` - The path name of the Certificate Authority (CA) certificate file in PEM format.
+* `SOURCE_SSL_CERT` - The path name of the server SSL public key certificate file in PEM format.
+* `SOURCE_SSL_KEY` - The path name of the server SSL private key file in PEM format.
+
+
 
 Environment variable are used here instead of usual arguments so that it's not possible to see credentials in the list
 of long-running processes. As for the `mysqldump/mysql` subprocesses they won't be visible, because they are hidden by

aiven_mysql_migrate/config.py

@@ -11,3 +11,7 @@
 SOURCE_SERVICE_URI = os.getenv("SOURCE_SERVICE_URI")
 TARGET_SERVICE_URI = os.getenv("TARGET_SERVICE_URI")
 TARGET_MASTER_SERVICE_URI = os.getenv("TARGET_MASTER_SERVICE_URI")
+
+SOURCE_SSL_CA = os.getenv("SOURCE_SSL_CA")
+SOURCE_SSL_CERT = os.getenv("SOURCE_SSL_CERT")
+SOURCE_SSL_KEY = os.getenv("SOURCE_SSL_KEY")

aiven_mysql_migrate/migration.py

@@ -17,6 +17,7 @@
 import concurrent
 import enum
 import logging
+import os
 import pymysql
 import shlex
 import signal
@@ -52,7 +53,14 @@ def __init__(
         self.mysqldump_proc: Optional[Popen] = None
         self.mysql_proc: Optional[Popen] = None
 
-        self.source = MySQLConnectionInfo.from_uri(source_uri, name="source")
+        self.source = MySQLConnectionInfo.from_uri(
+            source_uri,
+            name="source",
+            sslca=config.SOURCE_SSL_CA,
+            sslcert=config.SOURCE_SSL_CERT,
+            sslkey=config.SOURCE_SSL_KEY
+        )
+
         self.target = MySQLConnectionInfo.from_uri(target_uri, name="target")
         self.target_master = MySQLConnectionInfo.from_uri(
             target_master_uri, name="target master"
@@ -160,6 +168,8 @@ def _check_connections(self):
             conn_infos.append(self.target_master)
 
         for conn_info in conn_infos:
+            LOGGER.debug("conn_info.name :[%s]", conn_info.name)
+
             try:
                 with conn_info.cur():
                     pass
@@ -198,6 +208,13 @@ def _check_bin_log_format(self):
             if row_format.upper() != "ROW":
                 raise UnsupportedBinLogFormatException(f"Unsupported binary log format: {row_format}, only ROW is supported")
 
+    def _check_ssl_files(self):
+        if not (config.SOURCE_SSL_CA is None and config.SOURCE_SSL_CERT is None and config.SOURCE_SSL_KEY is None):
+            if not (os.path.exists(config.SOURCE_SSL_CA) and os.path.exists(config.SOURCE_SSL_CERT) and
+                    os.path.exists(config.SOURCE_SSL_KEY)):
+                LOGGER.debug("SSL files:[%s],[%s],[%s]", config.SOURCE_SSL_CA, config.SOURCE_SSL_CERT, config.SOURCE_SSL_KEY)
+                raise WrongMigrationConfigurationException("SSL files error!")
+
     def run_checks(
         self,
         force_method: Optional[MySQLMigrateMethod] = None,
@@ -220,6 +237,7 @@ def run_checks(
             )
             migration_method = MySQLMigrateMethod.dump
 
+        self._check_ssl_files()
         self._check_connections()
         self._check_databases_count()
         if dbs_max_total_size is not None:

aiven_mysql_migrate/utils.py

@@ -6,8 +6,12 @@
 from urllib.parse import parse_qs, urlparse
 
 import contextlib
+import logging
 import pymysql
 import re
+import urllib
+
+LOGGER = logging.getLogger(__name__)
 
 DEFAULT_MYSQL_PORT = 3306
 
@@ -30,14 +34,23 @@ class MySQLConnectionInfo:
     username: str
     password: str
     ssl: Optional[bool] = True
+    sslca: Optional[str] = None
+    sslcert: Optional[str] = None
+    sslkey: Optional[str] = None
 
     name: Optional[str] = None
 
     _version: Optional[str] = None
     _global_grants: Optional[List[str]] = None
 
     @staticmethod
-    def from_uri(uri: str, name: Optional[str] = None):
+    def from_uri(
+        uri: str,
+        name: Optional[str] = None,
+        sslca: Optional[str] = None,
+        sslcert: Optional[str] = None,
+        sslkey: Optional[str] = None
+    ):
         try:
             res = urlparse(uri, scheme="mysql")
             if res.scheme != "mysql" or not res.username or not res.password or not res.hostname:
@@ -49,13 +62,31 @@ def from_uri(uri: str, name: Optional[str] = None):
         port = res.port or DEFAULT_MYSQL_PORT
         options = parse_qs(res.query)
         ssl = not (options and options.get("ssl-mode", ["DISABLE"]) == ["DISABLE"])
+
         return MySQLConnectionInfo(
-            hostname=res.hostname, port=port, username=res.username, password=res.password, ssl=ssl, name=name
-        )
+            hostname=res.hostname,
+            port=port,
+            username=res.username,
+            password=res.password,
+            ssl=ssl,
+            sslca=sslca,
+            sslcert=sslcert,
+            sslkey=sslkey,
+            name=name
+            )
 
     def to_uri(self):
         ssl_mode = "DISABLE" if not self.ssl else "REQUIRE"
-        return f"mysql://{self.username}:{self.password}@{self.hostname}:{self.port}/?ssl-mode={ssl_mode}"
+
+        ssl_params = {
+            "ssl-ca": self.sslca,
+            "ssl-cert": self.sslcert,
+            "ssl-key": self.sslkey
+        }
+        ssl_auth = urllib.parse.urlencode(ssl_params) \
+            if self.sslca and self.sslcert and self.sslcert else ""
+        LOGGER.debug("ssl_auth:[%s]]", ssl_auth)
+        return f"mysql://{self.username}:{self.password}@{self.hostname}:{self.port}/?ssl-mode={ssl_mode}{ssl_auth}"
 
     def repr(self):
         return self.name
@@ -64,6 +95,9 @@ def _connect(self):
         ssl = None
         if self.ssl:
             ssl = {"require": True}
+
+        LOGGER.debug("connect [%s]- sslca:[%s], sslcert:[%s], sslkey:[%s]", self.name, self.sslca, self.sslcert, self.sslkey)
+
         return pymysql.connect(
             charset="utf8mb4",
             connect_timeout=config.MYSQL_CONNECTION_TIMEOUT,
@@ -73,6 +107,9 @@ def _connect(self):
             read_timeout=config.MYSQL_READ_TIMEOUT,
             port=self.port,
             ssl=ssl,
+            ssl_ca=self.sslca,
+            ssl_cert=self.sslcert,
+            ssl_key=self.sslkey,
             user=self.username,
             write_timeout=config.MYSQL_WRITE_TIMEOUT,
         )

requirement-dev.txt

@@ -4,6 +4,6 @@ isort==5.6.4
 mypy==0.790
 pylint==2.9.0
 pylint-quotes==0.2.1
-pymysql==0.10.0
+pymysql==1.0.2
 pytest==6.1.2
 yapf==0.30.0

setup.py

@@ -21,7 +21,7 @@ def get_long_description():
         ],
     },
     install_requires=[
-        "pymysql~=0.10.0"
+        "pymysql~=1.0.2"
     ],
     license="Apache 2.0",
     name="aiven-mysql-migrate",