Paramiko – SSH-клиент

Введение

В условиях современных DevOps- и SecOps-практик программный доступ к удалённым серверам стал необходимостью. Одним из самых надёжных и мощных способов взаимодействия с удалёнными хостами через Python является библиотека Paramiko. Она предоставляет реализацию протокола SSH2, позволяя подключаться к серверам, выполнять команды, передавать файлы и строить автоматизированные системы управления инфраструктурой.

В этой статье мы рассмотрим все аспекты работы с Paramiko: от базовых понятий до конкретных реализаций. Подробно разберём классы, методы, функции и практические кейсы использования в реальных проектах.

Что такое Paramiko

Paramiko — это Python-имплементация протокола SSH2, разработанная Робби Уокером. Название происходит от «parallel» + «miko» (мико — жрица в японской культуре). Основной задачей библиотеки является обеспечение защищённого канала передачи данных между хостами с возможностью аутентификации через пароли или ключи.

Библиотека написана на чистом Python, что делает её кроссплатформенной и легко интегрируемой в любые проекты. Paramiko широко используется в таких инструментах, как Ansible, Fabric, и других системах автоматизации.

Ключевые особенности и преимущества

Полная реализация SSH2 протокола

Paramiko поддерживает все основные возможности SSH2, включая:

• Аутентификацию по паролю и ключам
• Выполнение команд на удалённых серверах
• Передачу файлов через SFTP
• Создание защищённых туннелей

Поддержка современных алгоритмов шифрования

• RSA, DSS, ECDSA, Ed25519 ключи
• AES, 3DES, Blowfish шифрование
• SHA-1, SHA-256, SHA-512 хэширование

Дополнительные возможности

• Совместимость с Python 3.6+
• Поддержка jump-серверов (bastion hosts)
• Возможность создания собственных SSH-серверов
• Интеграция с SSH-агентами
• Высокий уровень надёжности и безопасности

Установка и настройка

Стандартная установка

pip install paramiko

Установка с дополнительными зависимостями

pip install paramiko[gssapi]  # Для Kerberos/GSSAPI
pip install paramiko[invoke]  # Для интеграции с Invoke

Проверка установки

import paramiko
print(paramiko.__version__)

Архитектура библиотеки

Paramiko построена на модульной архитектуре с четким разделением ответственности между компонентами:

Основные компоненты

• Transport: Низкоуровневый транспортный слой SSH
• Channel: Абстракция для каналов передачи данных
• SSHClient: Высокоуровневый интерфейс для SSH-соединений
• SFTPClient: Клиент для работы с файлами через SFTP
• ServerInterface: Интерфейс для создания SSH-серверов

Основные классы и их методы

SSHClient - основной интерфейс

SSHClient является основным классом для работы с SSH-соединениями. Он предоставляет высокоуровневый интерфейс для подключения к серверам и выполнения команд.

Основные методы SSHClient

Пример базового использования SSHClient

import paramiko

# Создание клиента
client = paramiko.SSHClient()

# Настройка политики для неизвестных хостов
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# Подключение к серверу
client.connect(
    hostname='192.168.0.1',
    username='user',
    password='password',
    port=22,
    timeout=10
)

# Выполнение команды
stdin, stdout, stderr = client.exec_command('ls -la')

# Получение результата
output = stdout.read().decode('utf-8')
errors = stderr.read().decode('utf-8')

print("Вывод команды:")
print(output)

if errors:
    print("Ошибки:")
    print(errors)

# Закрытие соединения
client.close()

SFTPClient - работа с файлами

SFTPClient предоставляет возможности для работы с файлами и директориями на удалённом сервере через протокол SFTP.

Основные методы SFTPClient

Пример работы с SFTP

import paramiko
import os

# Создание SSH-клиента
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('hostname', username='user', password='password')

# Создание SFTP-клиента
sftp = ssh.open_sftp()

# Загрузка файла
sftp.put('local_file.txt', '/remote/path/file.txt')

# Скачивание файла
sftp.get('/remote/path/file.txt', 'downloaded_file.txt')

