316 lines
11 KiB
Python
316 lines
11 KiB
Python
import asyncio
|
|
from datetime import datetime
|
|
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import Horizontal, Vertical
|
|
from textual.reactive import reactive
|
|
from textual.worker import get_current_worker
|
|
from textual import work
|
|
|
|
from full_updater.backend.cache import ensure_cache_dir, read_cache, get_cache_timestamp
|
|
from full_updater.backend.scanner import (
|
|
Target, ScanResult, get_lxc_list, lxc_is_running,
|
|
ensure_debsecan_installed, scan_apt, scan_cve, write_cache
|
|
)
|
|
from full_updater.backend.executor import UpgradeExecutor
|
|
from full_updater.ui.loader import LoaderScreen
|
|
from full_updater.ui.sidebar import Sidebar
|
|
from full_updater.ui.summary import SummaryPanel
|
|
from full_updater.ui.package_table import PackageTable
|
|
from full_updater.ui.cve_table import CVETable
|
|
from full_updater.ui.log_panel import LogPanel
|
|
from full_updater.ui.confirm_modal import ConfirmModal
|
|
|
|
|
|
class FullUpdaterApp(App):
|
|
CSS = """
|
|
Screen { align: center middle; }
|
|
#main-layout { layout: horizontal; height: 100%; }
|
|
Sidebar { width: 30; }
|
|
#content { width: 1fr; }
|
|
"""
|
|
|
|
BINDINGS = [("q", "quit", "Quitter")]
|
|
|
|
targets: list[Target] = []
|
|
results: dict[str, ScanResult] = {}
|
|
selected_target: str | None = None
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.loader_screen = None
|
|
self._log_panel = None
|
|
|
|
def compose(self) -> ComposeResult:
|
|
with Horizontal(id="main-layout"):
|
|
yield Sidebar()
|
|
with Vertical(id="content"):
|
|
yield SummaryPanel()
|
|
|
|
def on_mount(self):
|
|
ensure_cache_dir()
|
|
self.targets = [Target(target_id="host", name="hote", is_host=True)] + get_lxc_list()
|
|
sidebar = self.query_one(Sidebar)
|
|
sidebar.set_targets(self.targets)
|
|
self.loader_screen = LoaderScreen(self.targets)
|
|
self.push_screen(self.loader_screen)
|
|
self.run_scan_all()
|
|
|
|
@work(thread=True)
|
|
def run_scan_all(self):
|
|
total = len(self.targets)
|
|
completed = 0
|
|
|
|
def progress_cb():
|
|
nonlocal completed
|
|
completed += 1
|
|
pct = min(100.0, (completed / (total * 4)) * 100)
|
|
self.call_from_thread(self._update_loader, pct)
|
|
|
|
for idx, target in enumerate(self.targets):
|
|
self.call_from_thread(self._update_loader_status, idx, "running")
|
|
|
|
if not target.is_host and not lxc_is_running(target.target_id):
|
|
self.call_from_thread(self._update_loader_status, idx, "skipped")
|
|
result = ScanResult(target=target)
|
|
result.status = "skipped"
|
|
self.results[target.target_id] = result
|
|
for _ in range(4):
|
|
progress_cb()
|
|
continue
|
|
|
|
debsecan_ok, debsecan_err = ensure_debsecan_installed(target.is_host, target.target_id)
|
|
progress_cb()
|
|
|
|
apt_ok, apt_packages, apt_err = scan_apt(target)
|
|
progress_cb()
|
|
|
|
cve_ok = False
|
|
cve_list = []
|
|
cve_err = ""
|
|
if debsecan_ok:
|
|
cve_ok, cve_list, cve_err = scan_cve(target)
|
|
else:
|
|
cve_err = f"debsecan: {debsecan_err}"
|
|
progress_cb()
|
|
|
|
result = ScanResult(
|
|
target=target,
|
|
apt_ok=apt_ok,
|
|
cve_ok=cve_ok,
|
|
apt_count=len(apt_packages),
|
|
cve_count=len(cve_list),
|
|
apt_packages=apt_packages,
|
|
cve_list=cve_list,
|
|
error=apt_err or cve_err,
|
|
status="done" if (apt_ok and (cve_ok or not debsecan_ok)) else "error"
|
|
)
|
|
if result.error:
|
|
result.status = "error"
|
|
self.results[target.target_id] = result
|
|
|
|
cache_id = "host" if target.is_host else target.target_id
|
|
write_cache(cache_id, {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"apt_count": result.apt_count,
|
|
"apt_packages": result.apt_packages,
|
|
"cve_count": result.cve_count,
|
|
"cve_list": result.cve_list,
|
|
"error": result.error
|
|
})
|
|
|
|
self.call_from_thread(self._update_loader_status, idx, result.status)
|
|
self.call_from_thread(self._update_sidebar, target.target_id, result)
|
|
|
|
self.call_from_thread(self._finish_scan)
|
|
|
|
def _update_loader(self, pct: float):
|
|
if self.loader_screen:
|
|
self.loader_screen.progress = pct
|
|
|
|
def _update_loader_status(self, idx: int, status: str):
|
|
if self.loader_screen:
|
|
self.loader_screen.update_target_status(idx, status)
|
|
|
|
def _update_sidebar(self, target_id: str, result: ScanResult):
|
|
sidebar = self.query_one(Sidebar)
|
|
sidebar.update_status(
|
|
target_id,
|
|
result.apt_ok,
|
|
result.cve_ok,
|
|
result.apt_count,
|
|
result.cve_count,
|
|
result.error,
|
|
result.status == "skipped"
|
|
)
|
|
|
|
def _finish_scan(self):
|
|
self.pop_screen()
|
|
sidebar = self.query_one(Sidebar)
|
|
for tid, res in self.results.items():
|
|
sidebar.update_status(
|
|
tid,
|
|
res.apt_ok,
|
|
res.cve_ok,
|
|
res.apt_count,
|
|
res.cve_count,
|
|
res.error,
|
|
res.status == "skipped"
|
|
)
|
|
if self.targets:
|
|
self._select_target(self.targets[0].target_id)
|
|
|
|
def _select_target(self, target_id: str):
|
|
self.selected_target = target_id
|
|
cache_id = "host" if target_id == "host" else target_id
|
|
data = read_cache(cache_id) or {}
|
|
name = "hote"
|
|
for t in self.targets:
|
|
if t.target_id == target_id:
|
|
name = t.name
|
|
break
|
|
summary = self.query_one(SummaryPanel)
|
|
summary.set_target(
|
|
target_id=target_id,
|
|
name=name,
|
|
apt_count=data.get("apt_count", 0),
|
|
cve_count=data.get("cve_count", 0),
|
|
error=data.get("error", ""),
|
|
skipped=not data and any(t.target_id == target_id and not t.is_host and not lxc_is_running(t.target_id) for t in self.targets),
|
|
cache_time=get_cache_timestamp(cache_id)
|
|
)
|
|
|
|
def on_sidebar_target_selected(self, event: Sidebar.TargetSelected):
|
|
self._select_target(event.target_id)
|
|
|
|
def on_summary_panel_reload_pressed(self, event: SummaryPanel.ReloadPressed):
|
|
if self.selected_target:
|
|
self._reload_target(self.selected_target)
|
|
|
|
@work(thread=True)
|
|
def _reload_target(self, target_id: str):
|
|
target = None
|
|
for t in self.targets:
|
|
if t.target_id == target_id:
|
|
target = t
|
|
break
|
|
if not target:
|
|
return
|
|
|
|
if not target.is_host and not lxc_is_running(target.target_id):
|
|
result = ScanResult(target=target)
|
|
result.status = "skipped"
|
|
self.results[target.target_id] = result
|
|
write_cache(target.target_id, {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"apt_count": 0,
|
|
"apt_packages": [],
|
|
"cve_count": 0,
|
|
"cve_list": [],
|
|
"error": "LXC éteint"
|
|
})
|
|
self.call_from_thread(self._update_sidebar, target.target_id, result)
|
|
self.call_from_thread(self._select_target, target.target_id)
|
|
return
|
|
|
|
ensure_debsecan_installed(target.is_host, target.target_id)
|
|
apt_ok, apt_packages, apt_err = scan_apt(target)
|
|
cve_ok, cve_list, cve_err = scan_cve(target)
|
|
error = apt_err or cve_err
|
|
|
|
result = ScanResult(
|
|
target=target,
|
|
apt_ok=apt_ok,
|
|
cve_ok=cve_ok,
|
|
apt_count=len(apt_packages),
|
|
cve_count=len(cve_list),
|
|
apt_packages=apt_packages,
|
|
cve_list=cve_list,
|
|
error=error,
|
|
status="done" if (apt_ok and cve_ok) else "error"
|
|
)
|
|
self.results[target.target_id] = result
|
|
cache_id = "host" if target.is_host else target.target_id
|
|
write_cache(cache_id, {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"apt_count": result.apt_count,
|
|
"apt_packages": result.apt_packages,
|
|
"cve_count": result.cve_count,
|
|
"cve_list": result.cve_list,
|
|
"error": result.error
|
|
})
|
|
self.call_from_thread(self._update_sidebar, target.target_id, result)
|
|
self.call_from_thread(self._select_target, target.target_id)
|
|
|
|
def on_summary_panel_upgrade_pressed(self, event: SummaryPanel.UpgradePressed):
|
|
self.push_screen(
|
|
ConfirmModal(
|
|
"Confirmation",
|
|
f"Lancer la mise à jour sur {event.name} ?"
|
|
),
|
|
callback=lambda confirmed: self._on_upgrade_confirmed(confirmed, event.target_id, event.name)
|
|
)
|
|
|
|
def _on_upgrade_confirmed(self, confirmed: bool, target_id: str, name: str):
|
|
if not confirmed:
|
|
return
|
|
self.query_one("#main-layout").display = False
|
|
self._log_panel = LogPanel()
|
|
self.mount(self._log_panel)
|
|
self._log_panel.clear()
|
|
self._log_panel.write(f"Lancement de la mise à jour sur {name}...")
|
|
self.run_upgrade(target_id, target_id == "host")
|
|
|
|
@work
|
|
async def run_upgrade(self, target_id: str, is_host: bool):
|
|
log_panel = self._log_panel
|
|
if not log_panel:
|
|
return
|
|
executor = UpgradeExecutor(
|
|
target_id, is_host,
|
|
on_line=lambda line: self.call_from_thread(log_panel.write, line)
|
|
)
|
|
ok = await executor.run()
|
|
if ok:
|
|
log_panel.write("\n✅ Mise à jour terminée avec succès.")
|
|
else:
|
|
log_panel.write("\n❌ Erreur lors de la mise à jour.")
|
|
self._reload_target(target_id)
|
|
|
|
def on_log_panel_back_pressed(self, event: LogPanel.BackPressed):
|
|
self.query_one("#main-layout").display = True
|
|
if self._log_panel:
|
|
self._log_panel.remove()
|
|
self._log_panel = None
|
|
|
|
def on_summary_panel_apt_clicked(self, event: SummaryPanel.AptClicked):
|
|
cache_id = "host" if event.target_id == "host" else event.target_id
|
|
data = read_cache(cache_id) or {}
|
|
pkgs = data.get("apt_packages", [])
|
|
if not pkgs:
|
|
return
|
|
table = PackageTable()
|
|
self.mount(table)
|
|
self.query_one("#main-layout").display = False
|
|
table.load_data(pkgs)
|
|
|
|
def on_summary_panel_cve_clicked(self, event: SummaryPanel.CveClicked):
|
|
cache_id = "host" if event.target_id == "host" else event.target_id
|
|
data = read_cache(cache_id) or {}
|
|
cves = data.get("cve_list", [])
|
|
if not cves:
|
|
return
|
|
table = CVETable()
|
|
self.mount(table)
|
|
self.query_one("#main-layout").display = False
|
|
table.load_data(cves)
|
|
|
|
def on_package_table_back_pressed(self, event: PackageTable.BackPressed):
|
|
self.query_one("#main-layout").display = True
|
|
for widget in self.query(PackageTable):
|
|
widget.remove()
|
|
|
|
def on_cve_table_back_pressed(self, event: CVETable.BackPressed):
|
|
self.query_one("#main-layout").display = True
|
|
for widget in self.query(CVETable):
|
|
widget.remove()
|