Python для SSH-администрирования: автоматизация управления серверами

pip install paramiko

import paramiko

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

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

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

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

# Получение результата
output = stdout.read().decode()
print(output)

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

import paramiko

key = paramiko.RSAKey.from_private_key_file('/path/to/private_key')

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

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

# Остальной код аналогичен предыдущему примеру

# Создание интерактивной сессии
transport = client.get_transport()
channel = transport.open_session()
channel.get_pty()
channel.invoke_shell()

# Отправка команд в интерактивную сессию
channel.send('sudo -s\n')
time.sleep(1)
channel.send('password\n')
time.sleep(1)

# Получение вывода
output = channel.recv(4096).decode()
print(output)

try:
client.connect(hostname='server.example.com', username='username', password='password')
# Код выполнения команд
except paramiko.AuthenticationException:
print("Ошибка аутентификации")
except paramiko.SSHException as e:
print(f"Ошибка SSH: {e}")
except Exception as e:
print(f"Произошла ошибка: {e}")
finally:
client.close()

# Настройка проброса портов (localhost:8080 -> remote_server:80)
transport = client.get_transport()
transport.request_port_forward('', 8080, 'remote_server', 80)

# Теперь можно подключаться к localhost:8080, и трафик будет 
# перенаправлен на remote_server:80 через SSH-туннель

pip install fabric invoke

from fabric import Connection

# Создание соединения
conn = Connection('username@server.example.com')

# Выполнение команды
result = conn.run('hostname')
print(f"Hostname: {result.stdout.strip()}")

# Выполнение команды с sudo
result = conn.sudo('apt update', hide=True)
if result.ok:
print("Система успешно обновлена")

from fabric import SerialGroup, ThreadingGroup

# Последовательное выполнение на нескольких серверах
servers = SerialGroup('web1', 'web2', 'web3')
results = servers.run('uptime')

# Параллельное выполнение для ускорения
parallel_servers = ThreadingGroup('web1', 'web2', 'web3')
parallel_results = parallel_servers.put('local_file.txt', '/tmp/remote_file.txt')

from invoke import run

# Локальное выполнение команды
result = run('ls -la', hide=True)
files = result.stdout.split('\n')

from fabric import task

@task
def deploy(c):
"""Развертывание приложения на сервере"""
c.run('git pull')
c.run('pip install -r requirements.txt')
c.run('systemctl restart myapp')

@task
def backup(c):
"""Создание резервной копии базы данных"""
c.run('pg_dump -U user db > /tmp/backup.sql')
c.get('/tmp/backup.sql', 'backups/')

# Использование: fab -H server1,server2 deploy

@task
def deploy_full(c, branch="main"):
"""Полный цикл развертывания с локальной сборкой и удаленной установкой"""
# Локальная подготовка (используя Invoke)
c.local(f"git checkout {branch}")
c.local("npm run build")

# Удаленное развертывание (используя Fabric)
c.put("dist/", "/var/www/app/")
c.run("systemctl restart nginx")

# Проверка развертывания
result = c.run("curl -s http://localhost | grep 'Welcome'")
if not result.ok:
print("Deployment verification failed!")
c.run("systemctl restart nginx") # Попытка исправления

import paramiko

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

# Получение SFTP-клиента
sftp = ssh.open_sftp()

# Загрузка файла на сервер
sftp.put('local_file.txt', '/remote/path/remote_file.txt')

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

# Список файлов в директории
file_list = sftp.listdir('/remote/path/')
for file in file_list:
print(file)

# Создание директории
sftp.mkdir('/remote/path/new_directory')

# Изменение прав доступа
sftp.chmod('/remote/path/file.txt', 0o644) # Права в восьмеричной форме

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

import pysftp

# Настройка соединения без проверки хостового ключа (не рекомендуется для production)
cnopts = pysftp.CnOpts()
cnopts.hostkeys = None # Для production следует использовать хостовые ключи

with pysftp.Connection('server.example.com', username='user', password='password', cnopts=cnopts) as sftp:
# Загрузка файла на сервер
sftp.put('local_file.txt') # Загрузка в текущую директорию

# Скачивание файла с сервера
sftp.get('remote_file.txt')

# Переход в другую директорию
sftp.chdir('/var/www')

# Рекурсивная загрузка директории
sftp.put_r('local_directory', 'remote_directory')

# Рекурсивное скачивание директории
sftp.get_r('remote_directory', 'local_directory')

# Проверка существования файла
if sftp.exists('file.txt'):
print("Файл существует!")

try:
with pysftp.Connection('server.example.com', username='user', password='password') as sftp:
# Проверка доступного места перед загрузкой
stat = sftp.execute('df -h /')
print(f"Доступное место на сервере:\n{stat}")

# Загрузка с индикатором прогресса
sftp.put('large_file.zip', callback=lambda x, y: print(f"Загружено: {x} из {y} байт"))
except pysftp.ConnectionException:
print("Ошибка подключения к серверу")
except pysftp.CredentialException:
print("Ошибка аутентификации")
except pysftp.SSHException as e:
print(f"SSH ошибка: {e}")
except IOError as e:
print(f"Ошибка ввода-вывода: {e}")
except Exception as e:
print(f"Непредвиденная ошибка: {e}")

