mysqloperator/controller/backup/meb/meb_controller.py

# Copyright (c) 2025, Oracle and/or its affiliates.
#
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/

import base64
import datetime
import json
import shutil
import subprocess
import sys
import urllib.parse

import http.client
from urllib.parse import urlparse

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.primitives.hashes import SHA256

import threading
from datetime import timedelta

MYSQLBACKUP_BINARY = '/usr/bin/mysqlbackup'

MYCNF_PATH = '/etc/my.cnf'




class MebStorageLocalImage:
    def __init__(self, target_path):
        self.target_path = target_path

    def get_backup_args(self):
        return ['--backup-image='+self.target_path]


class OCIRequest:
    def __init__(self, credentials):
        self.tenancy_id = credentials['tenancy']
        self.user_id = credentials['user']
        self.fingerprint = credentials['fingerprint']
        self.private_key = credentials['privatekey']
        self.region = credentials['region']

    def request(self, host: str, request_target: str, method: str,
                request_body: dict) -> dict:
        private_key = serialization.load_pem_private_key(
            self.private_key.encode('utf-8'),
            password=None
        )

        endpoint = f'https://{host}{request_target}'

        date_header = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
        headers = {
            "Content-Type": "application/json",
            "date": date_header,
            "host": host,
            "opc-client-info": "public-objectstorage:0.1.0"
        }

        signing_string = f"(request-target): {method.lower()} {request_target}\n" \
                         f"host: {host}\n" \
                         f"date: {headers['date']}\n" \
                         f"opc-client-info: {headers['opc-client-info']}"

        # Sign the string
        signature = base64.b64encode(
            private_key.sign(
                signing_string.encode('utf-8'),
                padding.PKCS1v15(),
                SHA256()
            )
        ).decode('utf-8')

        authorization_header = (
            f'Signature version="1",'
            f'headers="(request-target) host date opc-client-info",'
            f'keyId="{self.tenancy_id}/{self.user_id}/{self.fingerprint}",'
            f'algorithm="rsa-sha256",'
            f'signature="{signature}"'
        )
        headers['authorization'] = authorization_header

        data = json.dumps(request_body) if request_body else None

        parsed_url = urlparse(endpoint)
        conn = http.client.HTTPSConnection(parsed_url.netloc)

        path = parsed_url.path
        if parsed_url.query:
            path += "?" + parsed_url.query

        headers = headers or {}
        if data:
            headers['Content-Type'] = 'application/json'

        conn.request(method, path, body=data, headers=headers)
        response = conn.getresponse()

        if response.status not in (200, 204):
            print(f"Failed to do OCI request: {response.status}")
            print(response.read().decode())
            raise Exception("Failed OCI API call")

        response_data = response.read()
        if response_data:
            return json.loads(response_data)

        return True


class OCIObjectStorage:
    def __init__(self, oci: OCIRequest, bucket_name: str, namespace_name: str):
        self.oci = oci
        self.bucket_name = bucket_name
        self.namespace_name = namespace_name

    def create_par(self, name:str, object_name: str, access_type: str, expiry: timedelta) -> str:
        host = f'objectstorage.{self.oci.region}.oraclecloud.com'
        request_target = f"/n/{self.namespace_name}/b/{self.bucket_name}/p"

        expiration_time = (datetime.datetime.utcnow() + expiry).strftime('%Y-%m-%dT%H:%M:%SZ')
        request_body = {
            "name": name,
            "objectName": object_name,
            "accessType": access_type,
            "timeExpires": expiration_time
        }

        response = self.oci.request(host, request_target, "POST", request_body)
        return (f'https://{host}{response["accessUri"]}', response['id'])


    def delete_par(self, id: str) -> None:
        id = urllib.parse.quote(id)
        host = f'objectstorage.{self.oci.region}.oraclecloud.com'
        request_target = f"/n/{self.namespace_name}/b/{self.bucket_name}/p/{id}"

        self.oci.request(host, request_target, "DELETE", None)

class MebStorageOCIPAR:
    def __init__(self, target_path, ocios: OCIObjectStorage):
        self.target_path = target_path
        self.ocios = ocios
        self.pars_to_delete = []
        self.hidden = []


    def clean(self):
        for par_id in self.pars_to_delete:
            try:
                self.ocios.delete_par(par_id)
            except Exception as exc:
                # Don't fail during cleanup
                print(f"Failed to delete PAR {par_id}: {exc}")


    def get_args(self):
        (par, par_id) = self.ocios.create_par("backup-par",
                                              self.target_path,
                                              "ObjectWrite",
                                              timedelta(hours=1))

        par_url=par.replace(self.target_path, "")
        self.pars_to_delete.append(par_id)
        self.hidden.append(par_url)

        return [
            '--backup-image=-',
            '--cloud-service=OCI',
            '--cloud-object='+self.target_path,
            '--cloud-par-url='+par.replace(self.target_path, "")
        ]

    def get_filter(self):
        return self.hidden


class MebStorageOCIRestore:
    par_base_url: str = ""

    def __init__(self, par_base_url: str):
        self.par_base_url = par_base_url

    def get_args(self, source):
        return [
            '--backup-image=-',
            '--cloud-service=OCI',
            '--cloud-par-url='+self.par_base_url+source,
        ]

    def get_filter(self):
        return [self.par_base_url]