# Работа с директориями
try:
    sftp.mkdir('/remote/new_directory')
    print("Директория создана")
except IOError:
    print("Директория уже существует")

# Получение списка файлов
files = sftp.listdir('/remote/path')
print("Файлы в директории:", files)

# Получение подробной информации о файлах
for file_attr in sftp.listdir_attr('/remote/path'):
    print(f"Файл: {file_attr.filename}, Размер: {file_attr.st_size}")

# Закрытие соединений
sftp.close()
ssh.close()

Transport - низкоуровневое управление

Transport предоставляет низкоуровневый доступ к SSH-соединению и позволяет более точно управлять сессиями.

Основные методы Transport

Классы ключей

Paramiko поддерживает различные типы криптографических ключей:

RSAKey - RSA ключи

DSAKey, ECDSAKey, Ed25519Key

Имеют аналогичный интерфейс с RSAKey, но работают с соответствующими алгоритмами.

Методы аутентификации

Аутентификация по паролю

client.connect(
    hostname='server.example.com',
    username='user',
    password='secure_password'
)

Аутентификация по ключу

# Загрузка ключа из файла
key = paramiko.RSAKey.from_private_key_file('/path/to/private_key')

# Подключение с ключом
client.connect(
    hostname='server.example.com',
    username='user',
    pkey=key
)

Аутентификация с защищённым ключом

# Ключ с паролем
key = paramiko.RSAKey.from_private_key_file(
    '/path/to/encrypted_key',
    password='key_password'
)

client.connect(
    hostname='server.example.com',
    username='user',
    pkey=key
)

Использование SSH-агента

# Получение ключей из SSH-агента
agent = paramiko.Agent()
agent_keys = agent.get_keys()

if agent_keys:
    client.connect(
        hostname='server.example.com',
        username='user',
        pkey=agent_keys[0]
    )

Политики безопасности

AutoAddPolicy - автоматическое добавление хостов

client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

RejectPolicy - отклонение неизвестных хостов (по умолчанию)

client.set_missing_host_key_policy(paramiko.RejectPolicy())

WarningPolicy - предупреждение о неизвестных хостах

client.set_missing_host_key_policy(paramiko.WarningPolicy())

Пользовательская политика

class CustomPolicy(paramiko.MissingHostKeyPolicy):
    def missing_host_key(self, client, hostname, key):
        # Логика проверки хоста
        print(f"Новый хост: {hostname}")
        return True  # Разрешить подключение

client.set_missing_host_key_policy(CustomPolicy())

Работа с каналами

Создание интерактивной оболочки

# Создание shell-сессии
shell = client.invoke_shell()

# Отправка команды
shell.send('ls -la\n')

# Получение ответа
import time
time.sleep(1)
output = shell.recv(1024).decode()
print(output)

# Закрытие канала
shell.close()

Работа с псевдо-терминалом

stdin, stdout, stderr = client.exec_command('sudo apt update', get_pty=True)

Обработка ошибок и исключений

Основные исключения

Пример обработки исключений

import paramiko

try:
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect('hostname', username='user', password='wrong_password')
    
except paramiko.AuthenticationException:
    print("Ошибка аутентификации")
except paramiko.SSHException as e:
    print(f"SSH ошибка: {e}")
except Exception as e:
    print(f"Общая ошибка: {e}")
finally:
    client.close()

Логирование и отладка

Включение логирования

import paramiko
import logging

# Настройка логирования
logging.basicConfig(level=logging.DEBUG)
paramiko.util.log_to_file('paramiko.log')

# Или для более детального контроля
logger = logging.getLogger('paramiko')
logger.setLevel(logging.DEBUG)
handler = logging.FileHandler('debug.log')
handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
logger.addHandler(handler)

Практические примеры использования

Массовое выполнение команд

import paramiko
import threading

def execute_command(host, username, password, command):
    try:
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(host, username=username, password=password)
        
        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read().decode()
        errors = stderr.read().decode()
        
        print(f"Хост {host}:")
        print(f"Вывод: {output}")
        if errors:
            print(f"Ошибки: {errors}")
            
    except Exception as e:
        print(f"Ошибка на хосте {host}: {e}")
    finally:
        client.close()

