Allow client certificate authentication for source database

Author: runwuf

README.md

@@ -73,6 +73,15 @@ 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.
+* `MYSQL_CONNECTION_TIMEOUT` - MySQL connect_timeout (default: 5 seconds)
+* `MYSQL_WRITE_TIMEOUT` - MySQL write_timeout (default: 5 seconds)
+* `MYSQL_READ_TIMEOUT` - MySQL read_timeout (default: 5 seconds)
+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

@@ -52,7 +52,17 @@ def __init__(
         self.mysqldump_proc: Optional[Popen] = None
         self.mysql_proc: Optional[Popen] = None
 
-        self.source = MySQLConnectionInfo.from_uri(source_uri, name="source")
+        if config.SOURCE_SSL_CA and config.SOURCE_SSL_CERT and config.SOURCE_SSL_KEY:
+            self.source = MySQLConnectionInfo.from_uri(
+                source_uri,
+                name="source",
+                sslca=config.SOURCE_SSL_CA,
+                sslcert=config.SOURCE_SSL_CERT,
+                sslkey=config.SOURCE_SSL_KEY
+            )
+        else:
+            self.source = MySQLConnectionInfo.from_uri(source_uri, name="source")
+
         self.target = MySQLConnectionInfo.from_uri(target_uri, name="target")
         self.target_master = MySQLConnectionInfo.from_uri(
             target_master_uri, name="target master"
@@ -160,6 +170,8 @@ def _check_connections(self):
             conn_infos.append(self.target_master)
 
         for conn_info in conn_infos:
+            LOGGER.debug("conn_info :[%s]", conn_info)
+
             try:
                 with conn_info.cur():
                     pass

aiven_mysql_migrate/utils.py

@@ -6,9 +6,12 @@
 from urllib.parse import parse_qs, urlparse
 
 import contextlib
+import logging
 import pymysql
 import re
 
+LOGGER = logging.getLogger(__name__)
+
 DEFAULT_MYSQL_PORT = 3306
 
 ROUTINE_DEFINER_RE = re.compile("^CREATE DEFINER *= *(`.*?`@`.*?`) +(.*$)")
@@ -30,14 +33,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 +61,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
-        )
+
+        LOGGER.debug("from_uri - sslca:[%s], sslcert:[%s], sslkey:[%s]", sslca, sslcert, sslkey)
+        if sslca and sslcert and sslkey:
+            return MySQLConnectionInfo(
+                hostname=res.hostname,
+                port=port,
+                username=res.username,
+                password=res.password,
+                ssl=ssl,
+                sslca=sslca,
+                sslcert=sslcert,
+                sslkey=sslkey,
+                name=name
+            )
+        else:
+            return MySQLConnectionInfo(
+                hostname=res.hostname, port=port, username=res.username, password=res.password, ssl=ssl, 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_auth = f"&ssl-ca={self.sslca}&ssl-cert={self.sslcert}&ssl-key={self.sslkey}" \
+                   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,18 +94,37 @@ def _connect(self):
         ssl = None
         if self.ssl:
             ssl = {"require": True}
-        return pymysql.connect(
-            charset="utf8mb4",
-            connect_timeout=config.MYSQL_CONNECTION_TIMEOUT,
-            cursorclass=pymysql.cursors.DictCursor,
-            host=self.hostname,
-            password=self.password,
-            read_timeout=config.MYSQL_READ_TIMEOUT,
-            port=self.port,
-            ssl=ssl,
-            user=self.username,
-            write_timeout=config.MYSQL_WRITE_TIMEOUT,
-        )
+
+        if self.sslca and self.sslcert and self.sslkey:
+            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,
+                cursorclass=pymysql.cursors.DictCursor,
+                host=self.hostname,
+                password=self.password,
+                read_timeout=config.MYSQL_READ_TIMEOUT,
+                port=self.port,
+                ssl_ca=self.sslca,
+                ssl_cert=self.sslcert,
+                ssl_key=self.sslkey,
+                user=self.username,
+                write_timeout=config.MYSQL_WRITE_TIMEOUT,
+            )
+        else:
+            return pymysql.connect(
+                charset="utf8mb4",
+                connect_timeout=config.MYSQL_CONNECTION_TIMEOUT,
+                cursorclass=pymysql.cursors.DictCursor,
+                host=self.hostname,
+                password=self.password,
+                read_timeout=config.MYSQL_READ_TIMEOUT,
+                port=self.port,
+                ssl=ssl,
+                user=self.username,
+                write_timeout=config.MYSQL_WRITE_TIMEOUT,
+            )
 
     @property
     def version(self) -> str: