# coding: utf-8 # OceanBase Deploy. # Copyright (C) 2021 OceanBase # # This file is part of OceanBase Deploy. # # OceanBase Deploy is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # OceanBase Deploy is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with OceanBase Deploy. If not, see . from __future__ import absolute_import, division, print_function import os import sys import getpass import warnings from copy import deepcopy from subprocess32 import Popen, PIPE # paramiko import cryptography 模块在python2下会报不支持警报 warnings.filterwarnings("ignore") from paramiko import AuthenticationException, SFTPClient from paramiko.client import SSHClient, AutoAddPolicy from paramiko.ssh_exception import NoValidConnectionsError __all__ = ("SshClient", "SshConfig", "LocalClient") class SshConfig(object): def __init__(self, host, username='root', password=None, key_filename=None, port=22, timeout=30): self.host = host self.username = username self.password = password self.key_filename = key_filename self.port = port self.timeout = timeout def __str__(self): return '%s@%s' % (self.username ,self.host) class SshReturn(object): def __init__(self, code, stdout, stderr): self.code = code self.stdout = stdout self.stderr = stderr def __bool__(self): return self.code == 0 def __nonzero__(self): return self.__bool__() class LocalClient(object): @staticmethod def execute_command(command, env=None, timeout=None, stdio=None): stdio and getattr(stdio, 'verbose', print)('local execute: %s ' % command, end='') try: p = Popen(command, env=env, shell=True, stdout=PIPE, stderr=PIPE) output, error = p.communicate(timeout=timeout) code = p.returncode output = output.decode(errors='replace') error = error.decode(errors='replace') verbose_msg = 'exited code %s' % code if code: verbose_msg += ', error output:\n%s' % error stdio and getattr(stdio, 'verbose', print)(verbose_msg) except Exception as e: output = '' error = str(e) code = 255 verbose_msg = 'exited code 255, error output:\n%s' % error stdio and getattr(stdio, 'verbose', print)(verbose_msg) stdio and getattr(stdio, 'exception', print)('') return SshReturn(code, output, error) @staticmethod def put_file(local_path, remote_path, stdio=None): if LocalClient.execute_command('mkdir -p %s && cp -f %s %s' % (os.path.dirname(remote_path), local_path, remote_path), stdio=stdio): return True return False @staticmethod def put_dir(local_dir, remote_dir, stdio=None): if LocalClient.execute_command('mkdir -p && cp -fr %s %s' % (os.path.dirname(remote_dir), local_dir, remote_dir), stdio=stdio): return True return False class SshClient(object): def __init__(self, config, stdio=None): self.config = config self.stdio = stdio self.sftp = None self.is_connected = False self.ssh_client = SSHClient() self.env_str = '' if self._is_local(): self.env = deepcopy(os.environ.copy()) else: self.env = {'PATH': '/sbin:/usr/local/bin:/usr/bin:/usr/local/sbin:/usr/sbin:'} self._update_env() def _update_env(self): env = [] for key in self.env: if self.env[key]: env.append('export %s=%s$%s;' % (key, self.env[key], key)) self.env_str = ''.join(env) def add_env(self, key, value, rewrite=False, stdio=None): stdio = stdio if stdio else self.stdio if key not in self.env or not self.env[key] or rewrite: stdio and getattr(stdio, 'verbose', print)('%s@%s set env %s to \'%s\'' % (self.config.username, self.config.host, key, value)) self.env[key] = value else: stdio and getattr(stdio, 'verbose', print)('%s@%s append \'%s\' to %s' % (self.config.username, self.config.host, value, key)) self.env[key] += value self._update_env() def get_env(self, key): return self.env[key] if key in self.env else None def __str__(self): return '%s@%s:%d' % (self.config.username, self.config.host, self.config.port) def _is_local(self): return self.is_localhost() and self.config.username == getpass.getuser() def is_localhost(self, stdio=None): return self.config.host in ['127.0.0.1', 'localhost', '127.1', '127.0.1'] def _login(self, stdio=None): if self.is_connected: return True stdio = stdio if stdio else self.stdio try: self.ssh_client.set_missing_host_key_policy(AutoAddPolicy()) self.ssh_client.connect( self.config.host, port=self.config.port, username=self.config.username, password=self.config.password, key_filename=self.config.key_filename, timeout=self.config.timeout ) except AuthenticationException: stdio and getattr(stdio, 'exception', print)('') stdio and getattr(stdio, 'critical', print)('%s@%s username or password error' % (self.config.username, self.config.host)) return False except NoValidConnectionsError: stdio and getattr(stdio, 'exception', print)('') stdio and getattr(stdio, 'critical', print)('%s@%s connect failed: time out' % (self.config.username, self.config.host)) return False except Exception as e: stdio and getattr(stdio, 'exception', print)('') stdio and getattr(stdio, 'critical', print)('%s@%s connect failed: %s' % (self.config.username, self.config.host, e)) return False self.is_connected = True return True def _open_sftp(self, stdio=None): if self.sftp: return True if self._login(stdio): SFTPClient.from_transport(self.ssh_client.get_transport()) self.sftp = self.ssh_client.open_sftp() return True return False def connect(self, stdio=None): if self._is_local(): return True return self._login(stdio) def reconnect(self, stdio=None): self.close(stdio) return self.connect(stdio) def close(self, stdio=None): if self._is_local(): return True if self.is_connected: self.ssh_client.close() if self.sftp: self.sftp = None def __del__(self): self.close() def execute_command(self, command, stdio=None): if self._is_local(): return LocalClient.execute_command(command, self.env, self.config.timeout, stdio) if not self._login(stdio): return SshReturn(255, '', 'connect failed') stdio = stdio if stdio else self.stdio verbose_msg = '%s execute: %s ' % (self.config, command) stdio and getattr(stdio, 'verbose', print)(verbose_msg, end='') command = '%s %s;echo -e "\n$?\c"' % (self.env_str, command.strip(';')) stdin, stdout, stderr = self.ssh_client.exec_command(command) output = stdout.read().decode(errors='replace') error = stderr.read().decode(errors='replace') idx = output.rindex('\n') code = int(output[idx:]) verbose_msg = 'exited code %s' % code if code: verbose_msg += ', error output:\n%s' % error stdio and getattr(stdio, 'verbose', print)(verbose_msg) return SshReturn(code, output[:idx], error) def put_file(self, local_path, remote_path, stdio=None): stdio = stdio if stdio else self.stdio if self._is_local(): return LocalClient.put_file(local_path, remote_path, stdio) if not os.path.isfile(local_path): stdio and getattr(stdio, 'critical', print)('%s is not file' % local_path) return False if not self._open_sftp(stdio): return False return self._put_file(local_path, remote_path, stdio) def _put_file(self, local_path, remote_path, stdio=None): if self.execute_command('mkdir -p %s && rm -fr %s' % (os.path.dirname(remote_path), remote_path), stdio): stdio and getattr(stdio, 'verbose', print)('send %s to %s' % (local_path, remote_path)) if self.sftp.put(local_path, remote_path): return self.execute_command('chmod %s %s' % (oct(os.stat(local_path).st_mode)[-3: ], remote_path)) return False def put_dir(self, local_dir, remote_dir, stdio=None): stdio = stdio if stdio else self.stdio if self._is_local(): return LocalClient.put_dir(local_dir, remote_dir, stdio) if not self._open_sftp(stdio): return False if not self.execute_command('mkdir -p %s' % remote_dir, stdio): return False failed = [] failed_dirs = [] local_dir_path_len = len(local_dir) for root, dirs, files in os.walk(local_dir): for path in failed_dirs: if root.find(path) == 0: # 父目录已经在被标记为失败,该层可直接跳过 # break退出不执行else代码段 break else: for name in files: local_path = os.path.join(root, name) remote_path = os.path.join(remote_dir, root[local_dir_path_len:].lstrip('/'), name) if not self._put_file(local_path, remote_path, stdio): failed.append(remote_path) for name in dirs: local_path = os.path.join(root, name) remote_path = os.path.join(remote_dir, root[local_dir_path_len:].lstrip('/'), name) if not self.execute_command('mkdir -p %s' % remote_path, stdio): failed_dirs.append(local_dir) failed.append(remote_path) for path in failed: stdio and getattr(stdio, 'critical', print)('send %s to %s@%s failed' % (path, self.config.username, self.config.host)) return True