# Список серверов
servers = ['192.168.1.10', '192.168.1.11', '192.168.1.12']

# Параллельное выполнение
threads = []
for host in servers:
    t = threading.Thread(target=execute_command, args=(host, 'admin', 'password', 'uptime'))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

Синхронизация файлов

import paramiko
import os

def sync_directory(local_dir, remote_dir, hostname, username, password):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname, username=username, password=password)
    
    sftp = ssh.open_sftp()
    
    try:
        # Создание удалённой директории
        sftp.mkdir(remote_dir)
    except IOError:
        pass  # Директория уже существует
    
    # Загрузка файлов
    for root, dirs, files in os.walk(local_dir):
        for file in files:
            local_path = os.path.join(root, file)
            relative_path = os.path.relpath(local_path, local_dir)
            remote_path = os.path.join(remote_dir, relative_path).replace('\\', '/')
            
            # Создание директорий на удалённом сервере
            remote_dir_path = os.path.dirname(remote_path)
            try:
                sftp.mkdir(remote_dir_path)
            except IOError:
                pass
            
            # Загрузка файла
            print(f"Загружаю {local_path} -> {remote_path}")
            sftp.put(local_path, remote_path)
    
    sftp.close()
    ssh.close()

Создание backup-системы

import paramiko
import datetime
import os

def create_backup(servers, backup_paths, local_backup_dir):
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    
    for server in servers:
        try:
            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh.connect(server['host'], username=server['username'], password=server['password'])
            
            sftp = ssh.open_sftp()
            
            server_backup_dir = os.path.join(local_backup_dir, server['host'], timestamp)
            os.makedirs(server_backup_dir, exist_ok=True)
            
            for remote_path in backup_paths:
                filename = os.path.basename(remote_path)
                local_path = os.path.join(server_backup_dir, filename)
                
                print(f"Создаю backup: {server['host']}:{remote_path} -> {local_path}")
                sftp.get(remote_path, local_path)
            
            sftp.close()
            ssh.close()
            
        except Exception as e:
            print(f"Ошибка backup для {server['host']}: {e}")

# Пример использования
servers = [
    {'host': '192.168.1.10', 'username': 'admin', 'password': 'password'},
    {'host': '192.168.1.11', 'username': 'admin', 'password': 'password'},
]

backup_paths = ['/etc/nginx/nginx.conf', '/etc/hosts', '/etc/passwd']
create_backup(servers, backup_paths, '/local/backup/dir')

Работа с jump-серверами (Bastion hosts)

Подключение через промежуточный сервер

import paramiko

def connect_through_bastion(bastion_host, bastion_user, bastion_pass, 
                           target_host, target_user, target_pass):
    # Подключение к bastion-серверу
    bastion = paramiko.SSHClient()
    bastion.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    bastion.connect(bastion_host, username=bastion_user, password=bastion_pass)
    
    # Создание туннеля
    transport = bastion.get_transport()
    dest_addr = (target_host, 22)
    local_addr = ('127.0.0.1', 0)
    channel = transport.open_channel('direct-tcpip', dest_addr, local_addr)
    
    # Подключение к целевому серверу через туннель
    target = paramiko.SSHClient()
    target.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    target.connect(target_host, username=target_user, password=target_pass, sock=channel)
    
    return bastion, target

# Использование
bastion, target = connect_through_bastion(
    'bastion.example.com', 'bastion_user', 'bastion_pass',
    'internal.server.com', 'target_user', 'target_pass'
)

# Выполнение команды на целевом сервере
stdin, stdout, stderr = target.exec_command('hostname')
print(stdout.read().decode())

# Закрытие соединений
target.close()
bastion.close()

Создание SSH-сервера

Базовый SSH-сервер

import paramiko
import socket
import threading

