Developed the 'MariadbStartEncryptionEvent' (#415)

* implement StartEncryptionEvent

* Completed setup of MariaDB test environment.

* Fix: Correct assertion in test_basic:_allowed_event_list

* Fixed docker-compose.yaml

* Fix: Correct assertion in test_basic:_allowed_event_list

* Add unit tests for MariadbStartEncryptionEvent class

* Fix:find file from current directroy

* Fix: find the file path

* Rename StartEncryptionEvent to MariadbStartEncryptionEvent

* add description for MariadbStartEncruptionEvent

* fix description for MariadbStartEncryptionEvent

---------

Co-authored-by: Pilmo <pilmokim99@gmail.com>
Co-authored-by: BeautterLife <pgh9801@gmail.com>

.mariadb/my.cnf

@@ -0,0 +1,23 @@
+[client-server]
+# Port or socket location where to connect
+# port = 3306
+socket = /run/mysqld/mysqld.sock
+
+# Import all .cnf files from configuration directory
+
+!includedir /etc/mysql/mariadb.conf.d/
+!includedir /etc/mysql/conf.d/
+
+
+[mariadb]
+plugin_load_add = file_key_management
+# Key files that are not encrypted
+loose_file_key_management_filename = /opt/key_file/no_encryption_key.key
+
+# Encrypted key file
+# loose_file_key_management_filename=/opt/key_file/keyfile.enc
+# loose_file_key_management_filekey=FILE:/opt/key_file/no_encryption_key.key
+# file_key_management_encryption_algorithm=aes_ctr
+
+# Set encrypt_binlog
+encrypt_binlog=ON

.mariadb/no_encryption_key.key

@@ -0,0 +1 @@
+1;dda0ccb18a28b0b4c2448b5f0217a134

docker-compose.yml

@@ -28,4 +28,10 @@ services:
       --log-bin=master-bin
       --binlog-format=row
       --log-slave-updates=on
-      
+    volumes:
+      - type: bind
+        source: ./.mariadb
+        target: /opt/key_file
+      - type: bind
+        source: ./.mariadb/my.cnf
+        target: /etc/mysql/my.cnf  

pymysqlreplication/binlogstream.py

@@ -15,7 +15,7 @@
     XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
     HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
-    MariadbAnnotateRowsEvent, RandEvent)
+    MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent)
 from .exceptions import BinLogNotEnabled
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
@@ -622,7 +622,8 @@ def _allowed_event_list(self, only_events, ignored_events,
                 NotImplementedEvent,
                 MariadbGtidEvent,
                 MariadbAnnotateRowsEvent,
-                RandEvent
+                RandEvent,
+                MariadbStartEncryptionEvent
                 ))
         if ignored_events is not None:
             for e in ignored_events:

pymysqlreplication/event.py

@@ -488,6 +488,35 @@ def _dump(self):
         print("seed1: %d" % (self.seed1))
         print("seed2: %d" % (self.seed2))
 
+class MariadbStartEncryptionEvent(BinLogEvent):
+    """
+    Since MariaDB 10.1.7, 
+    the START_ENCRYPTION event is written to every binary log file 
+    if encrypt_binlog is set to ON. Prior to enabling this setting, 
+    additional configuration steps are required in MariaDB. 
+    (Link: https://mariadb.com/kb/en/encrypting-binary-logs/)
+
+    This event is written just once, after the Format Description event
+
+    Attributes:
+        schema: The Encryption scheme, always set to 1 for system files.
+        key_version: The Encryption key version.
+        nonce: Nonce (12 random bytes) of current binlog file.
+    """
+
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(MariadbStartEncryptionEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+
+        self.schema = self.packet.read_uint8()
+        self.key_version = self.packet.read_uint32()
+        self.nonce = self.packet.read(12)
+
+    def _dump(self):
+        print("Schema: %d" % self.schema)
+        print("Key version: %d" % self.key_version)
+        print(f"Nonce: {self.nonce}")
+
+
 class NotImplementedEvent(BinLogEvent):
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
         super(NotImplementedEvent, self).__init__(

pymysqlreplication/packet.py

@@ -88,7 +88,7 @@ class BinLogPacketWrapper(object):
         constants.MARIADB_BINLOG_CHECKPOINT_EVENT: event.NotImplementedEvent,
         constants.MARIADB_GTID_EVENT: event.MariadbGtidEvent,
         constants.MARIADB_GTID_GTID_LIST_EVENT: event.NotImplementedEvent,
-        constants.MARIADB_START_ENCRYPTION_EVENT: event.NotImplementedEvent
+        constants.MARIADB_START_ENCRYPTION_EVENT: event.MariadbStartEncryptionEvent
     }
 
     def __init__(self, from_packet, table_map,

pymysqlreplication/tests/test_basic.py

@@ -18,6 +18,7 @@
 from pymysqlreplication.exceptions import TableMetadataUnavailableError
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.row_event import *
+from pathlib import Path
 
 __all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings", "TestGtidBinLogStreamReader", "TestMariadbBinlogStreamReader", "TestStatementConnectionSetting"]
 
@@ -27,9 +28,9 @@ def ignoredEvents(self):
         return [GtidEvent]
 
     def test_allowed_event_list(self):
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 18)
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 17)
-        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 17)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 19)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 18)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 18)
         self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
 
     def test_read_query_event(self):
@@ -1036,6 +1037,42 @@ def test_annotate_rows_event(self):
         self.assertEqual(event.sql_statement,insert_query)
         self.assertIsInstance(event,MariadbAnnotateRowsEvent)
 
+    def test_start_encryption_event(self):
+        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "INSERT INTO test (data) VALUES('Hello World')"
+        self.execute(query)
+        self.execute("COMMIT")
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+
+        start_encryption_event = self.stream.fetchone()
+        self.assertIsInstance(start_encryption_event, MariadbStartEncryptionEvent)
+
+        schema = start_encryption_event.schema
+        key_version = start_encryption_event.key_version
+        nonce = start_encryption_event.nonce
+
+        from pathlib import Path
+
+        encryption_key_file_path = Path(__file__).parent.parent.parent
+
+        try:
+            with open(f"{encryption_key_file_path}/.mariadb/no_encryption_key.key", "r") as key_file:
+                first_line = key_file.readline()
+                key_version_from_key_file = int(first_line.split(";")[0])
+        except Exception as e:
+            self.fail("raised unexpected exception: {exception}".format(exception=e))
+        finally:
+            self.resetBinLog()
+
+        # schema is always 1
+        self.assertEqual(schema, 1)
+        self.assertEqual(key_version, key_version_from_key_file)
+        self.assertEqual(type(nonce), bytes)
+        self.assertEqual(len(nonce), 12)        
+
 class TestStatementConnectionSetting(base.PyMySQLReplicationTestCase):
     def setUp(self):
         super(TestStatementConnectionSetting, self).setUp()
@@ -1065,8 +1102,8 @@ def test_rand_event(self):
     def tearDown(self):
         self.execute("SET @@binlog_format='ROW'")
         self.assertEqual(self.bin_log_format(), "ROW")
-        super(TestStatementConnectionSetting, self).tearDown()
-        
+        super(TestStatementConnectionSetting, self).tearDown()        
+
 
 if __name__ == "__main__":
     import unittest