Release 1.0.16

Author: dongwook-chan

CHANGELOG

@@ -1,3 +1,9 @@
+1.0.16 12/07/2026
+
+* Parse FormatDescriptionEvent post-header lengths and use them to align event payloads.
+* Add malformed binlog event validation for invalid post-header lengths.
+* Support MySQL 8.4 binary log coordinate discovery with SHOW BINARY LOG STATUS.
+
 1.0.15 07/02/2026
 
 * Security fix: Sensitive connection settings are no longer logged when logging is enabled.
@@ -284,4 +290,4 @@
 
 1.0.13 18/01/2026
 
-* Use a project logger instead of root logger (#637)
+* Use a project logger instead of root logger (#637)

pymysqlreplication/binlogstream.py

@@ -53,6 +53,7 @@
 MYSQL_EXPECTED_ERROR_CODES = [2013, 2006]
 
 PYMYSQL_VERSION_LT_06 = Version(pymysql.__version__) < Version("0.6")
+MYSQL_ER_PARSE_ERROR = 1064
 
 
 class ReportSlave(object):
@@ -263,12 +264,13 @@ def __init__(
 
         # We can't filter on packet level TABLE_MAP and rotate event because
         # we need them for handling other operations
-        self.__allowed_events_in_packet = frozenset([TableMapEvent, RotateEvent]).union(
-            self.__allowed_events
-        )
+        self.__allowed_events_in_packet = frozenset(
+            [FormatDescriptionEvent, TableMapEvent, RotateEvent]
+        ).union(self.__allowed_events)
 
         self.__server_id = server_id
         self.__use_checksum = False
+        self.__post_header_lengths = None
 
         # Store table meta information
         self.table_map = {}
@@ -332,6 +334,15 @@ def __checksum_enabled(self):
             return False
         return True
 
+    def __show_binary_log_status(self, cur):
+        try:
+            cur.execute("SHOW BINARY LOG STATUS")
+        except pymysql.err.ProgrammingError as exc:
+            if not exc.args or exc.args[0] != MYSQL_ER_PARSE_ERROR:
+                raise
+            cur.execute("SHOW MASTER STATUS")
+        return cur.fetchone()
+
     def _register_slave(self):
         if not self.report_slave:
             return
@@ -404,12 +415,13 @@ def __connect_to_stream(self):
                 # valid, if not, get the current position from master
                 if self.log_file is None or self.log_pos is None:
                     cur = self._stream_connection.cursor()
-                    cur.execute("SHOW MASTER STATUS")
-                    master_status = cur.fetchone()
-                    if master_status is None:
-                        raise BinLogNotEnabled()
-                    self.log_file, self.log_pos = master_status[:2]
-                    cur.close()
+                    try:
+                        master_status = self.__show_binary_log_status(cur)
+                        if master_status is None:
+                            raise BinLogNotEnabled()
+                        self.log_file, self.log_pos = master_status[:2]
+                    finally:
+                        cur.close()
 
                 prelude = struct.pack("<i", len(self.log_file) + 11) + bytes(
                     bytearray([COM_BINLOG_DUMP])
@@ -636,6 +648,7 @@ def fetchone(self):
                 self.__optional_meta_data,
                 self.__enable_logging,
                 self.__use_column_name_cache,
+                self.__post_header_lengths,
             )
 
             if binlog_event.event_type == ROTATE_EVENT:
@@ -660,6 +673,13 @@ def fetchone(self):
                 # We're currently at, or past, the specified end log position.
                 self.is_past_end_log_pos = True
 
+            if (
+                binlog_event.event_type == FORMAT_DESCRIPTION_EVENT
+                and binlog_event.event is not None
+            ):
+                self.mysql_version = binlog_event.event.mysql_version
+                self.__post_header_lengths = binlog_event.event.post_header_len
+
             # This check must not occur before clearing the ``table_map`` as a
             # result of a RotateEvent.
             #
@@ -705,9 +725,6 @@ def fetchone(self):
             ):
                 continue
 
-            if binlog_event.event_type == FORMAT_DESCRIPTION_EVENT:
-                self.mysql_version = binlog_event.event.mysql_version
-
             return binlog_event.event
 
     def _allowed_event_list(

pymysqlreplication/event.py

@@ -5,7 +5,7 @@
 import zlib
 
 from pymysqlreplication.constants.STATUS_VAR_KEY import *
-from pymysqlreplication.exceptions import StatusVariableMismatch
+from pymysqlreplication.exceptions import MalformedBinLogEvent, StatusVariableMismatch
 from pymysqlreplication.util.bytes import parse_decimal_from_bytes
 from pymysqlreplication.logger import logger
 from typing import Union, Optional
@@ -27,15 +27,18 @@ def __init__(
         freeze_schema=False,
         ignore_decode_errors=False,
         verify_checksum=False,
+        use_checksum=False,
         optional_meta_data=False,
         enable_logging=False,
         use_column_name_cache=False,
+        post_header_lengths=None,
     ):
         self.packet = from_packet
         self.table_map = table_map
         self.event_type = self.packet.event_type
         self.timestamp = self.packet.timestamp
         self.event_size = event_size
+        self._post_header_lengths = post_header_lengths
         self._ctl_connection = ctl_connection
         self.mysql_version = mysql_version
         self._ignore_decode_errors = ignore_decode_errors
@@ -48,6 +51,38 @@ def __init__(
         self._verify_event()
         self.dbms = self._ctl_connection._get_dbms()
 
+    def _declared_post_header_len(self, default):
+        if not self._post_header_lengths:
+            return default
+
+        index = self.event_type - 1
+        if index < 0 or index >= len(self._post_header_lengths):
+            return default
+
+        return self._post_header_lengths[index]
+
+    def _advance_to_payload(self, known_post_header_len):
+        declared_post_header_len = self._declared_post_header_len(known_post_header_len)
+
+        if declared_post_header_len < known_post_header_len:
+            raise MalformedBinLogEvent(
+                "Event type %s declares post-header length %s, shorter than "
+                "the supported minimum %s"
+                % (self.event_type, declared_post_header_len, known_post_header_len)
+            )
+
+        extra_len = declared_post_header_len - known_post_header_len
+        if self.packet.read_bytes + extra_len > self.event_size:
+            raise MalformedBinLogEvent(
+                "Event type %s declares post-header length %s beyond event size %s"
+                % (self.event_type, declared_post_header_len, self.event_size)
+            )
+
+        if extra_len:
+            self.packet.advance(extra_len)
+
+        return declared_post_header_len
+
     def _read_table_id(self):
         # Table ID is 6 byte
         # pad little-endian number
@@ -389,15 +424,37 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.mysql_version = tuple(map(int, numbers.split(".")))
         self.created = struct.unpack("<I", self.packet.read(4))[0]
         self.common_header_len = struct.unpack("<B", self.packet.read(1))[0]
-        offset = (
-            4 + 2 + 50 + 1
-        )  # created + binlog_version + mysql_version_str + common_header_len
-        checksum_algorithm = 1
-        checksum = 4
-        n = event_size - offset - self.common_header_len - checksum_algorithm - checksum
-        self.post_header_len = struct.unpack(f"<{n}B", self.packet.read(n))
-        self.server_version_split = struct.unpack("<3B", self.packet.read(3))
-        self.number_of_event_types = struct.unpack("<B", self.packet.read(1))[0]
+
+        fixed_part_len = 2 + 50 + 4 + 1
+        if event_size < fixed_part_len:
+            raise MalformedBinLogEvent(
+                "FormatDescriptionEvent size %s is shorter than fixed part %s"
+                % (event_size, fixed_part_len)
+            )
+
+        remaining_len = event_size - fixed_part_len
+        checksum_aware = self.mysql_version >= (5, 6) or self.mysql_version[0] >= 10
+        checksum_algorithm_len = 1 if checksum_aware else 0
+        checksum_len = 0 if kwargs.get("use_checksum") else 4 if checksum_aware else 0
+        post_header_len = remaining_len - checksum_algorithm_len - checksum_len
+
+        if post_header_len < 0:
+            raise MalformedBinLogEvent(
+                "FormatDescriptionEvent size %s is too short for checksum metadata"
+                % event_size
+            )
+
+        self.post_header_len = struct.unpack(
+            f"<{post_header_len}B", self.packet.read(post_header_len)
+        )
+        self.checksum_algorithm = (
+            self.packet.read_uint8() if checksum_algorithm_len else None
+        )
+        if checksum_len:
+            self.packet.advance(checksum_len)
+
+        self.server_version_split = self.mysql_version[:3]
+        self.number_of_event_types = len(self.post_header_len)
 
     def _dump(self):
         print(f"Binlog version: {self.binlog_version}")
@@ -488,6 +545,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
         self.schema_length = struct.unpack("!B", self.packet.read(1))[0]
         self.error_code = self.packet.read_uint16()
         self.status_vars_length = self.packet.read_uint16()
+        post_header_len = self._advance_to_payload(13)
 
         # Payload
         status_vars_end_pos = self.packet.read_bytes + self.status_vars_length
@@ -499,9 +557,19 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
 
         self.schema = self.packet.read(self.schema_length)
         self.packet.advance(1)
-        query = self.packet.read(
-            event_size - 13 - self.status_vars_length - self.schema_length - 1
+        query_len = (
+            event_size
+            - post_header_len
+            - self.status_vars_length
+            - self.schema_length
+            - 1
         )
+        if query_len < 0:
+            raise MalformedBinLogEvent(
+                "QueryEvent payload length is negative after applying post-header "
+                "length %s" % post_header_len
+            )
+        query = self.packet.read(query_len)
         self.query = query.decode("utf-8", errors="backslashreplace")
         # string[EOF]    query
 

pymysqlreplication/exceptions.py

@@ -8,6 +8,11 @@ def __init__(self):
         Exception.__init__(self, "MySQL binary logging is not enabled.")
 
 
+class MalformedBinLogEvent(Exception):
+    def __init__(self, message):
+        Exception.__init__(self, message)
+
+
 class StatusVariableMismatch(Exception):
     def __init__(self):
         Exception.__init__(

pymysqlreplication/packet.py

@@ -76,6 +76,7 @@ def __init__(
         optional_meta_data,
         enable_logging,
         use_column_name_cache=False,
+        post_header_lengths=None,
     ):
         # -1 because we ignore the ok byte
         self.read_bytes = 0
@@ -133,6 +134,8 @@ def __init__(
             optional_meta_data=optional_meta_data,
             enable_logging=enable_logging,
             use_column_name_cache=use_column_name_cache,
+            use_checksum=use_checksum,
+            post_header_lengths=post_header_lengths,
         )
         if not self.event._processed:
             self.event = None

pymysqlreplication/row_event.py

@@ -130,6 +130,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
             self.flags, self.extra_data_length = struct.unpack(
                 "<HH", self.packet.read(4)
             )
+            self._advance_to_payload(10)
             if self.extra_data_length > 2:
                 self.extra_data_type = struct.unpack("<B", self.packet.read(1))[0]
                 # ndb information
@@ -155,6 +156,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
                     self.extra_data = self.packet.read(self.extra_data_length - 3)
         else:
             self.flags = struct.unpack("<H", self.packet.read(2))[0]
+            self._advance_to_payload(8)
 
         # Body
         self.number_of_columns = self.packet.read_length_coded_binary()
@@ -833,6 +835,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
             return
 
         self.flags = struct.unpack("<H", self.packet.read(2))[0]
+        self._advance_to_payload(8)
 
         # Payload
         self.schema_length = struct.unpack("!B", self.packet.read(1))[0]