class SSHServer(paramiko.ServerInterface):
    def __init__(self):
        self.event = threading.Event()
    
    def check_channel_request(self, kind, chanid):
        if kind == 'session':
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
    
    def check_auth_password(self, username, password):
        if username == 'admin' and password == 'password':
            return paramiko.AUTH_SUCCESSFUL
        return paramiko.AUTH_FAILED
    
    def get_allowed_auths(self, username):
        return 'password'
    
    def check_channel_shell_request(self, channel):
        self.event.set()
        return True

def start_ssh_server():
    # Генерация ключа сервера
    host_key = paramiko.RSAKey.generate(2048)
    
    # Создание сокета
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(('localhost', 2222))
    sock.listen(1)
    
    print('SSH сервер запущен на порту 2222')
    
    while True:
        client, addr = sock.accept()
        print(f'Подключение от {addr}')
        
        transport = paramiko.Transport(client)
        transport.add_server_key(host_key)
        
        server = SSHServer()
        transport.start_server(server=server)
        
        # Ожидание shell-запроса
        channel = transport.accept(20)
        if channel is not None:
            server.event.wait(10)
            if server.event.is_set():
                channel.send('Добро пожаловать на SSH сервер!\r\n')
                channel.send('$ ')
                
                while True:
                    try:
                        data = channel.recv(1024)
                        if not data:
                            break
                        
                        command = data.decode().strip()
                        if command == 'exit':
                            break
                        
                        # Простая обработка команд
                        if command == 'date':
                            import datetime
                            response = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
                        elif command == 'whoami':
                            response = 'admin'
                        else:
                            response = f'Неизвестная команда: {command}'
                        
                        channel.send(f'{response}\r\n$ ')
                        
                    except Exception as e:
                        print(f'Ошибка: {e}')
                        break
                
                channel.close()
        
        transport.close()

if __name__ == '__main__':
    start_ssh_server()

Полная таблица методов и функций

Основные функции модуля paramiko

Методы для работы с каналами

Расширенные возможности

Работа с конфигурационными файлами SSH

import paramiko

# Загрузка конфигурации SSH
config = paramiko.SSHConfig()
config.parse(open(os.path.expanduser('~/.ssh/config')))

# Получение настроек для хоста
host_config = config.lookup('myserver')

# Подключение с использованием настроек
client = paramiko.SSHClient()
client.connect(
    hostname=host_config.get('hostname'),
    username=host_config.get('user'),
    port=int(host_config.get('port', 22)),
    key_filename=host_config.get('identityfile')
)

Использование SOCKS прокси

import paramiko
import socks
import socket

# Настройка SOCKS прокси
socks.set_default_proxy(socks.SOCKS5, "127.0.0.1", 1080)
socket.socket = socks.socksocket

# Подключение через прокси
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('target-server.com', username='user', password='password')

Мониторинг прогресса передачи файлов

def progress_callback(transferred, total):
    percentage = (transferred / total) * 100
    print(f'Передано: {transferred}/{total} байт ({percentage:.1f}%)')

# Использование callback при передаче файлов
sftp.put('large_file.bin', '/remote/path/large_file.bin', callback=progress_callback)

Сравнение с альтернативами

Оптимизация производительности

Переиспользование соединений

class SSHConnectionPool:
    def __init__(self):
        self.connections = {}
    
    def get_connection(self, host, username, password):
        key = f"{host}:{username}"
        if key not in self.connections:
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            client.connect(host, username=username, password=password)
            self.connections[key] = client
        return self.connections[key]
    
    def close_all(self):
        for client in self.connections.values():
            client.close()
        self.connections.clear()

Параллельная обработка

import concurrent.futures
import paramiko

def execute_on_server(server_info):
    host, username, password, command = server_info
    
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(host, username=username, password=password)
    
    stdin, stdout, stderr = client.exec_command(command)
    result = stdout.read().decode()
    client.close()
    
    return host, result

# Параллельное выполнение на множестве серверов
servers = [
    ('server1.com', 'user', 'pass', 'uptime'),
    ('server2.com', 'user', 'pass', 'uptime'),
    ('server3.com', 'user', 'pass', 'uptime'),
]

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(execute_on_server, servers))
    
