Compare commits

..

20 commits

Author SHA1 Message Date
11ada690a1 fix(cve): enrich CVEs with severity/vector from Debian Security Tracker HTML
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- debsecan default format has no [remote|local] or [severity] flags
- Now fetches severity and vector from security-tracker.debian.org HTML
- scan_cve uses default format (no --format report)
- enrich_cves replaces filter_actionable_cves, adds severity+vector
2026-05-13 04:33:05 +02:00
9107009370 fix(cve): robust parsing of [remote|local] and [severity] flags
All checks were successful
Build and Release .deb / build-deb (push) Successful in 22s
- Use re.findall to extract all bracketed flags instead of positional regex
- Fixes issue where optional groups were not captured correctly
2026-05-13 04:23:45 +02:00
13d826ecd2 feat(cve): add severity and vector columns to CVE list
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Parse [remote|local] and [severity] from debsecan output
- Display Severity and Vector columns in CVEListScreen
- Severity values: unimportant, low, medium, high, critical
2026-05-13 04:20:14 +02:00
721e677fa6 fix(os): use NAME directly, avoid double 'GNU/Linux'
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- NAME already contains 'Debian GNU/Linux', don't append it again
- With DEBIAN_VERSION_FULL: NAME + DEBIAN_VERSION_FULL + (codename)
- Without: NAME + VERSION_ID + (codename)
- Fixes duplication bug
2026-05-13 04:15:57 +02:00
c1f462d209 fix(os): parse /etc/os-release line-by-line instead of regex
All checks were successful
Build and Release .deb / build-deb (push) Successful in 22s
- pct exec returns stdout differently than local cat
- Line-by-line parsing is more robust across LXC boundaries
- Handles quotes correctly
2026-05-13 04:10:01 +02:00
f0f59f0468 fix(os): read DEBIAN_VERSION_FULL from /etc/os-release first
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Check DEBIAN_VERSION_FULL (ex: 13.4) as primary source
- Fallback to VERSION, then VERSION_ID, then /etc/debian_version
2026-05-13 04:05:45 +02:00
d90f74dd8f fix(os): use VERSION_ID from /etc/os-release for full version number in LXC
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
2026-05-13 03:59:47 +02:00
2d339cc397 feat(os): show full version number from /etc/debian_version
All checks were successful
Build and Release .deb / build-deb (push) Successful in 22s
- Use lsb_release -is/-cs for name and codename
- Use /etc/debian_version for full version (e.g. 12.9)
- Display format: 'Debian GNU/Linux 12.9 (bookworm)'
2026-05-13 03:52:10 +02:00
6e707da98d feat(ui): display OS name and version in SummaryPanel
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Add scan_os() to detect OS via lsb_release or /etc/os-release
- Store os_info in cache and ScanResult
- Display OS info in SummaryPanel below the target name
2026-05-13 03:07:28 +02:00
3bb213a00d fix(cve): parse HTML instead of non-existent JSON API
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Debian Security Tracker has no public JSON API for individual CVEs
- Now fetches and parses the HTML page directly
- Searches for 'bookworm ... fixed' pattern in the vulnerability table
- Cache files changed from .json to .html
2026-05-13 02:56:16 +02:00
9c4e40505b fix(summary): enable CVE button when non-fixable CVEs exist
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Use cve_total instead of cve_count to determine button disabled state
- Allows viewing all CVEs even when none are fixable
2026-05-13 02:38:58 +02:00
86eda73eb9 feat(cve): show all CVEs with fixable indicator in list screen
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- filter_actionable_cves now marks all CVEs with 'fixable' boolean
- cve_list in cache contains ALL CVEs (not just actionable ones)
- CVEListScreen adds 'Corrigeable' column with 🟢/🔴 indicator
- Sidebar counter still shows only actionable CVEs (cve_count)
2026-05-13 02:35:15 +02:00
e22f416500 feat(cve): filter actionable CVEs via Debian Security Tracker API
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Add filter_actionable_cves() that queries security-tracker.debian.org
- Cache API responses in /tmp/full-updater-cache/cve-api/
- Use ThreadPoolExecutor(max_workers=10) for parallel API calls
- cve_count now shows only actionable CVEs (with fixed_version)
- cve_total stored for info, shown as 'CVE: X (Y non corrigeables)'
2026-05-13 02:27:22 +02:00
af9e061ab5 fix(upgrade): remove call_from_thread in async worker, call log_panel.write directly
All checks were successful
Build and Release .deb / build-deb (push) Successful in 22s
2026-05-13 02:11:08 +02:00
eed84f36e8 fix(log_panel): rename self.log to self._log_widget to avoid Textual property conflict
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
2026-05-13 02:07:25 +02:00
f4db16327f fix(cache): purge cache only once at startup, not on every write
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Add clear_cache() called once in on_mount
- ensure_cache_dir() no longer deletes existing files
- This fixes the issue where only the last target's cache survived
- Add widget.refresh() calls to force UI updates
2026-05-13 02:00:48 +02:00
8f623e1df6 fix(ui): remove reactive vars in SummaryPanel, use direct widget update
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Replace reactive attributes with direct widget updates in set_target
- Delay _select_target by 0.1s after pop_screen to ensure DOM is stable
- Remove summary.refresh() calls that interfere with direct updates
2026-05-13 01:54:57 +02:00
07e84ae69f fix(ui): use DEFAULT_CSS with selectors in detail screens
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Replace inline CSS string with DEFAULT_CSS class attribute
- Add proper Textual CSS selectors (ScreenName, #toolbar, DataTable)
- Remove invalid align at root level
2026-05-13 01:48:00 +02:00
1715a6cf01 fix(ui): add proper CSS layout for detail screens
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Add dock: top and fixed height for toolbar buttons
- DataTable now fills remaining space with height: 1fr
- Add borders and padding for better visual separation
2026-05-13 01:44:28 +02:00
76eb75d4a6 refactor: sequential scan, screens for details, new sidebar layout
All checks were successful
Build and Release .deb / build-deb (push) Successful in 21s
- Scan sequential with asyncio.to_thread for proper Textual integration
- Use push_screen/pop_screen for package/CVE lists (no more mount/remove)
- Sidebar now shows 'APT X CVE X' with right-aligned indicators
- Loader uses black circles () for pending tasks
- Removed unused package_table.py and cve_table.py
2026-05-13 01:36:45 +02:00
10 changed files with 432 additions and 397 deletions

View file

@ -4,21 +4,18 @@ 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.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,
run_full_scan_async
ensure_debsecan_installed, scan_apt, scan_cve, write_cache, scan_target
)
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.detail_screens import PackageListScreen, CVEListScreen
from full_updater.ui.log_panel import LogPanel
from full_updater.ui.confirm_modal import ConfirmModal
@ -27,7 +24,7 @@ class FullUpdaterApp(App):
CSS = """
Screen { align: center middle; }
#main-layout { layout: horizontal; height: 100%; }
Sidebar { width: 30; }
Sidebar { width: 42; }
#content { width: 1fr; }
"""
@ -49,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)
@ -62,25 +60,19 @@ class FullUpdaterApp(App):
total = len(self.targets)
completed = 0
async def progress_cb():
def progress_cb():
nonlocal completed
completed += 1
pct = min(100.0, (completed / (total * 4)) * 100)
self._update_loader(pct)
def on_result(result: ScanResult):
self.results[result.target.target_id] = result
idx = 0
for i, t in enumerate(self.targets):
if t.target_id == result.target.target_id:
idx = i
break
for idx, target in enumerate(self.targets):
self._update_loader_status(idx, "running")
result = await asyncio.to_thread(scan_target, target, progress_cb)
self.results[target.target_id] = result
self._update_loader_status(idx, result.status)
self._update_sidebar(result.target.target_id, result)
self._update_sidebar(target.target_id, result)
results = await run_full_scan_async(progress_cb)
for r in results:
on_result(r)
self._finish_scan()
def _update_loader(self, pct: float):
@ -102,6 +94,7 @@ class FullUpdaterApp(App):
result.error,
result.status == "skipped"
)
sidebar.refresh()
def _finish_scan(self):
self.pop_screen()
@ -116,12 +109,9 @@ class FullUpdaterApp(App):
res.error,
res.status == "skipped"
)
# Forcer le rafraichissement de l'interface
sidebar.refresh()
if self.targets:
self._select_target(self.targets[0].target_id)
summary = self.query_one(SummaryPanel)
summary.refresh()
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
@ -138,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)
@ -151,8 +143,8 @@ class FullUpdaterApp(App):
if self.selected_target:
self._reload_target(self.selected_target)
@work(thread=True)
def _reload_target(self, target_id: str):
@work
async def _reload_target(self, target_id: str):
target = None
for t in self.targets:
if t.target_id == target_id:
@ -167,32 +159,19 @@ class FullUpdaterApp(App):
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": [],
"apt_count": 0, "apt_packages": [],
"cve_count": 0, "cve_total": 0, "cve_list": [],
"os_info": "LXC éteint",
"error": "LXC éteint"
})
self.call_from_thread(self._update_sidebar, target.target_id, result)
self.call_from_thread(self._select_target, target.target_id)
self._update_sidebar(target.target_id, result)
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
def dummy_progress():
pass
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"
)
result = await asyncio.to_thread(scan_target, target, dummy_progress)
self.results[target.target_id] = result
cache_id = "host" if target.is_host else target.target_id
write_cache(cache_id, {
@ -200,11 +179,13 @@ 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
})
self.call_from_thread(self._update_sidebar, target.target_id, result)
self.call_from_thread(self._select_target, target.target_id)
self._update_sidebar(target.target_id, result)
self._select_target(target.target_id)
def on_summary_panel_upgrade_pressed(self, event: SummaryPanel.UpgradePressed):
self.push_screen(
@ -232,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:
@ -253,11 +234,7 @@ class FullUpdaterApp(App):
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)
table.focus()
self.push_screen(PackageListScreen(pkgs))
def on_summary_panel_cve_clicked(self, event: SummaryPanel.CveClicked):
cache_id = "host" if event.target_id == "host" else event.target_id
@ -265,18 +242,4 @@ class FullUpdaterApp(App):
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)
table.focus()
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()
self.push_screen(CVEListScreen(cves))

