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()
|