dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
273 lines
12 KiB
Python
273 lines
12 KiB
Python
"""Вкладка Терминал: PTY-сессия + библиотека команд обслуживания."""
|
||
import flet as ft
|
||
|
||
from app.core.terminal import TerminalSession
|
||
from app.db.models import Server
|
||
|
||
# Категории → [(название, команда), ...]
|
||
# {server} — подстановка не нужна, команды выполняются на самом сервере
|
||
COMMANDS = {
|
||
"Пользователи": [
|
||
("Список пользователей", "cat /etc/passwd | awk -F: '$3>=1000{print $1}'"),
|
||
("Кто залогинен", "who -a"),
|
||
("Добавить пользователя", "sudo adduser"),
|
||
("Удалить пользователя", "sudo deluser --remove-home"),
|
||
("Сменить пароль пользователя", "sudo passwd"),
|
||
("Дать sudo", "sudo usermod -aG sudo"),
|
||
("Убрать sudo", "sudo deluser $(whoami) sudo"),
|
||
("Список групп", "getent group | awk -F: '$3>=1000{print $1}'"),
|
||
("Добавить в группу", "sudo usermod -aG"),
|
||
("Сессии пользователя", "last -n 20"),
|
||
],
|
||
"Система": [
|
||
("Аптайм и нагрузка", "uptime"),
|
||
("Инфо о системе", "hostnamectl"),
|
||
("Версия ядра", "uname -a"),
|
||
("Время и часовой пояс", "timedatectl"),
|
||
("Установить часовой пояс", "sudo timedatectl set-timezone Asia/Ulaanbaatar"),
|
||
("Синхронизация времени", "sudo timedatectl set-ntp true"),
|
||
("Обновить locale", "sudo dpkg-reconfigure locales"),
|
||
("Память", "free -h"),
|
||
("Диски", "lsblk"),
|
||
("SMART дисков", "sudo smartctl -a /dev/sda 2>/dev/null || echo 'smartmontools не установлен'"),
|
||
],
|
||
"Процессы": [
|
||
("Топ процессов", "ps aux --sort=-%cpu | head -20"),
|
||
("Дерево процессов", "pstree -p"),
|
||
("Найти процесс", "ps aux | grep"),
|
||
("Убить по имени", "pkill -f"),
|
||
("Открытые файлы процесса", "sudo lsof -p"),
|
||
],
|
||
"Сеть": [
|
||
("IP-адреса", "ip a"),
|
||
("Маршруты", "ip r"),
|
||
("Открытые порты", "sudo ss -tlnp"),
|
||
("Соединения", "sudo ss -tnp"),
|
||
("DNS", "cat /etc/resolv.conf"),
|
||
("Проверить связь", "ping -c 4 8.8.8.8"),
|
||
("Скачать файл", "wget"),
|
||
("Открыть порт в ufw", "sudo ufw allow"),
|
||
],
|
||
"Логи": [
|
||
("Системный журнал", "sudo journalctl -n 100 --no-pager"),
|
||
("Ошибки за час", "sudo journalctl -p err -S '1 hour ago' --no-pager"),
|
||
("Логи сервиса", "sudo journalctl -u"),
|
||
("dmesg", "sudo dmesg | tail -30"),
|
||
("Авторизации", "sudo last -n 20"),
|
||
("Неудачные логины", "sudo lastb -n 20 2>/dev/null || sudo journalctl _COMM=sshd | grep -i fail | tail -20"),
|
||
],
|
||
"Пакеты": [
|
||
("Обновить индекс", "sudo apt update"),
|
||
("Обновить систему", "sudo apt upgrade -y"),
|
||
("Найти пакет", "apt-cache search"),
|
||
("Инфо о пакете", "apt-cache show"),
|
||
("Установленные вручную", "apt-mark showmanual"),
|
||
("Очистка", "sudo apt autoremove -y && sudo apt autoclean"),
|
||
("История apt", "grep -h ' install \\| upgrade ' /var/log/apt/history.log* | tail -20"),
|
||
],
|
||
"Безопасность": [
|
||
("Статус fail2ban", "sudo fail2ban-client status 2>/dev/null || echo 'не установлен'"),
|
||
("Правила ufw", "sudo ufw status numbered"),
|
||
("Открытые сокеты", "sudo lsof -i -P -n | grep LISTEN"),
|
||
("SUID-файлы", "sudo find / -perm -4000 -type f 2>/dev/null | head -20"),
|
||
("Пользователи с shell", "grep -E '/bin/(bash|sh|zsh)' /etc/passwd"),
|
||
],
|
||
}
|
||
|
||
|
||
class TerminalTab(ft.Column):
|
||
def __init__(self, page: ft.Page, srv: Server):
|
||
super().__init__(expand=True, spacing=8)
|
||
self._page = page
|
||
self.srv = srv
|
||
self._term: TerminalSession | None = None
|
||
self._connected = False
|
||
|
||
# вывод терминала
|
||
self.output = ft.Column(
|
||
scroll=ft.ScrollMode.AUTO, spacing=0, expand=True,
|
||
auto_scroll=True,
|
||
)
|
||
self.output_container = ft.Container(
|
||
expand=True, bgcolor=ft.Colors.BLACK, border_radius=6, padding=8,
|
||
content=self.output,
|
||
)
|
||
|
||
# строка ввода
|
||
self.input = ft.TextField(
|
||
hint_text="команда… (Enter — отправить)",
|
||
dense=True, expand=True, text_size=13,
|
||
text_style=ft.TextStyle(font_family="Consolas"),
|
||
on_submit=self._send,
|
||
disabled=True,
|
||
)
|
||
|
||
# статус подключения
|
||
self.status_icon = ft.Icon(ft.Icons.CIRCLE, color=ft.Colors.GREY, size=12)
|
||
self.status_text = ft.Text("не подключено", size=12, color=ft.Colors.GREY_500)
|
||
|
||
# селектор команд
|
||
cat_names = list(COMMANDS.keys())
|
||
self.cat_dd = ft.Dropdown(
|
||
label="Категория", value=cat_names[0], width=180,
|
||
options=[ft.dropdown.Option(c) for c in cat_names],
|
||
on_select=lambda e: self._fill_cmd_dd(),
|
||
)
|
||
self.cmd_dd = ft.Dropdown(
|
||
label="Команда", width=350,
|
||
options=[ft.dropdown.Option(n) for n, _ in COMMANDS[cat_names[0]]],
|
||
on_select=lambda e: self._on_cmd_selected(),
|
||
)
|
||
self.param_tf = ft.TextField(
|
||
hint_text="параметр (если нужен)", width=200, dense=True, visible=False,
|
||
)
|
||
|
||
self.controls = [
|
||
ft.Container(height=8),
|
||
# панель подключения
|
||
ft.Row([
|
||
ft.FilledButton("Подключиться", icon=ft.Icons.TERMINAL,
|
||
on_click=self._connect),
|
||
ft.OutlinedButton("Отключиться", icon=ft.Icons.CLOSE,
|
||
on_click=self._disconnect),
|
||
self.status_icon, self.status_text,
|
||
ft.Container(expand=True),
|
||
ft.TextButton("Очистить вывод", on_click=lambda e: self._clear()),
|
||
]),
|
||
# селектор команд
|
||
ft.Row([
|
||
self.cat_dd, self.cmd_dd, self.param_tf,
|
||
ft.FilledButton("Выполнить", icon=ft.Icons.PLAY_ARROW,
|
||
on_click=self._run_selected),
|
||
]),
|
||
# терминал
|
||
ft.Container(expand=True, content=self.output_container),
|
||
# ввод
|
||
ft.Row([self.input]),
|
||
]
|
||
|
||
# ---------- подключение ----------
|
||
|
||
def _connect(self, e=None):
|
||
if self._connected:
|
||
return
|
||
self._set_status("подключение…", ft.Colors.ORANGE_300)
|
||
self._page.run_task(self._do_connect)
|
||
|
||
async def _do_connect(self):
|
||
try:
|
||
self._term = TerminalSession(self.srv, self._on_output)
|
||
await self._term.open()
|
||
self._connected = True
|
||
self.input.disabled = False
|
||
self._set_status(f"подключено: {self.srv.username}@{self.srv.host}",
|
||
ft.Colors.GREEN_400)
|
||
# приглашение
|
||
self._term.write(f"echo '--- Debian Manager Terminal | {self.srv.name} ---'\n")
|
||
except Exception as ex:
|
||
self._set_status(f"ошибка: {ex}", ft.Colors.RED_400)
|
||
|
||
def _disconnect(self, e=None):
|
||
if self._term:
|
||
self._term.close()
|
||
self._term = None
|
||
self._connected = False
|
||
self.input.disabled = True
|
||
self._set_status("не подключено", ft.Colors.GREY_500)
|
||
|
||
def _set_status(self, text, color):
|
||
self.status_text.value = text
|
||
self.status_icon.color = color
|
||
self._page.update()
|
||
|
||
def _on_output(self, data: str):
|
||
# разбиваем ANSI-строки на Text-строки (упрощённо — без цветов)
|
||
for line in data.splitlines():
|
||
# убираем ANSI escape-коды
|
||
clean = self._strip_ansi(line)
|
||
if clean is not None:
|
||
self.output.controls.append(
|
||
ft.Text(clean, size=12, font_family="Consolas",
|
||
color=ft.Colors.GREEN_100, selectable=True)
|
||
)
|
||
# ограничиваем буфер
|
||
if len(self.output.controls) > 500:
|
||
self.output.controls = self.output.controls[-400:]
|
||
self._page.update()
|
||
|
||
@staticmethod
|
||
def _strip_ansi(s: str) -> str:
|
||
import re
|
||
return re.sub(r"\x1b\[[0-9;]*[a-zA-Z]", "", s)
|
||
|
||
def _clear(self):
|
||
self.output.controls = []
|
||
self._page.update()
|
||
|
||
# ---------- ввод ----------
|
||
|
||
def _send(self, e=None):
|
||
if self._term and self._connected and self.input.value:
|
||
self._term.write(self.input.value + "\n")
|
||
self.input.value = ""
|
||
self._page.update()
|
||
|
||
# ---------- библиотека команд ----------
|
||
|
||
def _fill_cmd_dd(self):
|
||
cat = self.cat_dd.value
|
||
self.cmd_dd.options = [
|
||
ft.dropdown.Option(n) for n, _ in COMMANDS[cat]
|
||
]
|
||
self.cmd_dd.value = None
|
||
self.param_tf.visible = False
|
||
self._page.update()
|
||
|
||
def _on_cmd_selected(self):
|
||
"""Показать поле параметра, если команда его требует."""
|
||
cat = self.cat_dd.value
|
||
name = self.cmd_dd.value
|
||
if not name:
|
||
return
|
||
cmd = next((c for n, c in COMMANDS[cat] if n == name), "")
|
||
# команды, заканчивающиеся на пробел/утилиту без аргумента — нужен параметр
|
||
needs_param = cmd.endswith((" ", "adduser", "passwd", "deluser", "usermod -aG",
|
||
"journalctl -u", "grep", "pkill -f", "lsof -p",
|
||
"allow", "search", "show", "wget"))
|
||
self.param_tf.visible = needs_param
|
||
self._page.update()
|
||
|
||
def _run_selected(self, e=None):
|
||
if not self._connected:
|
||
self._connect()
|
||
# повторим после подключения
|
||
self._page.run_task(self._run_after_connect)
|
||
return
|
||
self._execute_selected()
|
||
|
||
async def _run_after_connect(self):
|
||
import asyncio
|
||
for _ in range(30):
|
||
await asyncio.sleep(0.2)
|
||
if self._connected:
|
||
break
|
||
if self._connected:
|
||
self._execute_selected()
|
||
|
||
def _execute_selected(self):
|
||
cat = self.cat_dd.value
|
||
name = self.cmd_dd.value
|
||
if not name:
|
||
return
|
||
cmd = next((c for n, c in COMMANDS[cat] if n == name), "")
|
||
param = (self.param_tf.value or "").strip()
|
||
full = f"{cmd} {param}".rstrip() if param else cmd
|
||
# подсветим выполняемую команду в выводе
|
||
self.output.controls.append(
|
||
ft.Text(f"$ {full}", size=12, font_family="Consolas",
|
||
color=ft.Colors.YELLOW_300, weight=ft.FontWeight.BOLD)
|
||
)
|
||
self._term.write(full + "\n")
|
||
self.param_tf.value = ""
|
||
self._page.update()
|