import pysftp

# Создание объекта опций с загрузкой хостовых ключей
cnopts = pysftp.CnOpts()

# Для первого подключения можно добавить хостовый ключ
# cnopts.hostkeys.add('server.example.com', 'ssh-rsa', 'KEY_DATA_HERE')

# Подключение с использованием приватного ключа
with pysftp.Connection('server.example.com', 
username='user', 
private_key='/path/to/private_key',
private_key_pass='key_passphrase', # если ключ защищен паролем
cnopts=cnopts) as sftp:
# Операции с файлами
pass

import pysftp
import os
from datetime import datetime

def sync_directory(local_path, remote_path, server, username, password):
"""Синхронизирует локальную директорию с удаленной"""
with pysftp.Connection(server, username=username, password=password) as sftp:
# Убедимся, что удаленная директория существует
if not sftp.exists(remote_path):
sftp.mkdir(remote_path)

sftp.chdir(remote_path)

# Получим список файлов на сервере
remote_files = {f: sftp.stat(f) for f in sftp.listdir()}

# Сравним с локальными файлами
for local_file in os.listdir(local_path):
local_file_path = os.path.join(local_path, local_file)

# Пропускаем директории для простоты
if os.path.isdir(local_file_path):
continue

local_mtime = datetime.fromtimestamp(os.path.getmtime(local_file_path))

# Если файл новее или не существует на сервере – загружаем
if (local_file not in remote_files or 
local_mtime > datetime.fromtimestamp(remote_files[local_file].st_mtime)):
print(f"Uploading: {local_file}")
sftp.put(local_file_path)

# Использование
sync_directory('./local_data', '/var/data', 'server.example.com', 'user', 'password')

from fabric import ThreadingGroup
import logging

# Настройка логирования
logging.basicConfig(level=logging.INFO, 
format='%(asctime)s – %(levelname)s – %(message)s',
filename='update.log')

# Список серверов по группам
server_groups = {
'web': ['web1.example.com', 'web2.example.com', 'web3.example.com'],
'db': ['db1.example.com', 'db2.example.com'],
'cache': ['redis1.example.com', 'redis2.example.com']
}

def update_servers(group_name, hosts):
"""Обновляет группу серверов с минимальным простоем"""
logging.info(f"Начинаем обновление группы {group_name}")

# Создаем группу для параллельного выполнения
servers = ThreadingGroup(*hosts, user='admin')

try:
# Обновление пакетов
logging.info(f"Обновление пакетов на {group_name}")
results = servers.run('sudo apt update && sudo apt upgrade -y')

# Проверка наличия необходимости перезагрузки
logging.info(f"Проверка необходимости перезагрузки на {group_name}")
reboot_check = servers.run('if [ -f /var/run/reboot-required ]; then echo "REBOOT"; else echo "NO_REBOOT"; fi')

# Перезагрузка серверов, которым это необходимо
for conn, result in reboot_check.items():
if "REBOOT" in result.stdout:
logging.info(f"Перезагрузка {conn.host}")
# Перезагрузка асинхронно, чтобы не ждать её завершения
conn.run('sudo shutdown -r +1 "Перезагрузка после обновления" &', warn=True)

except Exception as e:
logging.error(f"Ошибка при обновлении группы {group_name}: {e}")
return False

logging.info(f"Обновление группы {group_name} завершено")
return True

# Последовательное обновление по группам
for group, hosts in server_groups.items():
update_servers(group, hosts)

import paramiko
import time
import json
from datetime import datetime

def check_server(hostname, username, key_path):
"""Собирает основные метрики с сервера"""
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

metrics = {
'hostname': hostname,
'timestamp': datetime.now().isoformat(),
'status': 'offline',
'metrics': {}
}

try:
client.connect(hostname, username=username, key_filename=key_path, timeout=10)
metrics['status'] = 'online'

# Загрузка CPU
stdin, stdout, stderr = client.exec_command("top -bn1 | grep 'Cpu(s)' | awk '{print $2 + $4}'")
metrics['metrics']['cpu_usage'] = float(stdout.read().decode().strip())

# Использование памяти
stdin, stdout, stderr = client.exec_command("free | grep Mem | awk '{print $3/$2 * 100.0}'")
metrics['metrics']['memory_usage'] = float(stdout.read().decode().strip())

# Использование диска
stdin, stdout, stderr = client.exec_command("df / | tail -1 | awk '{print $5}' | sed 's/%//'")
metrics['metrics']['disk_usage'] = float(stdout.read().decode().strip())

# Аптайм
stdin, stdout, stderr = client.exec_command("uptime -p")
metrics['metrics']['uptime'] = stdout.read().decode().strip()

except Exception as e:
metrics['error'] = str(e)
finally:
client.close()