class MebStorageS3:
    def __init__(self, object_key:str, region:str, bucket:str,
                 key_id:str, secret_access_key:str,
                 host:str=None):

        self.opts = [
            "--cloud-service=s3",
            "--cloud-aws-region="+region,
            "--cloud-access-key-id="+key_id,
            "--cloud-secret-access-key="+secret_access_key,
            "--cloud-bucket="+bucket,
            "--cloud-object-key="+object_key,
            "--backup-image=-"
        ]

        if host:
            self.opts.append("--cloud-host="+host)


    def get_args(self):
        return self.opts

    def get_filter(self):
        return None


class MySQLEnterpriseBackup:
    def __init__(self, storage, auth_options: list, tmppath: str):
        self.auth_options = auth_options
        self.storage = storage
        self.tmppath = tmppath
        self.log = ""

    def _run_cmd(self, cmd: str, args=None):
        command = [
            MYSQLBACKUP_BINARY,
            '--backup-dir='+self.tmppath
        ]

        if args:
            command += args

        if cmd != "copy-back-and-apply-log":
            command += self.storage.get_args() + self.auth_options

        command.append(cmd)

        def print_output(pipe, stream):
            hidden = self.storage.get_filter()

            try:
                for line in iter(pipe.readline, ''):
                    if line:
                        if hidden:
                            # Avoid printing PAR URLs and similar into logs
                            for term in hidden:
                                line = line.replace(term, 'X' * len(term))

                        self.log += line

                        stream.write(line)
                        stream.flush()
            finally:
                pipe.close()

        # in order to pass password via stdin we got to also pipe stdout/stderr
        # through our code, if we end up using localroot (passwordless) we
        # might use the simpler subprocess.run ...
        process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            bufsize=1,
            text=True
        )

        process.stdin.close()

        stdout_thread = threading.Thread(target=print_output, args=(process.stdout, sys.stdout))
        stderr_thread = threading.Thread(target=print_output, args=(process.stderr, sys.stderr))

        stdout_thread.start()
        stderr_thread.start()

        stdout_thread.join()
        stderr_thread.join()

        error_code = process.wait()

        if error_code:
            raise RuntimeError(f"Running Backup failed: {error_code}")

        return True


    def backup(self, incremental=False, incremental_base='history:last_backup'):
        if incremental:
            args = [
                '--incremental', '--incremental-base='+incremental_base,
            ]
            return self._run_cmd('backup-to-image', args)

        return self._run_cmd('backup-to-image')

    def extract(self, image: str):
        command = [
            MYSQLBACKUP_BINARY,
            '--backup-dir='+self.tmppath,
            '--datadir=/var/lib/mysql'
        ]

        subprocess.run(command + self.storage.get_args(image) + ['image-to-backup-dir'], check=True)

    def restore(self, full_image: str, incrementals: list = ()):
        command = [
            #'--defaults-file='+MYCNF_PATH,
            '--backup-dir='+self.tmppath,
            '--datadir=/var/lib/mysql'
        ]

        try:
            self._run_cmd('copy-back-and-apply-log', args=command+self.storage.get_args(full_image))

        except FileNotFoundError as exc:
            if exc.filename == MYSQLBACKUP_BINARY:
                raise NotImplementedError("The MySQL Operator image does " \
                                          "not contain MySQL Enterprise " \
                                          "Backup, please verify you are " \
                                          "using MySQL Enterprise Edition"
                                          ) from exc
            raise
        except  RuntimeError: # subprocess.CalledProcessError:
            sys.exit(1)


        # /var/run/mysqld is a location shared between restore initContainer,
        # which creates this file and sidecar container, which wants to read it
        shutil.copy(self.tmppath+'/meta/backup_variables.txt', '/var/run/mysqld/backup_variables.txt')
        shutil.rmtree(self.tmppath)

        for incremental in incrementals:
            self._run_cmd('copy-back-and-apply-log', args=command+['--incremental']+self.storage.get_args(incremental))
            # This will override hte previous - we are only interested to keep the last
            shutil.copy(self.tmppath+'/meta/backup_variables.txt', '/var/run/mysqld/backup_variables.txt')
            shutil.rmtree(self.tmppath)


    def cleanup(self):
        if hasattr(self.storage, 'clean'):
            self.storage.clean()

        try:
            shutil.rmtree(self.tmppath)
        except Exception as exc:
            print(f"Failed to delete temp directory, this is likely okay: {exc}")

def testmain():
    import argparse

    parser = argparse.ArgumentParser(
                    prog='Backup Test Interrface',
                    description='Create MySQL Enterprise Backup')

    parser.add_argument("target")
    parser.add_argument("-u", "--user")
    parser.add_argument("-p", "--password")
    parser.add_argument("-t", "--host") # argparse wants -h for itself ...
    parser.add_argument("-i", "--incremental", action='store_true')

    args = parser.parse_args()

    storage = MebStorageOCIPAR(args.target, OCIObjectStorage(OCIRequest(), 'jschluet', 'idylsdbcgx0d'))

    meb = MySQLEnterpriseBackup(
        storage,
        [f"-u{args.user}", f"-p{args.password}", f"--host={args.host}"],
        "/tmp/backup-tmp"
    )
    try:
        meb.backup(args.incremental)
    finally:
        meb.cleanup()


if __name__ == "__main__":
    testmain()