zfs-snapshot-backup/compare_checksums.py

#!/usr/local/bin/python

import pathlib
import subprocess

from copy_snapshots_to_remote import get_member_name
from create_snapshot_backup import call, remote_call

SSH_KEY_FILE = "/mnt/Datenspeicher/snap-backup-dataset/backup_key"
SSH_REMOTE = "zfs_snap_backup@etha.cpi.imtek.uni-freiburg.de"

REMOTE_PATH = "zfs-backups"

ZFS_POOL = "Datenspeicher"
ZFS_ELAB_PREFIX = "elabfs-"

TMP_BACKUP_FOLDER = "/mnt/Datenspeicher/snap-backup-dataset/temporary-backups"


def get_local_backups():
    """ returns a list of .gz files in the local temp backup folder """
    tmp_folder = pathlib.Path(TMP_BACKUP_FOLDER)
    return [i for i in tmp_folder.iterdir() if i.suffix==".gz"]


def get_remote_checksum(local_backup_path):
    """ gets the checksum of a remote backup of a local file

    :params local_backup_path: path of the local backup file
    :returns: sh256 hash of the remote backup or None
    """
    member = get_member_name(local_backup_path.name)
    remote_path =  f"{REMOTE_PATH}/{member}/{local_backup_path.name}"
    try:
        result = remote_call(["sha256sum", remote_path])
        parts = result.strip().split()
        return parts[0].strip()
    except subprocess.CalledProcessError as e:
        return None

def get_local_checksum(local_backup_path):
    """ gets the checksum of a local file

    :params local_backup_path: path of the local backup file
    :returns: sh256 hash of the file or None
    """
    result = call(["sha256", str(local_backup_path)], as_text=True)
    parts = result.split("=")
    return parts[1].strip()

def batch_compare():
    """ compares local file checksums to the remote backups """
    for local_backup_path in get_local_backups():
        remote_checksum = get_remote_checksum(local_backup_path)
        local_checksum = get_local_checksum(local_backup_path)
        if remote_checksum == local_checksum:
            print(f"     {local_checksum} -> {remote_checksum}")
        else:
            print(f"XXX  {local_checksum} -> {remote_checksum}")


if __name__ == "__main__":
    batch_compare()