return metrics

# Пример использования для группы серверов
servers = [
{'hostname': 'web1.example.com', 'username': 'admin', 'key_path': '/path/to/key'},
{'hostname': 'web2.example.com', 'username': 'admin', 'key_path': '/path/to/key'},
]

while True:
results = []
for server in servers:
metrics = check_server(**server)
results.append(metrics)

# Вывод предупреждений при необходимости
if metrics['status'] == 'online':
if metrics['metrics'].get('cpu_usage', 0) > 80:
print(f"WARNING: High CPU usage on {server['hostname']}: {metrics['metrics']['cpu_usage']}%")
if metrics['metrics'].get('memory_usage', 0) > 90:
print(f"WARNING: High memory usage on {server['hostname']}: {metrics['metrics']['memory_usage']}%")
else:
print(f"ERROR: Server {server['hostname']} is offline!")

# Сохранение результатов в JSON
with open(f"metrics_{datetime.now().strftime('%Y%m%d_%H%M')}.json", 'w') as f:
json.dump(results, f, indent=2)

# Пауза перед следующей проверкой
time.sleep(300) # Проверка каждые 5 минут

from fabric import Connection
import time
from datetime import datetime

def deploy_application(host, repo_url, app_dir, env="production"):
"""Развертывание веб-приложения на сервере"""
conn = Connection(host)

# Создаем каталог для релизов если он не существует
releases_dir = f"{app_dir}/releases"
conn.run(f"mkdir -p {releases_dir}")

# Создаем новый каталог для текущего релиза
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
release_path = f"{releases_dir}/{timestamp}"

print(f"Deploying new release to {host}...")

try:
# Клонирование репозитория
conn.run(f"git clone {repo_url} {release_path}")

# Переключение на нужную ветку в зависимости от окружения
if env == "production":
branch = "main"
else:
branch = "develop"
conn.run(f"cd {release_path} && git checkout {branch}")

# Установка зависимостей
conn.run(f"cd {release_path} && pip install -r requirements.txt")

# Выполнение миграций базы данных (для Django)
conn.run(f"cd {release_path} && python manage.py migrate")

# Сборка статических файлов
conn.run(f"cd {release_path} && python manage.py collectstatic --noinput")

# Обновление символической ссылки на текущий релиз
conn.run(f"ln -sfn {release_path} {app_dir}/current")

# Перезапуск сервисов
conn.run(f"sudo systemctl restart gunicorn")
conn.run(f"sudo systemctl restart nginx")

# Проверка работоспособности
time.sleep(5) # Даем время на запуск
response = conn.run("curl -s -o /dev/null -w '%{http_code}' http://localhost")
if response.stdout.strip() == "200":
print("Deployment successful! Application is responding.")
else:
print(f"Warning: Application returned status code {response.stdout.strip()}")

# Очистка старых релизов (оставляем 5 последних)
conn.run(f"ls -dt {releases_dir}/* | tail -n +6 | xargs rm -rf")

except Exception as e:
print(f"Deployment failed: {e}")
return False

return True

# Пример использования
deploy_application(
host="web1.example.com",
repo_url="git@github.com:company/project.git",
app_dir="/var/www/myapp",
env="production"
)

import paramiko
import os
from datetime import datetime

def backup_database(host, db_user, db_name, local_backup_dir, remote_user, key_path):
"""Создает резервную копию базы данных и загружает её локально"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M")
remote_filename = f"/tmp/{db_name}_{timestamp}.sql.gz"
local_filename = os.path.join(local_backup_dir, f"{db_name}_{timestamp}.sql.gz")

# Создаем локальную директорию, если она не существует
os.makedirs(local_backup_dir, exist_ok=True)

# Устанавливаем SSH соединение
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
# Подключаемся к серверу
client.connect(host, username=remote_user, key_filename=key_path)

print(f"Creating backup of {db_name} on {host}...")

# Создаем резервную копию и сразу сжимаем её
command = f"pg_dump -U {db_user} {db_name} | gzip > {remote_filename}"
stdin, stdout, stderr = client.exec_command(command)

# Проверяем успешность выполнения
exit_status = stdout.channel.recv_exit_status()
if exit_status != 0:
error = stderr.read().decode()
raise Exception(f"Database dump failed: {error}")

print(f"Backup created, downloading to local storage...")

# Загружаем файл
sftp = client.open_sftp()
sftp.get(remote_filename, local_filename)

# Удаляем временный файл на сервере
client.exec_command(f"rm {remote_filename}")

print(f"Backup completed and saved to {local_filename}")

# Получаем размер файла
file_size = os.path.getsize(local_filename)
print(f"Backup size: {file_size / 1024 / 1024:.2f} MB")

except Exception as e:
print(f"Backup failed: {e}")
return None
finally:
client.close()

return local_filename

# Пример использования
backup = backup_database(
host="db.example.com",
db_user="postgres",
db_name="production_db",
local_backup_dir="./backups",
remote_user="admin",
key_path="/path/to/private_key"
)