for host, result in results:
    print(f"{host}: {result}")

Часто задаваемые вопросы

Как подключиться к серверу по ключу без пароля?

key = paramiko.RSAKey.from_private_key_file('/path/to/private_key')
client.connect(hostname, username=username, pkey=key)

Как выполнить команду с sudo?

stdin, stdout, stderr = client.exec_command('sudo command', get_pty=True)
stdin.write('password\n')
stdin.flush()

Как передать большой файл с отображением прогресса?

def progress(transferred, total):
    print(f'Передано: {transferred}/{total} байт')

sftp.put('large_file.bin', '/remote/path/file.bin', callback=progress)

Как обработать timeout при подключении?

try:
    client.connect(hostname, timeout=10)
except socket.timeout:
    print("Превышено время ожидания подключения")

Как проверить существование файла на удалённом сервере?

try:
    sftp.stat('/path/to/file')
    print("Файл существует")
except IOError:
    print("Файл не найден")

Как создать SSH-туннель?

import select
import socketserver

class ForwardServer(socketserver.ThreadingTCPServer):
    daemon_threads = True
    allow_reuse_address = True

# Создание туннеля для перенаправления локального порта
def forward_tunnel(local_port, remote_host, remote_port, transport):
    class SubHandler(socketserver.BaseRequestHandler):
        def handle(self):
            try:
                chan = transport.open_channel('direct-tcpip',
                                            (remote_host, remote_port),
                                            self.request.getpeername())
            except Exception as e:
                return
            
            while True:
                r, w, x = select.select([self.request, chan], [], [])
                if self.request in r:
                    data = self.request.recv(1024)
                    if len(data) == 0:
                        break
                    chan.send(data)
                if chan in r:
                    data = chan.recv(1024)
                    if len(data) == 0:
                        break
                    self.request.send(data)
            
            chan.close()
            self.request.close()
    
    server = ForwardServer(('', local_port), SubHandler)
    server.serve_forever()

Лучшие практики безопасности

Использование контекстных менеджеров

from contextlib import contextmanager

@contextmanager
def ssh_connection(hostname, username, password):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(hostname, username=username, password=password)
        yield client
    finally:
        client.close()

# Использование
with ssh_connection('server.com', 'user', 'pass') as ssh:
    stdin, stdout, stderr = ssh.exec_command('ls -la')
    print(stdout.read().decode())

Валидация host keys

import binascii

def validate_host_key(hostname, key):
    # Загрузка известных ключей
    known_hosts = paramiko.util.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
    
    if hostname in known_hosts:
        host_keys = known_hosts[hostname]
        if key.get_name() in host_keys:
            expected_key = host_keys[key.get_name()]
            if key.asbytes() == expected_key.asbytes():
                return True
    
    return False

class ValidatingPolicy(paramiko.MissingHostKeyPolicy):
    def missing_host_key(self, client, hostname, key):
        if validate_host_key(hostname, key):
            return
        raise paramiko.SSHException(f'Неизвестный host key для {hostname}')

Безопасное хранение учетных данных

import keyring
import getpass

def get_credentials(hostname):
    username = input('Username: ')
    password = keyring.get_password('ssh', f'{username}@{hostname}')
    
    if not password:
        password = getpass.getpass('Password: ')
        keyring.set_password('ssh', f'{username}@{hostname}', password)
    
    return username, password

username, password = get_credentials('server.com')

Заключение

Paramiko является мощным и гибким инструментом для работы с SSH в Python. Библиотека предоставляет полный контроль над SSH-соединениями, начиная от простых команд и заканчивая сложными сценариями автоматизации. Благодаря своей надёжности, безопасности и богатому функционалу, Paramiko остаётся стандартом де-факто для SSH-взаимодействия в Python-проектах.

Правильное использование Paramiko с соблюдением принципов безопасности позволяет создавать надёжные системы автоматизации, мониторинга и управления инфраструктурой. Библиотека активно развивается и поддерживается сообществом, что гарантирует её актуальность и совместимость с современными требованиями.