"""Nexus — Sync Engine v2 Rsync push from Nexus server to target servers via SSH. Supports: local source path / uploaded ZIP extraction. Parallel execution with Semaphore concurrency control. """ import asyncio import logging import os import re import tempfile import uuid from datetime import datetime, timezone from typing import List, Optional from server.domain.models import Server, SyncLog, AuditLog from server.infrastructure.database.server_repo import ServerRepositoryImpl from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl from server.infrastructure.database.crypto import decrypt_value from server.utils.posix_paths import ( PosixPathError, UPLOAD_STAGING_PREFIX, normalize_remote_abs_path, resolve_nexus_push_source_path, to_posix, ) def _nexus_source_path(path: str) -> str: """Normalize and validate push source path on the Nexus host.""" return resolve_nexus_push_source_path(path) logger = logging.getLogger("nexus.sync_v2") MAX_CONCURRENT = 10 RSYNC_TIMEOUT = 300 # rsync --chown / --chmod value validation (prevent shell injection via .env) _RSYNC_CHOWN_RE = re.compile(r"^[a-zA-Z0-9_.-]+(:[a-zA-Z0-9_.-]+)?$") _RSYNC_CHMOD_RE = re.compile(r"^[DF0-9a-zA-Z=+,:-]+$") def rsync_push_permission_summary(server: Server | None = None) -> str | None: """Human/log snapshot of rsync --chown/--chmod and elevation applied on real pushes.""" from server.config import settings from server.utils.rsync_elevation import rsync_elevation_log_hint parts: list[str] = [] chown = (settings.RSYNC_PUSH_CHOWN or "").strip() chmod = (settings.RSYNC_PUSH_CHMOD or "").strip() if chown and _RSYNC_CHOWN_RE.fullmatch(chown): parts.append(f"chown={chown}") if chmod and _RSYNC_CHMOD_RE.fullmatch(chmod): parts.append(f"chmod={chmod}") if server is not None: elevation_hint = rsync_elevation_log_hint(server) if elevation_hint: parts.append(elevation_hint) return " ".join(parts) if parts else None def rsync_push_permission_args() -> list[str]: """Extra rsync flags: set remote owner/mode after push (default www:www, 755). Requires SSH/rsync receiver to run as root (typical Baota push). Empty env disables. """ from server.config import settings args: list[str] = [] chown = (settings.RSYNC_PUSH_CHOWN or "").strip() chmod = (settings.RSYNC_PUSH_CHMOD or "").strip() if chown: if _RSYNC_CHOWN_RE.fullmatch(chown): args.extend(["--chown", chown]) else: logger.warning("Ignoring invalid NEXUS_RSYNC_PUSH_CHOWN: %r", chown) if chmod: if _RSYNC_CHMOD_RE.fullmatch(chmod): args.extend(["--chmod", chmod]) else: logger.warning("Ignoring invalid NEXUS_RSYNC_PUSH_CHMOD: %r", chmod) return args class SyncEngineV2: """Unified sync engine: rsync push from Nexus + preview + ZIP upload""" def __init__( self, server_repo: ServerRepositoryImpl, sync_log_repo: SyncLogRepositoryImpl, audit_repo: AuditLogRepositoryImpl, retry_repo: PushRetryJobRepositoryImpl, ): self.server_repo = server_repo self.sync_log_repo = sync_log_repo self.audit_repo = audit_repo self.retry_repo = retry_repo # ── Rsync push from Nexus to target servers ── async def sync_files( self, server_ids: List[int], source_path: str, target_path: Optional[str] = None, sync_mode: str = "incremental", operator: str = "admin", batch_size: int = 50, concurrency: int = 10, trigger_type: str = "manual", batch_id: Optional[str] = None, ) -> dict: """Push files from Nexus to target servers via rsync over SSH""" try: source_path = _nexus_source_path(source_path) except PosixPathError as exc: return {"total": 0, "completed": 0, "failed": 0, "results": {}, "error": str(exc)} batch_id = (batch_id or uuid.uuid4().hex[:12])[:12] concurrency = min(concurrency, MAX_CONCURRENT) sem = asyncio.Semaphore(concurrency) servers = await self.server_repo.get_by_ids(server_ids) total = len(servers) completed = 0 failed = 0 cancelled = 0 _counter_lock = asyncio.Lock() _db_lock = asyncio.Lock() results = {} async def _sync_one(server: Server): nonlocal completed, failed, cancelled async with sem: # Check cancel flag before starting try: from server.infrastructure.redis.client import get_redis redis = get_redis() cancel_flag = await redis.get(f"sync:cancel:{batch_id}") if cancel_flag: sync_log = SyncLog( server_id=server.id, source_path=source_path, target_path=target_path or server.target_path or "/tmp/sync", trigger_type=trigger_type, operator=operator, status="cancelled", sync_mode=sync_mode, started_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc), error_message="推送已取消", ) async with _db_lock: sync_log = await self.sync_log_repo.create(sync_log) async with _counter_lock: cancelled += 1 try: from server.api.websocket import broadcast_sync_progress await broadcast_sync_progress( batch_id=batch_id, server_id=server.id, server_name=server.name, status="cancelled", completed=completed, failed=failed, total=total, error_message="推送已取消", ) except Exception as e: logger.debug(f"WS broadcast failed for cancelled server {server.id}: {e}") results[server.id] = (sync_log, None) return except Exception as e: logger.warning(f"Cancel check failed for batch {batch_id}, proceeding with push: {e}") dest = target_path or server.target_path or "/tmp/sync" sync_log = SyncLog( server_id=server.id, source_path=source_path, target_path=dest, trigger_type=trigger_type, operator=operator, status="running", sync_mode=sync_mode, push_permission=rsync_push_permission_summary(server), started_at=datetime.now(timezone.utc), ) async with _db_lock: sync_log = await self.sync_log_repo.create(sync_log) try: from server.api.websocket import broadcast_sync_progress await broadcast_sync_progress( batch_id=batch_id, server_id=server.id, server_name=server.name, status="running", completed=completed, failed=failed, total=total, ) except Exception as e: logger.debug(f"WS running broadcast failed for server {server.id}: {e}") try: result = await _rsync_push(server, source_path, dest, sync_mode) except Exception as exc: logger.exception( "rsync push raised for server %s (%s): %s", server.id, server.name, exc, ) result = {"exit_code": -1, "stdout": "", "stderr": str(exc)} sync_log.status = "success" if result["exit_code"] == 0 else "failed" sync_log.finished_at = datetime.now(timezone.utc) sync_log.duration_seconds = int( (sync_log.finished_at - sync_log.started_at).total_seconds() ) if sync_log.started_at else 0 if result["exit_code"] != 0: from server.utils.sync_error_message import translate_sync_error_message raw_err = (result.get("stderr") or "")[:1000] sync_log.error_message = translate_sync_error_message(raw_err) or raw_err or None else: sync_log.error_message = None if result["exit_code"] == 0: stats = _parse_rsync_stats(result.get("stdout") or "") sync_log.files_total = int(stats.get("files_total") or 0) sync_log.files_transferred = int(stats.get("files_transferred") or 0) sync_log.bytes_transferred = int(stats.get("transfer_size_bytes") or 0) sync_log.diff_summary = (result.get("stdout") or "")[:2000] else: sync_log.diff_summary = None retry_job_id = None async with _db_lock: sync_log = await self.sync_log_repo.update(sync_log) if sync_log.status == "failed": try: from server.domain.models import PushRetryJob from sqlalchemy import select as _select existing = await self.retry_repo.session.execute( _select(PushRetryJob).where( PushRetryJob.server_id == server.id, PushRetryJob.source_path == source_path, PushRetryJob.target_path == dest, PushRetryJob.status == "pending", ) ) if existing.scalars().first() is None: retry_job = PushRetryJob( server_id=server.id, server_name=server.name, operator=operator, source_path=source_path, target_path=dest, status="pending", retry_count=0, max_retries=3, next_retry_at=datetime.now(timezone.utc), last_error=sync_log.error_message[:500] if sync_log.error_message else None, ) retry_job = await self.retry_repo.create(retry_job) retry_job_id = retry_job.id logger.info(f"Auto-created retry job #{retry_job.id} for server {server.id}") else: logger.debug(f"Retry job already pending for server {server.id}, skipping duplicate") except Exception as e: logger.warning(f"Failed to create retry job for server {server.id}: {e}") async with _counter_lock: if sync_log.status == "success": completed += 1 else: failed += 1 # Broadcast per-server progress via WebSocket try: from server.api.websocket import broadcast_sync_progress await broadcast_sync_progress( batch_id=batch_id, server_id=server.id, server_name=server.name, status=sync_log.status, completed=completed, failed=failed, total=total, error_message=sync_log.error_message[:200] if sync_log.error_message else None, duration_seconds=sync_log.duration_seconds, retry_job_id=retry_job_id, cancelled=cancelled, ) except Exception as e: logger.debug(f"WS broadcast failed for server {server.id}: {e}") results[server.id] = (sync_log, retry_job_id) tasks = [asyncio.create_task(_sync_one(s)) for s in servers] task_outcomes = await asyncio.gather(*tasks, return_exceptions=True) for outcome in task_outcomes: if isinstance(outcome, BaseException): logger.error("sync_files worker raised: %s", outcome, exc_info=outcome) unaccounted = total - len(results) if unaccounted > 0: failed += unaccounted logger.error( "sync_files: %s/%s servers missing results (task exception or early exit)", unaccounted, total, ) await self._audit( "sync_files", "sync", 0, f"文件推送: {source_path} → {total} 台服务器 ({completed} 成功, {failed} 失败)", operator, ) # Telegram notification for push completion try: from server.infrastructure.telegram import send_telegram_sync_complete failed_names = [] for sid, (log, _) in results.items(): if log.status == "failed": srv = next((s for s in servers if s.id == sid), None) failed_names.append(srv.name if srv else f"#{sid}") # Calculate total duration from first start to last finish durations = [log.duration_seconds for _, (log, _) in results.items() if log.duration_seconds] total_duration = max(durations) if durations else None await send_telegram_sync_complete( completed=completed, failed=failed, total=total, source_path=source_path, operator=operator, failed_servers=failed_names if failed_names else None, duration_seconds=total_duration, ) except Exception as e: logger.warning(f"Telegram sync complete notification failed: {e}") # Build results with retry_job_id result_dicts = {} for sid, (log, retry_jid) in results.items(): d = _log_to_dict(log) if retry_jid: d["retry_job_id"] = retry_jid result_dicts[sid] = d _maybe_cleanup_upload_staging(source_path, trigger_type, failed, cancelled) return { "total": total, "completed": completed, "failed": failed, "cancelled": cancelled, "batch_id": batch_id, "results": result_dicts, } # ── Preview (dry-run) ── async def preview_sync( self, server_id: int, source_path: str, target_path: Optional[str] = None, sync_mode: str = "incremental", verbose: bool = False, ) -> dict: """Dry-run rsync --dry-run --stats against one server""" server = await self.server_repo.get_by_id(server_id) if not server: return {"error": "Server not found", "exit_code": -1} try: source_path = _nexus_source_path(source_path) except PosixPathError as exc: return {"error": str(exc), "exit_code": -1} dest = target_path or server.target_path or "/tmp/sync" result = await _rsync_push(server, source_path, dest, sync_mode, dry_run=True, verbose=verbose) if result["exit_code"] not in (0, 23): from server.utils.sync_error_message import translate_sync_error_message raw_err = (result.get("stderr") or "")[:500] or f"rsync exited with code {result['exit_code']}" return { "server_id": server_id, "server_name": server.name, "error": translate_sync_error_message(raw_err) or raw_err, "exit_code": result["exit_code"], } stdout = result.get("stdout", "") stats = _parse_rsync_stats(stdout) files: list[str] = [] files_truncated = False if verbose: all_lines = [ l for l in stdout.split("\n") if l.strip() and not l.startswith("Number of") and not l.startswith("Total") and not l.startswith("Literal") and not l.startswith("Matched") and not l.startswith("File list") and not l.startswith("sent ") and not l.startswith("speedup") and not l.startswith("sending") ] files = all_lines[:200] files_truncated = len(all_lines) > 200 return { "server_id": server_id, "server_name": server.name, "source_path": source_path, "target_path": dest, "sync_mode": sync_mode, "dry_run": True, "stats": stats, "files": files, "files_truncated": files_truncated, "exit_code": result["exit_code"], } async def _audit(self, action: str, target_type: str, target_id: int, detail: str, operator: str = "admin"): try: audit_log = AuditLog( admin_username=operator, action=action, target_type=target_type, target_id=target_id, detail=detail, ) await self.audit_repo.create(audit_log) except Exception as e: logger.error(f"Audit log failed: {e}") # ── Rsync execution on Nexus host ── async def _rsync_push( server: Server, source_path: str, target_path: str, sync_mode: str = "incremental", dry_run: bool = False, verbose: bool = False, ) -> dict: """Execute rsync on Nexus host, pushing to target server via SSH. Non-root servers with ``files_elevation`` not ``off`` use ``--rsync-path=sudo -n rsync`` (default ``auto_sudo``, same as file manager). """ from server.utils.rsync_elevation import rsync_receiver_path_for_push sudo_path = rsync_receiver_path_for_push(server) result = await _execute_rsync_push( server, source_path, target_path, sync_mode=sync_mode, dry_run=dry_run, verbose=verbose, rsync_receiver_path=sudo_path, ) result["elevated"] = bool(sudo_path and result.get("exit_code") == 0) return result async def _execute_rsync_push( server: Server, source_path: str, target_path: str, *, sync_mode: str = "incremental", dry_run: bool = False, verbose: bool = False, rsync_receiver_path: str | None = None, ) -> dict: """Run one rsync attempt on the Nexus host. Returns {exit_code, stdout, stderr}.""" import shlex ssh_user = server.username or "root" ssh_host = server.domain ssh_port = server.port or 22 remote_dest = normalize_remote_abs_path(to_posix(target_path)) rsync_remote = f"{ssh_user}@{ssh_host}:{remote_dest}" args = ["rsync", "-az"] if sync_mode == "full": args.append("--delete") elif sync_mode == "checksum": args.append("--checksum") elif sync_mode == "overwrite": args.append("--inplace") if dry_run: args.extend(["--dry-run", "--stats"]) else: args.append("--stats") if verbose: args.append("-v") if not dry_run: args.extend(rsync_push_permission_args()) if rsync_receiver_path: args.extend(["--rsync-path", rsync_receiver_path]) ssh_opts = f"ssh -o StrictHostKeyChecking=no -p {ssh_port}" key_file = None env_override = None try: if server.auth_method == "password" and server.password: password = decrypt_value(server.password) env_override = {"SSHPASS": password} args = ["sshpass", "-e"] + args ssh_opts += " -o PubkeyAuthentication=no" elif server.ssh_key_configured and server.ssh_key_private: key_content = decrypt_value(server.ssh_key_private) key_file = tempfile.NamedTemporaryFile( mode="w", suffix=".key", prefix=f"nexus_rsync_{server.id}_", delete=False, ) key_file.write(key_content) key_file.close() os.chmod(key_file.name, 0o600) ssh_opts += f" -i {shlex.quote(key_file.name)}" elif server.ssh_key_path: ssh_opts += f" -i {shlex.quote(server.ssh_key_path)}" else: return {"exit_code": -1, "stdout": "", "stderr": f"服务器 {server.name} 无有效 SSH 凭据"} args.extend(["-e", ssh_opts]) args.append(source_path.rstrip("/") + "/") args.append(rsync_remote.rstrip("/") + "/") proc = await asyncio.create_subprocess_exec( *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env={**os.environ, **(env_override or {})}, ) try: stdout, stderr = await asyncio.wait_for( proc.communicate(), timeout=RSYNC_TIMEOUT + 30 ) except asyncio.TimeoutError: proc.kill() await proc.communicate() return {"exit_code": -1, "stdout": "", "stderr": f"rsync 超时 ({RSYNC_TIMEOUT}s)"} return { "exit_code": proc.returncode or 0, "stdout": stdout.decode("utf-8", errors="replace")[:10000], "stderr": stderr.decode("utf-8", errors="replace")[:10000], } finally: if key_file: try: os.unlink(key_file.name) except OSError: pass def _maybe_cleanup_upload_staging( source_path: str, trigger_type: str, failed: int, cancelled: int, ) -> None: """Remove ZIP extract dir after a fully successful manual push (retry needs the tree if failed).""" if trigger_type != "manual" or failed > 0 or cancelled > 0: return normalized = _nexus_source_path(source_path) if not normalized.startswith(UPLOAD_STAGING_PREFIX): return staging_dir = normalized.rstrip("/") if not os.path.isdir(staging_dir): return import shutil shutil.rmtree(staging_dir, ignore_errors=True) logger.info("Cleaned upload staging dir after push: %s", staging_dir) def _parse_rsync_stats(output: str) -> dict: """Extract key numbers from rsync --stats output.""" stats: dict = {} patterns = { "files_total": r"Number of files:\s+([\d,]+)", "files_created": r"Number of created files:\s+([\d,]+)", "files_deleted": r"Number of deleted files:\s+([\d,]+)", "files_transferred": r"Number of regular files transferred:\s+([\d,]+)", "total_size_bytes": r"Total file size:\s+([\d,]+)\s+bytes", "transfer_size_bytes": r"Total transferred file size:\s+([\d,]+)\s+bytes", } for key, pattern in patterns.items(): m = re.search(pattern, output) if m: try: stats[key] = int(m.group(1).replace(",", "")) except ValueError: stats[key] = m.group(1) return stats def _log_to_dict(log: SyncLog) -> dict: return { "id": log.id, "server_id": log.server_id, "status": log.status, "sync_mode": log.sync_mode, "files_transferred": log.files_transferred, "files_skipped": log.files_skipped, "bytes_transferred": log.bytes_transferred, "diff_summary": log.diff_summary, "push_permission": log.push_permission, "error_message": log.error_message, "started_at": str(log.started_at) if log.started_at else None, "finished_at": str(log.finished_at) if log.finished_at else None, "duration_seconds": log.duration_seconds, }