Compare commits
13 commits
v2026.05.1
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 11ada690a1 | |||
| 9107009370 | |||
| 13d826ecd2 | |||
| 721e677fa6 | |||
| c1f462d209 | |||
| f0f59f0468 | |||
| d90f74dd8f | |||
| 2d339cc397 | |||
| 6e707da98d | |||
| 3bb213a00d | |||
| 9c4e40505b | |||
| 86eda73eb9 | |||
| e22f416500 |
4 changed files with 204 additions and 11 deletions
|
|
@ -128,6 +128,8 @@ class FullUpdaterApp(App):
|
||||||
name=name,
|
name=name,
|
||||||
apt_count=data.get("apt_count", 0),
|
apt_count=data.get("apt_count", 0),
|
||||||
cve_count=data.get("cve_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", ""),
|
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),
|
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)
|
cache_time=get_cache_timestamp(cache_id)
|
||||||
|
|
@ -158,7 +160,8 @@ class FullUpdaterApp(App):
|
||||||
write_cache(target.target_id, {
|
write_cache(target.target_id, {
|
||||||
"timestamp": datetime.now().isoformat(),
|
"timestamp": datetime.now().isoformat(),
|
||||||
"apt_count": 0, "apt_packages": [],
|
"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"
|
"error": "LXC éteint"
|
||||||
})
|
})
|
||||||
self._update_sidebar(target.target_id, result)
|
self._update_sidebar(target.target_id, result)
|
||||||
|
|
@ -176,6 +179,8 @@ class FullUpdaterApp(App):
|
||||||
"apt_count": result.apt_count,
|
"apt_count": result.apt_count,
|
||||||
"apt_packages": result.apt_packages,
|
"apt_packages": result.apt_packages,
|
||||||
"cve_count": result.cve_count,
|
"cve_count": result.cve_count,
|
||||||
|
"cve_total": result.cve_total,
|
||||||
|
"os_info": result.os_info,
|
||||||
"cve_list": result.cve_list,
|
"cve_list": result.cve_list,
|
||||||
"error": result.error
|
"error": result.error
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -1,10 +1,115 @@
|
||||||
|
import json
|
||||||
|
import os
|
||||||
import re
|
import re
|
||||||
import subprocess
|
import subprocess
|
||||||
|
import urllib.request
|
||||||
|
from concurrent.futures import ThreadPoolExecutor
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from typing import Callable
|
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: <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
|
@dataclass
|
||||||
|
|
@ -21,6 +126,8 @@ class ScanResult:
|
||||||
cve_ok: bool = False
|
cve_ok: bool = False
|
||||||
apt_count: int = 0
|
apt_count: int = 0
|
||||||
cve_count: int = 0
|
cve_count: int = 0
|
||||||
|
cve_total: int = 0
|
||||||
|
os_info: str = ""
|
||||||
apt_packages: list[dict[str, str]] = field(default_factory=list)
|
apt_packages: list[dict[str, str]] = field(default_factory=list)
|
||||||
cve_list: list[dict[str, str]] = field(default_factory=list)
|
cve_list: list[dict[str, str]] = field(default_factory=list)
|
||||||
error: str = ""
|
error: str = ""
|
||||||
|
|
@ -56,6 +163,41 @@ def lxc_is_running(vmid: str) -> bool:
|
||||||
return ok and "running" in stdout.lower()
|
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]:
|
def ensure_debsecan_installed(is_host: bool, vmid: str = "") -> tuple[bool, str]:
|
||||||
if is_host:
|
if is_host:
|
||||||
ok, _, _ = run_cmd(["which", "debsecan"])
|
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]:
|
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)
|
ok, stdout, stderr = run_cmd(cmd, timeout=120)
|
||||||
if not ok:
|
if not ok:
|
||||||
return False, [], stderr or stdout
|
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():
|
for line in stdout.splitlines():
|
||||||
m = re.match(r"(CVE-\d{4}-\d+)\s+(\S+)", line)
|
m = re.match(r"(CVE-\d{4}-\d+)\s+(\S+)", line)
|
||||||
if m:
|
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, ""
|
return True, cves, ""
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -139,15 +288,28 @@ def scan_target(target: Target, progress_cb: Callable) -> ScanResult:
|
||||||
if debsecan_ok:
|
if debsecan_ok:
|
||||||
cve_ok, cve_list, cve_err = scan_cve(target)
|
cve_ok, cve_list, cve_err = scan_cve(target)
|
||||||
result.cve_ok = cve_ok
|
result.cve_ok = cve_ok
|
||||||
result.cve_count = len(cve_list)
|
|
||||||
result.cve_list = cve_list
|
|
||||||
if not cve_ok:
|
if not cve_ok:
|
||||||
result.error = cve_err
|
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:
|
else:
|
||||||
result.cve_ok = False
|
result.cve_ok = False
|
||||||
result.error = f"debsecan: {debsecan_err}"
|
result.error = f"debsecan: {debsecan_err}"
|
||||||
progress_cb()
|
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"
|
result.status = "done" if (result.apt_ok and result.cve_ok) else "error"
|
||||||
if result.error:
|
if result.error:
|
||||||
result.status = "error"
|
result.status = "error"
|
||||||
|
|
@ -157,6 +319,8 @@ def scan_target(target: Target, progress_cb: Callable) -> ScanResult:
|
||||||
"apt_count": result.apt_count,
|
"apt_count": result.apt_count,
|
||||||
"apt_packages": result.apt_packages,
|
"apt_packages": result.apt_packages,
|
||||||
"cve_count": result.cve_count,
|
"cve_count": result.cve_count,
|
||||||
|
"cve_total": result.cve_total,
|
||||||
|
"os_info": result.os_info,
|
||||||
"cve_list": result.cve_list,
|
"cve_list": result.cve_list,
|
||||||
"error": result.error
|
"error": result.error
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -9,6 +9,21 @@ except Exception:
|
||||||
PYPERCLIP_OK = False
|
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):
|
class PackageListScreen(Screen):
|
||||||
BINDINGS = [("b", "back", "Retour")]
|
BINDINGS = [("b", "back", "Retour")]
|
||||||
DEFAULT_CSS = """
|
DEFAULT_CSS = """
|
||||||
|
|
@ -75,14 +90,17 @@ class CVEListScreen(Screen):
|
||||||
with Horizontal(id="toolbar"):
|
with Horizontal(id="toolbar"):
|
||||||
yield Button("⬅ Retour", id="cve-back", variant="default")
|
yield Button("⬅ Retour", id="cve-back", variant="default")
|
||||||
table = DataTable(id="cve-table")
|
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"
|
table.cursor_type = "row"
|
||||||
for i, cve in enumerate(self.cves):
|
for i, cve in enumerate(self.cves):
|
||||||
cve_id = cve.get("id", "?")
|
cve_id = cve.get("id", "?")
|
||||||
pkg = cve.get("package", "?")
|
pkg = cve.get("package", "?")
|
||||||
url = cve.get("url", "")
|
url = cve.get("url", "")
|
||||||
|
severity = cve.get("severity", "?")
|
||||||
|
vector = cve.get("vector", "?")
|
||||||
|
fixable = "🟢 Oui" if cve.get("fixable") else "🔴 Non"
|
||||||
self.urls[i] = url
|
self.urls[i] = url
|
||||||
table.add_row(cve_id, pkg, url)
|
table.add_row(cve_id, pkg, severity, vector, fixable, url)
|
||||||
yield table
|
yield table
|
||||||
|
|
||||||
def on_data_table_row_selected(self, event: DataTable.RowSelected):
|
def on_data_table_row_selected(self, event: DataTable.RowSelected):
|
||||||
|
|
|
||||||
|
|
@ -60,16 +60,18 @@ class SummaryPanel(Vertical):
|
||||||
|
|
||||||
with Vertical(id="summary-body"):
|
with Vertical(id="summary-body"):
|
||||||
yield Static("Sélectionnez une cible", id="summary-title")
|
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("Mises à jour : -", id="btn-apt", classes="summary-row", variant="default")
|
||||||
yield Button("CVE : -", id="btn-cve", 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 Static("", id="summary-error", classes="summary-row summary-error")
|
||||||
yield Button("📦 Mettre à jour", id="btn-upgrade", variant="primary")
|
yield Button("📦 Mettre à jour", id="btn-upgrade", variant="primary")
|
||||||
|
|
||||||
def set_target(self, target_id: str, name: str, apt_count: int, cve_count: int, error: str, skipped: bool, cache_time: str):
|
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_id = target_id
|
||||||
self._current_target_name = name
|
self._current_target_name = name
|
||||||
|
|
||||||
title = self.query_one("#summary-title", Static)
|
title = self.query_one("#summary-title", Static)
|
||||||
|
os_label = self.query_one("#summary-os", Static)
|
||||||
apt_btn = self.query_one("#btn-apt", Button)
|
apt_btn = self.query_one("#btn-apt", Button)
|
||||||
cve_btn = self.query_one("#btn-cve", Button)
|
cve_btn = self.query_one("#btn-cve", Button)
|
||||||
err_label = self.query_one("#summary-error", Static)
|
err_label = self.query_one("#summary-error", Static)
|
||||||
|
|
@ -77,6 +79,7 @@ class SummaryPanel(Vertical):
|
||||||
cache_label = self.query_one("#summary-cache", Static)
|
cache_label = self.query_one("#summary-cache", Static)
|
||||||
|
|
||||||
title.update(name if name else "Sélectionnez une cible")
|
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 "")
|
cache_label.update(f"Cache : {cache_time}" if cache_time else "")
|
||||||
|
|
||||||
if skipped:
|
if skipped:
|
||||||
|
|
@ -98,11 +101,14 @@ class SummaryPanel(Vertical):
|
||||||
return
|
return
|
||||||
|
|
||||||
apt_btn.label = f"Mises à jour : {apt_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}"
|
cve_btn.label = f"CVE : {cve_count}"
|
||||||
err_label.update("")
|
err_label.update("")
|
||||||
btn.disabled = False
|
btn.disabled = False
|
||||||
apt_btn.disabled = apt_count == 0
|
apt_btn.disabled = apt_count == 0
|
||||||
cve_btn.disabled = cve_count == 0
|
cve_btn.disabled = cve_total == 0
|
||||||
|
|
||||||
# Forcer le refresh de tous les widgets modifiés
|
# Forcer le refresh de tous les widgets modifiés
|
||||||
apt_btn.refresh()
|
apt_btn.refresh()
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue