Python Script for MySQL Checksum Validation

#!/usr/bin/env python3
"""MySQL backup checksum validator.

Exit codes (consumed by the DR drill orchestrator):
    0  every table checksum matches the signed manifest -> proceed with promotion
    1  at least one MISMATCH, MISSING, or ERROR          -> halt failover, quarantine
    2  usage / configuration error                       -> abort pipeline
"""
import json
import logging
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import contextmanager
from typing import Dict, List, Tuple

import mysql.connector
from mysql.connector import pooling

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("mysql_checksum_validator")

FAILURE_STATES = {"MISMATCH", "MISSING", "ERROR"}


class MySQLChecksumValidator:
    def __init__(self, db_config: dict, manifest: Dict[str, str],
                 max_workers: int = 8, lock_wait_timeout: int = 5):
        self.db_config = db_config
        self.manifest = manifest
        self.max_workers = max_workers
        self.lock_wait_timeout = lock_wait_timeout
        self.pool = pooling.MySQLConnectionPool(
            pool_name="checksum_pool",
            pool_size=self.max_workers,
            **db_config,
        )

    @contextmanager
    def _get_connection(self):
        conn = self.pool.get_connection()
        try:
            cursor = conn.cursor(dictionary=True)
            # Bound lock waits so a busy table cannot stall a worker or spike lag.
            cursor.execute(f"SET SESSION lock_wait_timeout = {self.lock_wait_timeout}")
            cursor.execute(f"SET SESSION innodb_lock_wait_timeout = {self.lock_wait_timeout}")
            yield cursor
        finally:
            conn.close()

    def _validate_single_table(self, schema: str, table: str) -> dict:
        expected = self.manifest.get(f"{schema}.{table}")
        with self._get_connection() as cursor:
            try:
                cursor.execute(f"CHECKSUM TABLE `{schema}`.`{table}`")
                result = cursor.fetchone()
                # A nonexistent table returns one row with a NULL Checksum, not no rows.
                if result is None or result["Checksum"] is None:
                    return {"schema": schema, "table": table,
                            "status": "MISSING", "checksum": None}

                live = str(result["Checksum"])
                status = "VALID" if live == expected else "MISMATCH"
                return {"schema": schema, "table": table,
                        "status": status, "checksum": live}
            except mysql.connector.Error as exc:
                logger.error("Validation failed for %s.%s: %s", schema, table, exc)
                return {"schema": schema, "table": table,
                        "status": "ERROR", "checksum": None, "error": str(exc)}

    def run_validation(self) -> Tuple[bool, List[dict]]:
        results: List[dict] = []
        # Split on the last dot so schema names containing dots stay intact.
        table_list = [tuple(key.rsplit(".", 1)) for key in self.manifest]

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            future_to_table = {
                executor.submit(self._validate_single_table, schema, table): f"{schema}.{table}"
                for schema, table in table_list
            }
            for future in as_completed(future_to_table):
                results.append(future.result())

        failures = [r for r in results if r["status"] in FAILURE_STATES]
        return len(failures) == 0, results


def load_json(path: str) -> dict:
    with open(path, "r", encoding="utf-8") as handle:
        return json.load(handle)


def main() -> int:
    if len(sys.argv) != 3:
        logger.error("Usage: mysql_checksum_validator.py <db_config.json> <manifest.json>")
        return 2

    try:
        db_config = load_json(sys.argv[1])
        manifest = load_json(sys.argv[2])
    except (OSError, json.JSONDecodeError) as exc:
        logger.error("Configuration error: %s", exc)
        return 2

    max_workers = int(db_config.get("max_workers", 8))
    validator = MySQLChecksumValidator(db_config["db"], manifest, max_workers)
    success, results = validator.run_validation()

    for record in results:
        level = logging.INFO if record["status"] == "VALID" else logging.WARNING
        logger.log(level, "%s.%s | Status: %s | Checksum: %s",
                   record["schema"], record["table"], record["status"], record["checksum"])

    if not success:
        logger.critical("Validation failed. Aborting DR promotion.")
        return 1

    logger.info("All checksums validated successfully. Proceeding with failover.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

{
  "db": {
    "host": "dr-replica-01.internal",
    "port": 3306,
    "user": "dr_validator",
    "password": "${VAULT_DR_PASS}",
    "database": "information_schema"
  },
  "max_workers": 12
}

{
  "app_production.users": "3847291056",
  "app_production.orders": "9281736450",
  "app_production.sessions": "1029384756"
}

#!/usr/bin/env python3
"""Emit a signed-ready manifest of CHECKSUM TABLE values for a schema."""
import json
import sys

import mysql.connector


def build_manifest(schema: str, **db) -> dict:
    conn = mysql.connector.connect(database="information_schema", **db)
    cursor = conn.cursor(dictionary=True)
    cursor.execute(
        "SELECT table_name FROM information_schema.tables "
        "WHERE table_schema = %s AND table_type = 'BASE TABLE'",
        (schema,),
    )
    tables = [row["table_name"] for row in cursor.fetchall()]
    manifest = {}
    for table in tables:
        cursor.execute(f"CHECKSUM TABLE `{schema}`.`{table}`")
        row = cursor.fetchone()
        manifest[f"{schema}.{table}"] = str(row["Checksum"])
    conn.close()
    return manifest


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: build_manifest.py <schema>", file=sys.stderr)
        sys.exit(2)
    out = build_manifest(sys.argv[1], host="db-primary.internal",
                         user="backup_reader", password="${VAULT_BACKUP_PASS}")
    json.dump(out, sys.stdout, indent=2)
    sys.exit(0)

2026-07-05 04:12:01 | INFO | mysql_checksum_validator | app_production.users | Status: VALID | Checksum: 3847291056
2026-07-05 04:12:01 | INFO | mysql_checksum_validator | app_production.orders | Status: VALID | Checksum: 9281736450
2026-07-05 04:12:01 | INFO | mysql_checksum_validator | All checksums validated successfully. Proceeding with failover.

2026-07-05 04:12:01 | WARNING | mysql_checksum_validator | app_production.orders | Status: MISMATCH | Checksum: 5510293847
2026-07-05 04:12:01 | CRITICAL | mysql_checksum_validator | Validation failed. Aborting DR promotion.

#!/usr/bin/env bash
set -euo pipefail

python3 mysql_checksum_validator.py db_config.json manifest.json
case $? in
  0) echo "[$(date -u)] checksums valid; promoting" ;;
  1) curl -s -X POST "$PAGERDUTY_WEBHOOK" -d '{"event":"dr_checksum_mismatch"}'; exit 1 ;;
  *) echo "[$(date -u)] validator misconfigured; aborting"; exit 2 ;;
esac