diff --git a/full_updater/app.py b/full_updater/app.py index 969a526..0bee8a2 100644 --- a/full_updater/app.py +++ b/full_updater/app.py @@ -6,7 +6,7 @@ from textual.containers import Horizontal, Vertical from textual.reactive import reactive from textual import work -from full_updater.backend.cache import ensure_cache_dir, read_cache, get_cache_timestamp +from full_updater.backend.cache import clear_cache, 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, scan_target @@ -46,6 +46,7 @@ class FullUpdaterApp(App): yield SummaryPanel() def on_mount(self): + clear_cache() ensure_cache_dir() self.targets = [Target(target_id="host", name="hote", is_host=True)] + get_lxc_list() sidebar = self.query_one(Sidebar) @@ -93,6 +94,7 @@ class FullUpdaterApp(App): result.error, result.status == "skipped" ) + sidebar.refresh() def _finish_scan(self): self.pop_screen() @@ -109,7 +111,7 @@ class FullUpdaterApp(App): ) sidebar.refresh() if self.targets: - self._select_target(self.targets[0].target_id) + self.set_timer(0.2, lambda: self._select_target(self.targets[0].target_id)) def _select_target(self, target_id: str): self.selected_target = target_id @@ -126,6 +128,8 @@ class FullUpdaterApp(App): name=name, apt_count=data.get("apt_count", 0), cve_count=data.get("cve_count", 0), + cve_total=data.get("cve_total", 0), + os_info=data.get("os_info", ""), 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) @@ -156,7 +160,8 @@ class FullUpdaterApp(App): write_cache(target.target_id, { "timestamp": datetime.now().isoformat(), "apt_count": 0, "apt_packages": [], - "cve_count": 0, "cve_list": [], + "cve_count": 0, "cve_total": 0, "cve_list": [], + "os_info": "LXC éteint", "error": "LXC éteint" }) self._update_sidebar(target.target_id, result) @@ -174,6 +179,8 @@ class FullUpdaterApp(App): "apt_count": result.apt_count, "apt_packages": result.apt_packages, "cve_count": result.cve_count, + "cve_total": result.cve_total, + "os_info": result.os_info, "cve_list": result.cve_list, "error": result.error }) @@ -206,7 +213,7 @@ class FullUpdaterApp(App): return executor = UpgradeExecutor( target_id, is_host, - on_line=lambda line: self.call_from_thread(log_panel.write, line) + on_line=lambda line: log_panel.write(line) ) ok = await executor.run() if ok: diff --git a/full_updater/backend/cache.py b/full_updater/backend/cache.py index 1edb94d..5172ca9 100644 --- a/full_updater/backend/cache.py +++ b/full_updater/backend/cache.py @@ -7,9 +7,12 @@ from typing import Any CACHE_DIR = "/tmp/full-updater-cache" -def ensure_cache_dir() -> None: +def clear_cache() -> None: if os.path.exists(CACHE_DIR): shutil.rmtree(CACHE_DIR) + + +def ensure_cache_dir() -> None: os.makedirs(CACHE_DIR, exist_ok=True) diff --git a/full_updater/backend/scanner.py b/full_updater/backend/scanner.py index 4271c74..f00185a 100644 --- a/full_updater/backend/scanner.py +++ b/full_updater/backend/scanner.py @@ -1,10 +1,115 @@ +import json +import os import re import subprocess +import urllib.request +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from datetime import datetime from typing import Callable -from full_updater.backend.cache import write_cache +from full_updater.backend.cache import write_cache, ensure_cache_dir + +CVE_API_CACHE = "/tmp/full-updater-cache/cve-api" + + +def _ensure_cve_api_cache() -> None: + os.makedirs(CVE_API_CACHE, exist_ok=True) + + +def _fetch_cve_html(cve_id: str) -> str: + """Récupère le HTML de la page Debian Security Tracker pour une CVE.""" + _ensure_cve_api_cache() + cache_path = os.path.join(CVE_API_CACHE, f"{cve_id}.html") + + if os.path.exists(cache_path): + try: + with open(cache_path, "r", encoding="utf-8") as f: + return f.read() + except Exception: + pass + + url = f"https://security-tracker.debian.org/tracker/{cve_id}" + try: + req = urllib.request.Request(url, headers={"User-Agent": "full-updater/1.0"}) + with urllib.request.urlopen(req, timeout=15) as resp: + html = resp.read().decode("utf-8") + with open(cache_path, "w", encoding="utf-8") as f: + f.write(html) + return html + except Exception: + return "" + + +def _enrich_cve(cve_id: str) -> dict[str, str]: + """Récupère la sévérité et le vecteur d'attaque depuis le HTML Debian Security Tracker.""" + html = _fetch_cve_html(cve_id) + if not html: + return {"severity": "?", "vector": "?"} + + result = {"severity": "?", "vector": "?"} + + # Extraction de la sévérité depuis le bloc "Vulnerable and fixed packages" + # Recherche du motif: remote ou local + vector_match = re.search(r']*>\s*(remote|local)\s*', html, re.IGNORECASE) + if vector_match: + result["vector"] = vector_match.group(1).lower() + + # Extraction de la sévérité depuis les notes ou le tableau + # Format courant: [low], [medium], [high], [critical] dans les notes + severity_match = re.search(r'\b(unimportant|low|medium|high|critical)\b', html, re.IGNORECASE) + if severity_match: + result["severity"] = severity_match.group(1).lower() + + return result + + +def _is_cve_actionable(cve_id: str, suite: str = "bookworm") -> bool: + """Retourne True si la CVE est marquée 'fixed' pour le suite donné dans le HTML.""" + html = _fetch_cve_html(cve_id) + if not html: + return False + + # Chercher dans le tableau des packages vulnérables/fixed + # Pattern: suite (security)? version fixed + # Ex: bookworm 3.0.18-1~deb12u1 fixed + pattern = re.compile(rf"]*>\s*{re.escape(suite)}\s*(?:\(security\))?\s*\s*]*>[^<]+\s*]*>\s*fixed\s*", re.IGNORECASE) + return bool(pattern.search(html)) + + +def enrich_cves(cves: list[dict]) -> tuple[list[dict], int]: + """Enrichit les CVEs avec fixable, severity, vector via l'API Debian. + Retourne (cve_enrichies, nombre_actionnables).""" + if not cves: + return [], 0 + + def check(cve: dict) -> dict: + try: + # Vérifie si corrigeable + cve["fixable"] = _is_cve_actionable(cve["id"]) + # Enrichit avec severity et vector + enriched = _enrich_cve(cve["id"]) + cve["severity"] = enriched["severity"] + cve["vector"] = enriched["vector"] + return cve + except Exception: + cve["fixable"] = False + return cve + + all_cves = [] + actionable_count = 0 + with ThreadPoolExecutor(max_workers=10) as executor: + futures = [executor.submit(check, cve) for cve in cves] + for future in futures: + try: + cve = future.result() + all_cves.append(cve) + if cve.get("fixable"): + actionable_count += 1 + except Exception: + pass + + return all_cves, actionable_count @dataclass @@ -21,6 +126,8 @@ class ScanResult: cve_ok: bool = False apt_count: int = 0 cve_count: int = 0 + cve_total: int = 0 + os_info: str = "" apt_packages: list[dict[str, str]] = field(default_factory=list) cve_list: list[dict[str, str]] = field(default_factory=list) error: str = "" @@ -56,6 +163,41 @@ def lxc_is_running(vmid: str) -> bool: return ok and "running" in stdout.lower() +def scan_os(target: Target) -> str: + """Récupère le nom, la version complète et le codename de l'OS.""" + prefix = [] if target.is_host else ["pct", "exec", target.target_id, "--"] + + # 1. Lire /etc/os-release et parser ligne par ligne + ok, osrel_out, _ = run_cmd(prefix + ["cat", "/etc/os-release"], timeout=30) + os_data = {} + if ok: + for line in osrel_out.splitlines(): + line = line.strip() + if "=" in line: + key, val = line.split("=", 1) + val = val.strip().strip('"').strip("'") + os_data[key] = val + + # 2. Extraire les champs + name = os_data.get("NAME", "") + codename = os_data.get("VERSION_CODENAME", "") + debian_version_full = os_data.get("DEBIAN_VERSION_FULL", "") + version_id = os_data.get("VERSION_ID", "") + + # 3. Construire le libellé final + # Avec DEBIAN_VERSION_FULL : NAME DEBIAN_VERSION_FULL (VERSION_CODENAME) + # Sans : NAME VERSION_ID (VERSION_CODENAME) + version = debian_version_full if debian_version_full else version_id + + if name and version and codename: + return f"{name} {version} ({codename})" + if name and version: + return f"{name} {version}" + + # Fallback : PRETTY_NAME ou "OS inconnu" + return os_data.get("PRETTY_NAME", "OS inconnu") + + def ensure_debsecan_installed(is_host: bool, vmid: str = "") -> tuple[bool, str]: if is_host: ok, _, _ = run_cmd(["which", "debsecan"]) @@ -104,7 +246,8 @@ def scan_apt(target: Target) -> tuple[bool, list[dict[str, str]], str]: def scan_cve(target: Target) -> tuple[bool, list[dict[str, str]], str]: - cmd = ["debsecan", "--format", "report", "--suite", "bookworm"] if target.is_host else ["pct", "exec", target.target_id, "--", "debsecan", "--format", "report", "--suite", "bookworm"] + # Format par défaut (pas de --format) retourne: CVE-ID package + cmd = ["debsecan", "--suite", "bookworm"] if target.is_host else ["pct", "exec", target.target_id, "--", "debsecan", "--suite", "bookworm"] ok, stdout, stderr = run_cmd(cmd, timeout=120) if not ok: return False, [], stderr or stdout @@ -113,7 +256,13 @@ def scan_cve(target: Target) -> tuple[bool, list[dict[str, str]], str]: for line in stdout.splitlines(): m = re.match(r"(CVE-\d{4}-\d+)\s+(\S+)", line) if m: - cves.append({"id": m.group(1), "package": m.group(2), "url": f"https://security-tracker.debian.org/tracker/{m.group(1)}"}) + cves.append({ + "id": m.group(1), + "package": m.group(2), + "vector": "?", + "severity": "?", + "url": f"https://security-tracker.debian.org/tracker/{m.group(1)}" + }) return True, cves, "" @@ -139,15 +288,28 @@ def scan_target(target: Target, progress_cb: Callable) -> ScanResult: if debsecan_ok: cve_ok, cve_list, cve_err = scan_cve(target) result.cve_ok = cve_ok - result.cve_count = len(cve_list) - result.cve_list = cve_list if not cve_ok: result.error = cve_err + + # Enrichir les CVE via l'API Debian (severity, vector, fixable) + if cve_list: + all_cves, actionable_count = enrich_cves(cve_list) + result.cve_list = all_cves + result.cve_count = actionable_count + result.cve_total = len(all_cves) + else: + result.cve_list = [] + result.cve_count = 0 + result.cve_total = 0 else: result.cve_ok = False result.error = f"debsecan: {debsecan_err}" progress_cb() + # Scan OS + result.os_info = scan_os(target) + progress_cb() + result.status = "done" if (result.apt_ok and result.cve_ok) else "error" if result.error: result.status = "error" @@ -157,6 +319,8 @@ def scan_target(target: Target, progress_cb: Callable) -> ScanResult: "apt_count": result.apt_count, "apt_packages": result.apt_packages, "cve_count": result.cve_count, + "cve_total": result.cve_total, + "os_info": result.os_info, "cve_list": result.cve_list, "error": result.error }) diff --git a/full_updater/ui/detail_screens.py b/full_updater/ui/detail_screens.py index 7299705..5e95e10 100644 --- a/full_updater/ui/detail_screens.py +++ b/full_updater/ui/detail_screens.py @@ -26,7 +26,20 @@ SCREEN_CSS = """ class PackageListScreen(Screen): BINDINGS = [("b", "back", "Retour")] - CSS = SCREEN_CSS + DEFAULT_CSS = """ + PackageListScreen { + align: left top; + padding: 1 2; + } + PackageListScreen #toolbar { + height: auto; + margin-bottom: 1; + } + PackageListScreen DataTable { + height: 1fr; + border: solid $primary; + } + """ def __init__(self, packages: list[dict]): super().__init__() @@ -52,7 +65,20 @@ class PackageListScreen(Screen): class CVEListScreen(Screen): BINDINGS = [("b", "back", "Retour")] - CSS = SCREEN_CSS + DEFAULT_CSS = """ + CVEListScreen { + align: left top; + padding: 1 2; + } + CVEListScreen #toolbar { + height: auto; + margin-bottom: 1; + } + CVEListScreen DataTable { + height: 1fr; + border: solid $primary; + } + """ def __init__(self, cves: list[dict]): super().__init__() @@ -64,14 +90,17 @@ class CVEListScreen(Screen): with Horizontal(id="toolbar"): yield Button("⬅ Retour", id="cve-back", variant="default") table = DataTable(id="cve-table") - table.add_columns("CVE-ID", "Paquet", "Lien") + table.add_columns("CVE-ID", "Paquet", "Severite", "Vecteur", "Corrigeable", "Lien") table.cursor_type = "row" for i, cve in enumerate(self.cves): cve_id = cve.get("id", "?") pkg = cve.get("package", "?") url = cve.get("url", "") + severity = cve.get("severity", "?") + vector = cve.get("vector", "?") + fixable = "🟢 Oui" if cve.get("fixable") else "🔴 Non" self.urls[i] = url - table.add_row(cve_id, pkg, url) + table.add_row(cve_id, pkg, severity, vector, fixable, url) yield table def on_data_table_row_selected(self, event: DataTable.RowSelected): diff --git a/full_updater/ui/log_panel.py b/full_updater/ui/log_panel.py index 0ccfb15..1022d1c 100644 --- a/full_updater/ui/log_panel.py +++ b/full_updater/ui/log_panel.py @@ -23,21 +23,21 @@ class LogPanel(Vertical): def __init__(self): super().__init__() - self.log = None + self._log_widget = None def compose(self): with Horizontal(id="log-toolbar"): yield Button("⬅ Retour", id="log-back") - self.log = RichLog(id="log-view", highlight=True) - yield self.log + self._log_widget = RichLog(id="log-view", highlight=True) + yield self._log_widget def write(self, line: str): - if self.log: - self.log.write(line) + if self._log_widget: + self._log_widget.write(line) def clear(self): - if self.log: - self.log.clear() + if self._log_widget: + self._log_widget.clear() def on_button_pressed(self, event: Button.Pressed): if event.button.id == "log-back": diff --git a/full_updater/ui/summary.py b/full_updater/ui/summary.py index b012b9f..4b9fe7d 100644 --- a/full_updater/ui/summary.py +++ b/full_updater/ui/summary.py @@ -1,6 +1,5 @@ from textual.widgets import Static, Button from textual.containers import Vertical, Horizontal -from textual.reactive import reactive from textual.message import Message @@ -26,10 +25,6 @@ class SummaryPanel(Vertical): height: auto; margin: 1 0; } - .summary-count { - color: $primary; - text-style: bold; - } .summary-error { color: $error; text-style: bold; @@ -55,13 +50,8 @@ class SummaryPanel(Vertical): self.target_id = target_id super().__init__() - cache_time = reactive("") - apt_count = reactive(0) - cve_count = reactive(0) - target_id = reactive("") - target_name = reactive("") - error_msg = reactive("") - is_skipped = reactive(False) + _current_target_id: str = "" + _current_target_name: str = "" def compose(self): with Horizontal(id="summary-header"): @@ -70,36 +60,29 @@ class SummaryPanel(Vertical): with Vertical(id="summary-body"): yield Static("Sélectionnez une cible", id="summary-title") + yield Static("", id="summary-os", classes="summary-row") yield Button("Mises à jour : -", id="btn-apt", classes="summary-row", variant="default") yield Button("CVE : -", id="btn-cve", classes="summary-row", variant="default") yield Static("", id="summary-error", classes="summary-row summary-error") yield Button("📦 Mettre à jour", id="btn-upgrade", variant="primary") - def watch_cache_time(self, value: str): - self.query_one("#summary-cache", Static).update(f"Cache : {value}" if value else "") + def set_target(self, target_id: str, name: str, apt_count: int, cve_count: int, cve_total: int, os_info: str, error: str, skipped: bool, cache_time: str): + self._current_target_id = target_id + self._current_target_name = name - def watch_apt_count(self, value: int): - self._update_display() - - def watch_cve_count(self, value: int): - self._update_display() - - def watch_error_msg(self, value: str): - self._update_display() - - def watch_is_skipped(self, value: bool): - self._update_display() - - def watch_target_name(self, value: str): - self.query_one("#summary-title", Static).update(value if value else "Sélectionnez une cible") - - def _update_display(self): + title = self.query_one("#summary-title", Static) + os_label = self.query_one("#summary-os", Static) apt_btn = self.query_one("#btn-apt", Button) cve_btn = self.query_one("#btn-cve", Button) err_label = self.query_one("#summary-error", Static) btn = self.query_one("#btn-upgrade", Button) + cache_label = self.query_one("#summary-cache", Static) - if self.is_skipped: + title.update(name if name else "Sélectionnez une cible") + os_label.update(os_info if os_info else "") + cache_label.update(f"Cache : {cache_time}" if cache_time else "") + + if skipped: apt_btn.label = "Mises à jour : LXC éteint" cve_btn.label = "CVE : LXC éteint" err_label.update("") @@ -108,37 +91,38 @@ class SummaryPanel(Vertical): cve_btn.disabled = True return - if self.error_msg: - apt_btn.label = f"Mises à jour : {self.apt_count}" + if error: + apt_btn.label = f"Mises à jour : {apt_count}" cve_btn.label = "CVE : ERREUR" - err_label.update(self.error_msg) + err_label.update(error) btn.disabled = False - apt_btn.disabled = self.apt_count == 0 + apt_btn.disabled = apt_count == 0 cve_btn.disabled = True return - apt_btn.label = f"Mises à jour : {self.apt_count}" - cve_btn.label = f"CVE : {self.cve_count}" + apt_btn.label = f"Mises à jour : {apt_count}" + if cve_total > cve_count: + cve_btn.label = f"CVE : {cve_count} ({cve_total - cve_count} non corrigeables)" + else: + cve_btn.label = f"CVE : {cve_count}" err_label.update("") btn.disabled = False - apt_btn.disabled = self.apt_count == 0 - cve_btn.disabled = self.cve_count == 0 + apt_btn.disabled = apt_count == 0 + cve_btn.disabled = cve_total == 0 - def set_target(self, target_id: str, name: str, apt_count: int, cve_count: int, error: str, skipped: bool, cache_time: str): - self.target_id = target_id - self.target_name = name - self.apt_count = apt_count - self.cve_count = cve_count - self.error_msg = error - self.is_skipped = skipped - self.cache_time = cache_time + # Forcer le refresh de tous les widgets modifiés + apt_btn.refresh() + cve_btn.refresh() + err_label.refresh() + btn.refresh() + cache_label.refresh() def on_button_pressed(self, event: Button.Pressed): if event.button.id == "btn-reload": self.post_message(self.ReloadPressed()) elif event.button.id == "btn-upgrade": - self.post_message(self.UpgradePressed(self.target_id, self.target_name)) + self.post_message(self.UpgradePressed(self._current_target_id, self._current_target_name)) elif event.button.id == "btn-apt": - self.post_message(self.AptClicked(self.target_id)) + self.post_message(self.AptClicked(self._current_target_id)) elif event.button.id == "btn-cve": - self.post_message(self.CveClicked(self.target_id)) + self.post_message(self.CveClicked(self._current_target_id))