"""Вкладка Терминал: 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()