tools/server/ssh_client.py - master - guselnikov/inst

tools/server/ssh_client.py
93 строки3 KB
Viktor Guselnikov
first_commit
07 июл 2026, 11:54
07 июл 2026, 11:54098126a
"""SSH-клиент для сервера inst (Mustard Thallium)."""

from __future__ import annotations

import os
import sys
from pathlib import Path

import paramiko
from dotenv import load_dotenv

ENV_PATH = Path(__file__).resolve().parent / '.env'


def load_config() -> dict[str, str]:
    load_dotenv(ENV_PATH)
    cfg = {
        'host': os.getenv('SERVER_HOST', ''),
        'user': os.getenv('SERVER_USER', 'root'),
        'password': os.getenv('SERVER_PASSWORD', ''),
        'name': os.getenv('SERVER_NAME', 'server'),
    }
    if not cfg['host'] or not cfg['password']:
        raise SystemExit(f'Заполни SERVER_HOST и SERVER_PASSWORD в {ENV_PATH}')
    return cfg


def connect() -> paramiko.SSHClient:
    cfg = load_config()
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(
        cfg['host'],
        username=cfg['user'],
        password=cfg['password'],
        timeout=30,
        allow_agent=False,
        look_for_keys=False,
    )
    return client


def run(cmd: str, timeout: int = 300) -> tuple[int, str, str]:
    client = connect()
    try:
        _, stdout, stderr = client.exec_command(cmd, timeout=timeout)
        out = stdout.read().decode('utf-8', errors='replace')
        err = stderr.read().decode('utf-8', errors='replace')
        code = stdout.channel.recv_exit_status()
        return code, out, err
    finally:
        client.close()


def run_interactive_shell(cmd: str) -> int:
    """Выполнить одну команду и вывести stdout/stderr в realtime."""
    code, out, err = run(cmd)
    if out:
        print(out, end='')
    if err:
        print(err, end='', file=sys.stderr)
    return code


def upload_bytes(remote_path: str, data: bytes, mode: int = 0o644) -> None:
    cfg = load_config()
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    client.connect(
        cfg['host'],
        username=cfg['user'],
        password=cfg['password'],
        timeout=30,
        allow_agent=False,
        look_for_keys=False,
    )
    try:
        sftp = client.open_sftp()
        dir_path = os.path.dirname(remote_path)
        run(f'mkdir -p {dir_path}')
        with sftp.file(remote_path, 'w') as f:
            f.write(data)
        sftp.chmod(remote_path, mode)
        sftp.close()
    finally:
        client.close()


if __name__ == '__main__':
    if len(sys.argv) < 2:
        print('Usage: python ssh_client.py "<remote command>"')
        raise SystemExit(1)
    raise SystemExit(run_interactive_shell(' '.join(sys.argv[1:])))