巧用Python登陸遠(yuǎn)程服務(wù)器

$ ssh-keygen -t rsa

Generating a public/private rsa key pair. Enter the file in which you wish to save they key (i.e., /home/username/.ssh/id_rsa):

$ ssh-copy-id -i ~/.ssh/mykey username@my_remote_host.org

$ cd ~/.ssh

-----BEGIN RSA PRIVATE KEY-----...-----END RSA PRIVATE KEY-----

...class RemoteClient:    """Client to interact with a remote host via SSH & SCP."""    ...    def _connect(self):        """Open connection to remote host."""        if self.conn is None:            try:                self.client = SSHClient()                self.client.load_system_host_keys()                self.client.set_missing_host_key_policy(                    AutoAddPolicy()                )                self.client.connect(                    self.host,                    username=self.user,                    key_filename=self.ssh_key_filepath,                    look_for_keys=True,                    timeout=5000                )                self.scp = SCPClient(self.client.get_transport())            except AuthenticationException as error:                logger.error(f'Authentication failed: \                    did you remember to create an SSH key? {error}')                raise error        return self.client

class RemoteClient:    ...    def disconnect(self):        """Close ssh connection."""        if self.client:            self.client.close()        if self.scp:            self.scp.close()

class RemoteClient:    ...    def execute_commands(self, commands):        """        Execute multiple commands in succession.        :param commands: List of unix commands as strings.        :type commands: List[str]        """        self.conn = self._connect()        for cmd in commands:            stdin, stdout, stderr = self.client.exec_command(cmd)            stdout.channel.recv_exit_status()            response = stdout.readlines()            for line in response:                logger.info(f'INPUT: {cmd} | OUTPUT: {line}')

remote.execute_commands(['cd /var/www/ && ls',                        'tail /var/log/nginx/access.log',                        'ps aux | grep node'])

self.scp = SCPClient(self.client.get_transport())

class RemoteClient:    ...
    def bulk_upload(self, files):        """        Upload multiple files to a remote directory.
        :param files: List of paths to local files.        :type files: List[str]        """        self.conn = self._connect()        uploads = [self._upload_single_file(file) for file in files]        logger.info(f'Finished uploading {len(uploads)} files to {self.remote_path} on {self.host}')
    def _upload_single_file(self, file):        """Upload a single file to a remote directory."""        upload = None        try:            self.scp.put(                file,                recursive=True,                remote_path=self.remote_path            )            upload = file        except SCPException as error:            logger.error(error)            raise error        finally:            logger.info(f'Uploaded {file} to {self.remote_path}')            return upload

class RemoteClient:    ...    def download_file(self, file):        """Download file from remote host."""        self.conn = self._connect()        self.scp.get(file)