View file

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

View file

@ -1,17 +1,121 @@
import asyncio
import json
import os
import re
import subprocess
import urllib.request
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import Any
from full_updater.backend.cache import write_cache
from datetime import datetime
from typing import Callable
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: <td>remote</td> ou <td>local</td>
vector_match = re.search(r'<td[^>]*>\s*(remote|local)\s*</td>', 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"<td[^>]*>\s*{re.escape(suite)}\s*(?:\(security\))?\s*</td>\s*<td[^>]*>[^<]+</td>\s*<td[^>]*>\s*fixed\s*</td>", 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
class Target:
target_id: str # 'host' ou '100'
name: str # 'hote' ou '100 traefik'
target_id: str
name: str
is_host: bool
@ -22,17 +126,17 @@ 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 = ""
status: str = "pending" # pending | running | done | error | skipped
status: str = "pending"
def run_cmd(cmd: list[str], timeout: int = 120) -> tuple[bool, str, str]:
try:
proc = subprocess.run(
cmd, capture_output=True, text=True, timeout=timeout, check=False
)
proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False)
return proc.returncode == 0, proc.stdout, proc.stderr
except subprocess.TimeoutExpired:
return False, "", "Command timed out"
@ -41,7 +145,7 @@ def run_cmd(cmd: list[str], timeout: int = 120) -> tuple[bool, str, str]:
def get_lxc_list() -> list[Target]:
ok, stdout, stderr = run_cmd(["pct", "list"])
ok, stdout, _ = run_cmd(["pct", "list"])
if not ok:
return []
targets = []
@ -56,43 +160,67 @@ def get_lxc_list() -> list[Target]:
def lxc_is_running(vmid: str) -> bool:
ok, stdout, _ = run_cmd(["pct", "status", vmid])
if ok and "running" in stdout.lower():
return True
return False
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]:
"""Retourne (ok, error_msg)"""
if is_host:
ok, _, _ = run_cmd(["which", "debsecan"])
if ok:
return True, ""
ok, out, err = run_cmd(["apt-get", "update"], timeout=60)
ok, _, err = run_cmd(["apt-get", "update"], timeout=60)
if not ok:
return False, f"apt-get update failed: {err}"
ok, out, err = run_cmd(["apt-get", "install", "-y", "debsecan"], timeout=120)
if not ok:
return False, f"apt-get install debsecan failed: {err}"
return True, ""
ok, _, err = run_cmd(["apt-get", "install", "-y", "debsecan"], timeout=120)
return (True, "") if ok else (False, f"apt-get install debsecan failed: {err}")
else:
ok, _, _ = run_cmd(["pct", "exec", vmid, "--", "which", "debsecan"])
if ok:
return True, ""
ok, out, err = run_cmd(["pct", "exec", vmid, "--", "apt-get", "update"], timeout=60)
ok, _, err = run_cmd(["pct", "exec", vmid, "--", "apt-get", "update"], timeout=60)
if not ok:
return False, f"apt-get update failed in {vmid}: {err}"
ok, out, err = run_cmd(["pct", "exec", vmid, "--", "apt-get", "install", "-y", "debsecan"], timeout=120)
if not ok:
return False, f"apt-get install debsecan failed in {vmid}: {err}"
return True, ""
ok, _, err = run_cmd(["pct", "exec", vmid, "--", "apt-get", "install", "-y", "debsecan"], timeout=120)
return (True, "") if ok else (False, f"apt-get install debsecan failed in {vmid}: {err}")
def scan_apt(target: Target) -> tuple[bool, list[dict[str, str]], str]:
if target.is_host:
cmd = ["apt", "list", "--upgradable"]
else:
cmd = ["pct", "exec", target.target_id, "--", "apt", "list", "--upgradable"]
cmd = ["apt", "list", "--upgradable"] if target.is_host else ["pct", "exec", target.target_id, "--", "apt", "list", "--upgradable"]
ok, stdout, stderr = run_cmd(cmd, timeout=120)
if not ok:
return False, [], stderr or stdout
@ -105,35 +233,21 @@ def scan_apt(target: Target) -> tuple[bool, list[dict[str, str]], str]:
if len(parts) < 2:
continue
name = parts[0].strip()
# Essayer d'extraire la nouvelle version et l'ancienne
m = re.search(r"\[upgradable from:\s+([^\]]+)\]", line)
if m:
old_ver = m.group(1).strip()
prefix = line.split("[upgradable from:")[0].strip()
ver_m = re.search(r"/\S+\s+(\S+)\s+\S+", prefix)
new_ver = ver_m.group(1) if ver_m else "?"
packages.append({
"name": name,
"current": old_ver,
"new": new_ver,
"size": "-"
})
packages.append({"name": name, "current": old_ver, "new": new_ver, "size": "-"})
else:
packages.append({
"name": name,
"current": "?",
"new": "?",
"size": "-"
})
packages.append({"name": name, "current": "?", "new": "?", "size": "-"})
return True, packages, ""
def scan_cve(target: Target) -> tuple[bool, list[dict[str, str]], str]:
if target.is_host:
cmd = ["debsecan", "--format", "report", "--suite", "bookworm"]
else:
cmd = ["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
@ -142,24 +256,24 @@ 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:
cve_id = m.group(1)
pkg = m.group(2)
cves.append({
"id": cve_id,
"package": pkg,
"url": f"https://security-tracker.debian.org/tracker/{cve_id}"
"id": m.group(1),
"package": m.group(2),
"vector": "?",
"severity": "?",
"url": f"https://security-tracker.debian.org/tracker/{m.group(1)}"
})
return True, cves, ""
def scan_single_target(target: Target, progress_cb: Any) -> ScanResult:
"""Scanne une cible (APT + CVE) et retourne le résultat."""
def scan_target(target: Target, progress_cb: Callable) -> ScanResult:
result = ScanResult(target=target)
result.status = "running"
progress_cb()
if not target.is_host and not lxc_is_running(target.target_id):
result.status = "skipped"
progress_cb(); progress_cb(); progress_cb(); progress_cb()
return result
debsecan_ok, debsecan_err = ensure_debsecan_installed(target.is_host, target.target_id)
@ -174,92 +288,41 @@ def scan_single_target(target: Target, progress_cb: Any) -> 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"
cache_id = "host" if target.is_host else target.target_id
write_cache(cache_id, {
write_cache("host" if target.is_host else target.target_id, {
"timestamp": datetime.now().isoformat(),
"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
})
progress_cb()
return result
async def scan_single_target_async(target: Target, progress_cb: Any) -> ScanResult:
"""Version async de scan_single_target (pour asyncio.gather)."""
result = ScanResult(target=target)
result.status = "running"
await progress_cb()
if not target.is_host and not lxc_is_running(target.target_id):
result.status = "skipped"
await progress_cb()
await progress_cb()
await progress_cb()
await progress_cb()
return result
debsecan_ok, debsecan_err = await asyncio.to_thread(ensure_debsecan_installed, target.is_host, target.target_id)
await progress_cb()
apt_ok, apt_packages, apt_err = await asyncio.to_thread(scan_apt, target)
result.apt_ok = apt_ok
result.apt_count = len(apt_packages)
result.apt_packages = apt_packages
await progress_cb()
if debsecan_ok:
cve_ok, cve_list, cve_err = await asyncio.to_thread(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
else:
result.cve_ok = False
result.error = f"debsecan: {debsecan_err}"
await progress_cb()
result.status = "done" if (result.apt_ok and result.cve_ok) else "error"
if result.error:
result.status = "error"
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
})
await progress_cb()
return result
async def run_full_scan_async(progress_cb: Any) -> list[ScanResult]:
targets = [Target(target_id="host", name="hote", is_host=True)] + get_lxc_list()
coros = [scan_single_target_async(t, progress_cb) for t in targets]
results = await asyncio.gather(*coros, return_exceptions=True)
out = []
for r in results:
if isinstance(r, Exception):
out.append(ScanResult(target=Target(target_id="unknown", name="unknown", is_host=False), status="error", error=str(r)))
else:
out.append(r)
return out

View file

@ -1,69 +0,0 @@
from textual.widgets import DataTable, Button
from textual.containers import Vertical, Horizontal
from textual.message import Message
try:
import pyperclip
PYPERCLIP_OK = True
except Exception:
PYPERCLIP_OK = False
class CVETable(Vertical):
DEFAULT_CSS = """
CVETable {
padding: 1 2;
}
#cve-toolbar {
height: auto;
margin-bottom: 1;
}
DataTable {
height: 1fr;
}
"""
class BackPressed(Message):
pass
def __init__(self):
super().__init__()
self.table = DataTable(id="cve-table")
self.urls = {}
def compose(self):
with Horizontal(id="cve-toolbar"):
yield Button("⬅ Retour", id="cve-back")
self.table.add_columns("CVE-ID", "Paquet", "Lien")
self.table.cursor_type = "row"
yield self.table
def load_data(self, cves: list[dict]):
self.table.clear()
self.urls = {}
for i, cve in enumerate(cves):
cve_id = cve.get("id", "?")
pkg = cve.get("package", "?")
url = cve.get("url", "")
self.urls[i] = url
self.table.add_row(cve_id, pkg, url)
def on_data_table_row_selected(self, event: DataTable.RowSelected):
row_key = event.row_key
row_index = list(self.table.rows.keys()).index(row_key)
url = self.urls.get(row_index, "")
if not url:
return
if PYPERCLIP_OK:
try:
pyperclip.copy(url)
self.notify(f"Lien copié : {url}", severity="information")
return
except Exception:
pass
# Fallback : afficher le lien dans une notification
self.notify(f"Lien (copie manuelle) : {url}", severity="warning")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "cve-back":
self.post_message(self.BackPressed())

View file

@ -0,0 +1,126 @@
from textual.screen import Screen
from textual.widgets import DataTable, Button
from textual.containers import Vertical, Horizontal
try:
import pyperclip
PYPERCLIP_OK = True
except Exception:
PYPERCLIP_OK = False
SCREEN_CSS = """
align: left top;
padding: 1 2;
#toolbar {
height: auto;
dock: top;
margin-bottom: 1;
}
DataTable {
height: 1fr;
border: solid $primary;
}
"""
class PackageListScreen(Screen):
BINDINGS = [("b", "back", "Retour")]
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__()
self.packages = packages
def compose(self):
with Vertical():
with Horizontal(id="toolbar"):
yield Button("⬅ Retour", id="pkg-back", variant="default")
table = DataTable(id="pkg-table")
table.add_columns("Nom", "Version actuelle", "Nouvelle version", "Taille")
for pkg in self.packages:
table.add_row(pkg.get("name", "?"), pkg.get("current", "?"), pkg.get("new", "?"), pkg.get("size", "-"))
yield table
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "pkg-back":
self.app.pop_screen()
def action_back(self):
self.app.pop_screen()
class CVEListScreen(Screen):
BINDINGS = [("b", "back", "Retour")]
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__()
self.cves = cves
self.urls = {}
def compose(self):
with Vertical():
with Horizontal(id="toolbar"):
yield Button("⬅ Retour", id="cve-back", variant="default")
table = DataTable(id="cve-table")
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, severity, vector, fixable, url)
yield table
def on_data_table_row_selected(self, event: DataTable.RowSelected):
row_key = event.row_key
row_index = list(self.query_one("#cve-table").rows.keys()).index(row_key)
url = self.urls.get(row_index, "")
if not url:
return
if PYPERCLIP_OK:
try:
pyperclip.copy(url)
self.notify(f"Lien copié : {url}", severity="information")
return
except Exception:
pass
self.notify(f"Lien (copie manuelle) : {url}", severity="warning")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "cve-back":
self.app.pop_screen()
def action_back(self):
self.app.pop_screen()

View file

@ -16,7 +16,7 @@ class LoaderScreen(Screen):
self.status_lines = [self._fmt_line(t) for t in targets]
def _fmt_line(self, target) -> str:
return f"{target.name}: 🟠 🟠"
return f"{target.name:20} ⚫ ⚫"
def compose(self):
with Container(id="loader-container"):
@ -29,10 +29,8 @@ class LoaderScreen(Screen):
def watch_progress(self, value: float):
try:
bar = self.query_one("#loader-bar", ProgressBar)
bar.progress = value
pct = self.query_one("#loader-percent", Static)
pct.update(f"{value:.0f} %")
self.query_one("#loader-bar", ProgressBar).progress = value
self.query_one("#loader-percent", Static).update(f"{value:.0f} %")
except Exception:
pass
@ -51,11 +49,11 @@ class LoaderScreen(Screen):
lines = list(self.status_lines)
t = self._targets[idx]
if status == "done":
lines[idx] = f"{t.name}: 🟢 🟢"
lines[idx] = f"{t.name:20} 🟢 🟢"
elif status == "error":
lines[idx] = f"{t.name}: 🔴 🔴"
lines[idx] = f"{t.name:20} 🔴 🔴"
elif status == "skipped":
lines[idx] = f"{t.name}: ⚪ ⚪"
lines[idx] = f"{t.name:20} ⚪ ⚪"
else:
lines[idx] = f"{t.name}: 🟠 🟠"
lines[idx] = f"{t.name:20} 🟠 🟠"
self.status_lines = lines

View file

@ -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":

View file

@ -1,45 +0,0 @@
from textual.widgets import DataTable, Button
from textual.containers import Vertical, Horizontal
from textual.message import Message
class PackageTable(Vertical):
DEFAULT_CSS = """
PackageTable {
padding: 1 2;
}
#pkg-toolbar {
height: auto;
margin-bottom: 1;
}
DataTable {
height: 1fr;
}
"""
class BackPressed(Message):
pass
def __init__(self):
super().__init__()
self.table = DataTable(id="pkg-table")
def compose(self):
with Horizontal(id="pkg-toolbar"):
yield Button("⬅ Retour", id="pkg-back")
self.table.add_columns("Nom", "Version actuelle", "Nouvelle version", "Taille")
yield self.table
def load_data(self, packages: list[dict]):
self.table.clear()
for pkg in packages:
self.table.add_row(
pkg.get("name", "?"),
pkg.get("current", "?"),
pkg.get("new", "?"),
pkg.get("size", "-")
)
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "pkg-back":
self.post_message(self.BackPressed())

View file

@ -6,7 +6,7 @@ from textual.message import Message
class Sidebar(Vertical):
DEFAULT_CSS = """
Sidebar {
width: 28;
width: 40;
border-right: solid $primary-lighten-2;
padding: 0 1;
}
@ -42,7 +42,8 @@ class Sidebar(Vertical):
lv.clear()
self._items = []
for t in targets:
label = f"{t.name} 🟠 🟠"
# Format: "100 traefik APT ⚫ CVE ⚫"
label = f"{t.name:20} APT ⚫ CVE ⚫"
item = ListItem(Label(label), id=f"target-{t.target_id}")
lv.append(item)
self._items.append((t.target_id, t.name, item))
@ -51,13 +52,24 @@ class Sidebar(Vertical):
for tid, name, item in self._items:
if tid == target_id:
if skipped:
label = f"{name} ⚪ ⚪"
apt_color = ""
cve_color = ""
elif error:
label = f"{name} 🔴 🔴"
elif apt_count > 0 or cve_count > 0:
label = f"{name} {'🔴' if apt_count > 0 else '🟢'} {'🔴' if cve_count > 0 else '🟢'}"
apt_color = "🔴"
cve_color = "🔴"
elif apt_count > 0:
apt_color = "🔴"
else:
label = f"{name} 🟢 🟢"
apt_color = "🟢"
if error and cve_count == 0:
cve_color = "🔴"
elif cve_count > 0:
cve_color = "🔴"
else:
cve_color = "🟢"
label = f"{name:20} APT {apt_color} CVE {cve_color}"
item.children[0].update(label)
break

View file

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