import shutil import os import subprocess import tempfile import zipfile import boto3 import paramiko import posixpath from datetime import datetime, timedelta try: from .validation import is_valid_db_identifier except ImportError: from validation import is_valid_db_identifier MYSQL_SYSTEM_DATABASES = {'information_schema', 'performance_schema', 'mysql', 'sys'} def perform_mysql_backup(mysql_config, db_mode, db_name=None, db_list=None, backup_dir='/tmp/backups', prefix='mysql_backup'): """ Perform a MySQL backup using mysqldump. :param mysql_config: dict with keys: host, port, user, password (and optionally charset, socket) :param db_mode: 'all', 'single', or 'list' :param db_name: database name for mode 'single' :param db_list: comma-separated database names for mode 'list' :param backup_dir: local directory where backup zip will be stored :param prefix: filename prefix :returns: (success, message, output_path) """ # Validate mode if db_mode not in ('all', 'single', 'list'): return False, f"Invalid db_mode '{db_mode}': must be 'all', 'single', or 'list'", None # Determine target databases if db_mode == 'single': if not db_name or not db_name.strip(): return False, "db_mode 'single' requires a non-empty db_name", None db_name = db_name.strip() if not is_valid_db_identifier(db_name): return False, "Invalid database name", None databases = [db_name] elif db_mode == 'list': if not db_list or not db_list.strip(): return False, "db_mode 'list' requires a non-empty db_list", None databases = [d.strip() for d in db_list.split(',') if d.strip()] if not databases: return False, "db_mode 'list' requires at least one database name", None invalid = [db for db in databases if not is_valid_db_identifier(db)] if invalid: return False, f"Invalid database name(s): {', '.join(invalid)}", None else: # all databases = None # resolved later if not os.path.exists(backup_dir): os.makedirs(backup_dir) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") zip_filename = f"{prefix}_{timestamp}.zip" zip_path = os.path.join(backup_dir, zip_filename) host = mysql_config.get('host', '127.0.0.1') port = str(mysql_config.get('port', 3306)) user = mysql_config.get('user', 'root') password = mysql_config.get('password', '') socket = mysql_config.get('socket') # Resolve 'all' databases by querying MySQL if db_mode == 'all': try: import pymysql connect_kwargs = dict( host=host, port=int(port), user=user, password=password, charset=mysql_config.get('charset', 'utf8mb4'), cursorclass=pymysql.cursors.DictCursor, ) if socket: connect_kwargs['unix_socket'] = socket conn = pymysql.connect(**connect_kwargs) with conn.cursor() as cursor: cursor.execute("SHOW DATABASES") databases = [ row['Database'] for row in cursor.fetchall() if row['Database'] not in MYSQL_SYSTEM_DATABASES and is_valid_db_identifier(row['Database']) ] conn.close() except Exception as e: return False, f"Failed to list databases: {e}", None if not databases: return False, "No databases to backup", None print(f"[MySQL Backup] Selected databases: {', '.join(databases)}") results = {} failed = [] with zipfile.ZipFile(zip_path, 'w', compression=zipfile.ZIP_DEFLATED) as zf: for db in databases: dump_filename = f"{db}_{timestamp}.sql" try: cmd = ['mysqldump'] if socket: cmd += ['--socket', socket] else: cmd += ['--host', host, '--port', port] cmd += ['--user', user, '--single-transaction', '--routines', '--triggers', db] env = os.environ.copy() env['MYSQL_PWD'] = password # avoid password in process list result = subprocess.run( cmd, capture_output=True, text=True, env=env ) if result.returncode != 0: err = result.stderr.strip() print(f"[MySQL Backup] FAILED {db}: {err}") results[db] = f"FAILED: {err}" failed.append(db) else: zf.writestr(dump_filename, result.stdout) print(f"[MySQL Backup] OK {db} -> {dump_filename}") results[db] = "OK" except Exception as e: print(f"[MySQL Backup] ERROR {db}: {e}") results[db] = f"ERROR: {e}" failed.append(db) summary_lines = [f"{db}: {status}" for db, status in results.items()] summary = "; ".join(summary_lines) if failed and len(failed) == len(databases): # All failed — remove the empty/useless zip if os.path.exists(zip_path): os.remove(zip_path) return False, f"All databases failed. {summary}", None print(f"[MySQL Backup] Archive: {zip_path}") if failed: return True, f"Completed with errors ({len(failed)} failed). {summary}", zip_path return True, f"All {len(databases)} database(s) backed up successfully. {summary}", zip_path def create_local_backup(source_path, backup_dir='/tmp/backups', prefix='backup'): if not os.path.exists(backup_dir): os.makedirs(backup_dir) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{prefix}_{timestamp}" archive_path = shutil.make_archive(os.path.join(backup_dir, filename), 'zip', source_path) return archive_path def upload_to_s3(file_path, config): try: endpoint = config.get('endpoint') # Optional for generic S3 s3 = boto3.client( 's3', endpoint_url=endpoint if endpoint else None, aws_access_key_id=config['access_key'], aws_secret_access_key=config['secret_key'], region_name=config.get('region', 'us-east-1') ) file_name = os.path.basename(file_path) bucket = config.get('bucket') or config.get('bucket_name') s3.upload_file(file_path, bucket, file_name) return True, "Upload successful" except Exception as e: return False, str(e) def upload_to_sftp(file_path, config): try: transport = paramiko.Transport((config['host'], int(config.get('port', 22)))) # Support Key or Password if config.get('key'): pkey = paramiko.RSAKey.from_private_key_file(config['key']) transport.connect(username=config['user'], pkey=pkey) else: transport.connect(username=config['user'], password=config['pass']) sftp = paramiko.SFTPClient.from_transport(transport) file_name = os.path.basename(file_path) remote_dir = config.get('remote_path', '.') remote_path = posixpath.join(remote_dir, file_name) sftp.put(file_path, remote_path) sftp.close() transport.close() return True, "Upload successful" except Exception as e: return False, str(e) def cleanup_local_backups(dest_dir, prefix, retention_days, retention_count): if not os.path.isdir(dest_dir): return candidates = [] for name in os.listdir(dest_dir): if not name.startswith(prefix + "_") or not name.endswith(".zip"): continue path = os.path.join(dest_dir, name) if os.path.isfile(path): candidates.append((path, datetime.fromtimestamp(os.path.getmtime(path)))) _cleanup_sorted_items(candidates, retention_days, retention_count, lambda item: os.remove(item[0])) def cleanup_sftp_backups(config, prefix, retention_days, retention_count): transport = paramiko.Transport((config['host'], int(config.get('port', 22)))) try: if config.get('key'): pkey = paramiko.RSAKey.from_private_key_file(config['key']) transport.connect(username=config['user'], pkey=pkey) else: transport.connect(username=config['user'], password=config['pass']) sftp = paramiko.SFTPClient.from_transport(transport) try: remote_dir = config.get('remote_path', '.') candidates = [] for entry in sftp.listdir_attr(remote_dir): if entry.filename.startswith(prefix + "_") and entry.filename.endswith(".zip"): remote_path = posixpath.join(remote_dir, entry.filename) candidates.append((remote_path, datetime.fromtimestamp(entry.st_mtime))) _cleanup_sorted_items(candidates, retention_days, retention_count, lambda item: sftp.remove(item[0])) finally: sftp.close() finally: transport.close() def cleanup_s3_backups(config, prefix, retention_days, retention_count): endpoint = config.get('endpoint') s3 = boto3.client( 's3', endpoint_url=endpoint if endpoint else None, aws_access_key_id=config['access_key'], aws_secret_access_key=config['secret_key'], region_name=config.get('region', 'us-east-1') ) bucket = config.get('bucket') or config.get('bucket_name') paginator = s3.get_paginator('list_objects_v2') candidates = [] for page in paginator.paginate(Bucket=bucket): for item in page.get('Contents', []): key = item['Key'] name = os.path.basename(key) if name.startswith(prefix + "_") and name.endswith(".zip"): candidates.append((key, item['LastModified'].replace(tzinfo=None))) _cleanup_sorted_items(candidates, retention_days, retention_count, lambda item: s3.delete_object(Bucket=bucket, Key=item[0])) def _cleanup_sorted_items(candidates, retention_days, retention_count, delete_fn): if not candidates: return cutoff = datetime.now() - timedelta(days=retention_days) if retention_days is not None and retention_days > 0 else None candidates.sort(key=lambda i: i[1], reverse=True) for idx, item in enumerate(candidates): is_too_old = bool(cutoff and item[1] < cutoff) exceeds_count = bool(retention_count is not None and retention_count > 0 and idx >= retention_count) if is_too_old or exceeds_count: delete_fn(item) def perform_backup_job(job): staging_dir = tempfile.mkdtemp(prefix="limristem-web-backup-") archive = None try: prefix = f"backup_job{job.id}" if getattr(job, 'id', None) else "backup" archive = create_local_backup(job.source_path, backup_dir=staging_dir, prefix=prefix) import json config = json.loads(job.dest_config or '{}') retention_days = int(getattr(job, 'retention_days', None) or 0) retention_count = int(getattr(job, 'retention_count', None) or 0) if job.dest_type == 's3': success, msg = upload_to_s3(archive, config) elif job.dest_type == 'sftp': success, msg = upload_to_sftp(archive, config) elif job.dest_type == 'local': try: dest_dir = config.get('path', '/tmp') if not os.path.isabs(dest_dir): return False, "Local backup destination must be an absolute path" if not os.path.exists(dest_dir): os.makedirs(dest_dir) shutil.copy2(archive, os.path.join(dest_dir, os.path.basename(archive))) success, msg = True, "Local copy successful" except Exception as e: success, msg = False, str(e) else: success, msg = False, "Unknown destination type" if success: try: if job.dest_type == 'local': cleanup_local_backups(config.get('path', '/tmp'), prefix, retention_days, retention_count) elif job.dest_type == 'sftp': cleanup_sftp_backups(config, prefix, retention_days, retention_count) elif job.dest_type == 's3': cleanup_s3_backups(config, prefix, retention_days, retention_count) except Exception as e: msg = f"{msg} (retention cleanup warning: {e})" return success, msg except Exception as e: return False, str(e) finally: if archive and os.path.exists(archive): try: os.remove(archive) except OSError: pass shutil.rmtree(staging_dir, ignore_errors=True)